📁 Rust Backend with Axum · Part 40 of 42

Rust Web Server with Axum 0.8: How Future and Pin Really Work

November 3, 2026

Episode 40 of the Rust Backend series, where we build the backend of a blog step by step. After episode 39 we can write our own derive. We have typed async and .await hundreds of times. But what does .await actually do? Today we open the box: Future, poll, the Waker, and Pin. All the code below is the final, tested code from the video, ready to copy.

New to Rust? Start with the beginner series.
The full backend series playlist is here.

We continue in the same blog_api project from episode 39. The first part runs in examples/future_demo.rs (cargo run --example future_demo); the second part adds src/timed.rs to the blog. One new crate: pin-project-lite.

1. The Future trait

A Future is a trait with one method, poll. It answers either:

Poll::Ready(value): done, here is the result, or
Poll::Pending: not yet, call me again later.

rust

pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

2. Our own future: Countdown

Each poll counts down by one. At zero it is ready:

rust

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

struct Countdown(u32);

impl Future for Countdown {
    type Output = &'static str;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        println!("poll: {}", self.0);
        if self.0 == 0 {
            Poll::Ready("ready")
        } else {
            self.0 -= 1;
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

#[tokio::main]
async fn main() {
    println!("{}", Countdown(3).await);
}

When we return Pending, we must make sure someone polls us again. The Context holds a Waker; wake_by_ref() tells the runtime: poll this task again. (A real future, like a socket read, would give the waker to the operating system's event loop, which wakes it when data arrives.)

bash

cargo run --example future_demo

text

poll: 3
poll: 2
poll: 1
poll: 0
ready

Four polls for a countdown of three.

3. Forget the waker

Let's remove the cx.waker().wake_by_ref(); line:

bash

cargo run --example future_demo

text

poll: 3

One poll, and then nothing. The program hangs (Ctrl+C to stop it). Pending without a wake means: nobody will ever ask again. Put the line back.

4. async fn is a state machine, and why Pin exists

And async fn? The compiler turns it into a future like ours: a state machine. Each .await is a point where it can return Pending and later continue. Its local variables are stored inside the future.

That is where Pin comes in. A local variable can borrow another local variable, so the future can point into itself. If the future moved in memory, that pointer would point at the old place. Pin<&mut T> is a promise: this value will not move any more. That's why poll takes self: Pin<&mut Self>.

5. Timed: a wrapper future

Now something useful for our blog: Timed, a future that wraps another future and logs a warning when it takes longer than 50 ms. New file src/timed.rs (and mod timed; in main.rs):

rust

pub struct Timed<F> {
    inner: F,
    start: Option<Instant>,
    name: &'static str,
}

impl<F> Timed<F> {
    pub fn new(name: &'static str, inner: F) -> Timed<F> {
        Timed { inner, start: None, name }
    }
}

In poll we poll the inner future:

rust

impl<F: Future> Future for Timed<F> {
    type Output = F::Output;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<F::Output> {
        self.inner.poll(cx)
    }
}

bash

cargo build

text

error[E0599]: no method named `poll` found for type parameter `F` in the current scope

poll needs a Pin of the inner future, and we have a plain field. Pin::new of a mutable reference?

rust

Pin::new(&mut self.inner).poll(cx)

bash

cargo build

text

error[E0277]: `F` cannot be unpinned

Pin::new only works for types that don't care about moving: types that are Unpin. Futures from async blocks are not Unpin (they may point into themselves).

6. Pin projection with pin-project-lite

We need pin projection: from a pinned Timed, get a pinned inner field. Doing that by hand needs unsafe; the pin-project-lite crate writes it safely for us:

bash

cargo add pin-project-lite

rust

use pin_project_lite::pin_project;

pin_project! {
    pub struct Timed<F> {
        #[pin]
        inner: F,
        start: Option<Instant>,
        name: &'static str,
    }
}

#[pin] marks the field that stays pinned. self.project() gives us the fields: inner is a Pin<&mut F>, the others are normal &mut references:

rust

impl<F: Future> Future for Timed<F> {
    type Output = F::Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<F::Output> {
        let this = self.project();
        let start = *this.start.get_or_insert_with(Instant::now);
        let result = this.inner.poll(cx);
        if result.is_ready() {
            let took = start.elapsed();
            if took > Duration::from_millis(50) {
                tracing::warn!(name = *this.name, ?took, "slow future");
            }
        }
        result
    }
}

The clock starts at the first poll, and the time is checked when the inner future is ready.

7. Find a slow query

In the repository, list is wrapped: Timed::new, a name, and the query as an async move block:

rust

async fn list(&self, f: ListFilter) -> sqlx::Result<(Vec<Post>, i64)> {
    crate::timed::Timed::new("list_posts", async move {
    let mut q = QueryBuilder::<Sqlite>::new(format!("SELECT {} FROM posts p WHERE p.deleted_at IS NULL", Post::columns().join(", ")));
    // ... the rest of the query as before ...
    let total: i64 = c.build_query_scalar().fetch_one(&self.pool).await?;
    Ok::<_, sqlx::Error>((posts, total))
    }).await
}

Ok::<_, sqlx::Error> tells the compiler the error type of the block, because ? inside an async block can't infer it.

Run, and add fifty thousand posts:

bash

cargo run
sqlite3 blog.db "WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i+1 FROM n WHERE i<50000) INSERT INTO posts(title, body, slug, author_id) SELECT 'Post number '||i, 'Body '||i, 'post-number-'||i, 2 FROM n;"
curl -s 'localhost:8080/posts?sort=title' > /dev/null

The log says WARN slow future name="list_posts" took=.... Sorting by title has no index: a hint for episode 42.

The final code

examples/future_demo.rs

rust

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

struct Countdown(u32);

impl Future for Countdown {
    type Output = &'static str;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        println!("poll: {}", self.0);
        if self.0 == 0 {
            Poll::Ready("ready")
        } else {
            self.0 -= 1;
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

#[tokio::main]
async fn main() {
    println!("{}", Countdown(3).await);
}

src/main.rs

rust

mod auth;
mod config;
mod error;
mod extract;
mod handlers;
mod models;
mod repo;
mod storage;
mod mail;
mod jobs;
mod cache;
mod timed;
#[cfg(test)]
mod api_tests;
#[cfg(test)]
mod failing;

use axum::{routing::{get, post, put}, Extension, Router};
use repo::{Repo, SqliteRepo};
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
use std::str::FromStr;
use tower_http::trace::TraceLayer;
use tracing_subscriber::EnvFilter;
use axum::extract::DefaultBodyLimit;
use axum::http::{header, HeaderName, HeaderValue, Method, StatusCode};
use tower_http::cors::CorsLayer;
use tower_http::set_header::SetResponseHeaderLayer;
use tower_http::timeout::TimeoutLayer;
use std::sync::Arc;
use tower_governor::{governor::GovernorConfigBuilder, GovernorLayer};
use std::net::SocketAddr;
use storage::{LocalStorage, MemStorage, Storage};
use mail::Mailer;
use jobs::Job;
use tokio::sync::mpsc;
use tokio::time::Duration;
use redis::aio::{ConnectionManager, ConnectionManagerConfig};
use tokio::sync::broadcast;
use models::Comment;
use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle};
use axum::extract::{MatchedPath, Request};
use axum::middleware::{self, Next};
use axum::response::Response;
use utoipa::{Modify, OpenApi};
use utoipa_swagger_ui::SwaggerUi;
use utoipa::openapi::security::{HttpAuthScheme, HttpBuilder, SecurityScheme};

pub fn app<R: Repo, S: Storage>(repo: R, store: S, jobs: mpsc::Sender<Job>, cache: ConnectionManager, live: broadcast::Sender<Comment>, handle: PrometheusHandle) -> Router {
    let cors = CorsLayer::new()
        .allow_origin(HeaderValue::from_static("http://localhost:5500"))
        .allow_methods([Method::GET, Method::POST, Method::PUT, Method::DELETE])
        .allow_headers([header::AUTHORIZATION, header::CONTENT_TYPE])
        .allow_credentials(true);
    let governor = Arc::new(GovernorConfigBuilder::default().per_second(2).burst_size(5).finish().unwrap());
    let auth_routes = Router::new()
        .route("/signup", post(handlers::signup::<R>))
        .route("/login", post(handlers::login::<R>))
        .route("/auth/refresh", post(handlers::refresh::<R>))
        .layer(GovernorLayer::new(governor));
    let streaming = Router::new()
        .route("/posts/{id}/live", get(handlers::live))
        .route("/posts/{id}/events", get(handlers::events));
    Router::new()
        .nest("/api/v1", posts_v1())
        .nest("/api/v2", posts_v2())
        .merge(posts_v1())
        .route("/posts/{id}/restore", post(handlers::restore_post::<R>))
        .route("/me", get(handlers::me::<R>))
        .route("/admin/users", get(handlers::list_users::<R>))
        .route("/auth/logout", post(handlers::logout::<R>))
        .route(
            "/posts/{id}/images",
            post(handlers::upload_image::<R, S>).layer(DefaultBodyLimit::max(10 * 1024 * 1024)),
        )
        .route("/files/{name}", get(handlers::get_file::<S>))
        .route("/debug/slow", get(handlers::slow))
        .route("/health/live", get(handlers::health_live))
        .route("/health/ready", get(handlers::health_ready::<R>))
        .route("/metrics", get(handlers::metrics_text))
        .merge(auth_routes)
        .layer(TimeoutLayer::with_status_code(StatusCode::REQUEST_TIMEOUT, Duration::from_secs(10)))
        .merge(streaming)
        .merge(SwaggerUi::new("/swagger-ui").url("/api-docs/openapi.json", ApiDoc::openapi()))
        .route_layer(middleware::from_fn(track))
        .with_state(repo)
        .layer(Extension(Arc::new(store)))
        .layer(Extension(jobs))
        .layer(Extension(cache))
        .layer(Extension(live))
        .layer(Extension(handle))
        .layer(cors)
        .layer(SetResponseHeaderLayer::if_not_present(header::X_CONTENT_TYPE_OPTIONS, HeaderValue::from_static("nosniff")))
        .layer(SetResponseHeaderLayer::if_not_present(header::X_FRAME_OPTIONS, HeaderValue::from_static("DENY")))
        .layer(SetResponseHeaderLayer::if_not_present(header::REFERRER_POLICY, HeaderValue::from_static("no-referrer")))
        .layer(TraceLayer::new_for_http())
}

#[tokio::main]
async fn main() {
    let cfg = config::config();
    tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")))
        .init();
    let addr = format!("{}:{}", cfg.host, cfg.port);
    let options = SqliteConnectOptions::from_str(&cfg.database_url)
        .unwrap()
        .create_if_missing(true)
        .journal_mode(SqliteJournalMode::Wal);
    let pool = SqlitePoolOptions::new()
        .max_connections(5)
        .connect_with(options)
        .await
        .unwrap();
    sqlx::migrate!().run(&pool).await.unwrap();
    let repo = SqliteRepo { pool };
    let (tx, rx) = mpsc::channel::<Job>(100);
    let mailer = Mailer::new(cfg);
    tokio::spawn(jobs::worker(rx, mailer));
    tokio::spawn(jobs::cleanup(repo.clone(), Duration::from_secs(cfg.cleanup_secs)));
    let client = redis::Client::open(cfg.redis_url.as_str()).unwrap();
    let redis_config = ConnectionManagerConfig::new()
        .set_number_of_retries(1)
        .set_connection_timeout(Some(Duration::from_millis(300)))
        .set_response_timeout(Some(Duration::from_millis(300)));
    let cache = ConnectionManager::new_with_config(client, redis_config).await.unwrap();
    let (live, _) = broadcast::channel::<Comment>(100);
    let handle = PrometheusBuilder::new().set_buckets(&[0.005, 0.01, 0.05, 0.1, 0.5, 1.0]).unwrap().install_recorder().unwrap();
    match cfg.storage.as_str() {
        "memory" => serve(app(repo, MemStorage::default(), tx, cache, live, handle), &addr).await,
        _ => serve(app(repo, LocalStorage::new("uploads"), tx, cache, live, handle), &addr).await,
    }
}

async fn serve(app: Router, addr: &str) {
    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
    tracing::info!(%addr, "listening");
    axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>())
        .with_graceful_shutdown(shutdown_signal())
        .await
        .unwrap();
}

fn posts_common<R: Repo>() -> Router<R> {
    Router::new()
        .route("/posts", get(handlers::list_posts::<R>).post(handlers::create_post::<R>))
        .route(
            "/posts/{id}",
            put(handlers::update_post::<R>).delete(handlers::delete_post::<R>),
        )
        .route("/posts/{id}/comments", get(handlers::list_comments::<R>).post(handlers::add_comment::<R>))
        .route("/search", get(handlers::search_posts::<R>))
}

fn posts_v1<R: Repo>() -> Router<R> {
    posts_common()
        .route("/posts/{id}", get(handlers::get_post::<R>))
        .layer(SetResponseHeaderLayer::overriding(HeaderName::from_static("deprecation"), HeaderValue::from_static("true")))
        .layer(SetResponseHeaderLayer::overriding(
            HeaderName::from_static("sunset"),
            HeaderValue::from_static("Wed, 30 Jun 2027 00:00:00 GMT"),
        ))
        .layer(SetResponseHeaderLayer::overriding(header::LINK, HeaderValue::from_static(r#"</api/v2>; rel="successor-version""#)))
}

fn posts_v2<R: Repo>() -> Router<R> {
    posts_common().route("/posts/{id}", get(handlers::get_post_v2::<R>))
}

async fn shutdown_signal() {
    let ctrl_c = async {
        tokio::signal::ctrl_c().await.expect("Ctrl+C handler");
    };
    let terminate = async {
        tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
            .expect("SIGTERM handler")
            .recv()
            .await;
    };
    tokio::select! {
        _ = ctrl_c => {},
        _ = terminate => {},
    }
    tracing::info!("shutting down");
}

async fn track(req: Request, next: Next) -> Response {
    let start = std::time::Instant::now();
    let method = req.method().to_string();
    let path = req
        .extensions()
        .get::<MatchedPath>()
        .map(|p| p.as_str().to_string())
        .unwrap_or_else(|| "unmatched".to_string());
    let response = next.run(req).await;
    let status = response.status().as_u16().to_string();
    let labels = [("method", method), ("path", path), ("status", status)];
    metrics::counter!("http_requests_total", &labels).increment(1);
    metrics::histogram!("http_request_duration_seconds", &labels).record(start.elapsed().as_secs_f64());
    response
}

#[derive(OpenApi)]
#[openapi(
    paths(handlers::get_post_v2, handlers::create_post),
    components(schemas(models::PostV2, models::NewPost, models::PostWithTags, models::Post, error::ErrorBody)),
    modifiers(&SecurityAddon)
)]
struct ApiDoc;

struct SecurityAddon;

impl Modify for SecurityAddon {
    fn modify(&self, openapi: &mut utoipa::openapi::OpenApi) {
        let components = openapi.components.get_or_insert_with(Default::default);
        components.add_security_scheme(
            "bearer",
            SecurityScheme::Http(HttpBuilder::new().scheme(HttpAuthScheme::Bearer).bearer_format("JWT").build()),
        );
    }
}

src/repo.rs

rust

use std::future::Future;

use sqlx::{QueryBuilder, Sqlite, SqlitePool};

use crate::models::{slugify, Comment, Hit, Image, ListFilter, NewComment, NewPost, Post, PostWithTags, Sort, User, UserWithHash};

pub trait Repo: Clone + Send + Sync + 'static {
    fn list(&self, f: ListFilter) -> impl Future<Output = sqlx::Result<(Vec<Post>, i64)>> + Send;
    fn get(&self, id: i64) -> impl Future<Output = sqlx::Result<Option<Post>>> + Send;
    fn create(&self, author_id: i64, new: NewPost) -> impl Future<Output = sqlx::Result<PostWithTags>> + Send;
    fn update(&self, id: i64, new: NewPost) -> impl Future<Output = sqlx::Result<Option<Post>>> + Send;
    fn delete(&self, id: i64) -> impl Future<Output = sqlx::Result<bool>> + Send;
    fn restore(&self, id: i64) -> impl Future<Output = sqlx::Result<bool>> + Send;
    fn list_comments(&self, post_id: i64) -> impl Future<Output = sqlx::Result<Vec<Comment>>> + Send;
    fn add_comment(&self, post_id: i64, new: NewComment) -> impl Future<Output = sqlx::Result<Comment>> + Send;
    fn search(&self, q: &str) -> impl Future<Output = sqlx::Result<Vec<Hit>>> + Send;
    fn create_user(&self, email: &str, password_hash: &str) -> impl Future<Output = sqlx::Result<User>> + Send;
    fn find_user_by_email(&self, email: &str) -> impl Future<Output = sqlx::Result<Option<UserWithHash>>> + Send;
    fn get_user(&self, id: i64) -> impl Future<Output = sqlx::Result<Option<User>>> + Send;
    fn store_refresh(&self, hash: &str, user_id: i64, expires_at: i64) -> impl Future<Output = sqlx::Result<()>> + Send;
    fn take_refresh(&self, hash: &str, now: i64) -> impl Future<Output = sqlx::Result<Option<i64>>> + Send;
    fn delete_refresh(&self, hash: &str) -> impl Future<Output = sqlx::Result<()>> + Send;
    fn delete_expired_refresh(&self, now: i64) -> impl Future<Output = sqlx::Result<u64>> + Send;
    fn list_users(&self) -> impl Future<Output = sqlx::Result<Vec<User>>> + Send;
    fn ping(&self) -> impl Future<Output = sqlx::Result<()>> + Send;
    fn add_image(&self, post_id: i64, url: &str) -> impl Future<Output = sqlx::Result<Image>> + Send;
}

#[derive(Clone)]
pub struct SqliteRepo {
    pub pool: SqlitePool,
}

impl Repo for SqliteRepo {
    async fn list(&self, f: ListFilter) -> sqlx::Result<(Vec<Post>, i64)> {
        crate::timed::Timed::new("list_posts", async move {
        let mut q = QueryBuilder::<Sqlite>::new(format!("SELECT {} FROM posts p WHERE p.deleted_at IS NULL", Post::columns().join(", ")));
        if let Some(tag) = &f.tag {
            q.push(" AND p.id IN (SELECT pt.post_id FROM post_tags pt JOIN tags t ON t.id = pt.tag_id WHERE t.name = ")
                .push_bind(tag.clone())
                .push(")");
        }
        q.push(match f.sort {
            Sort::Newest => " ORDER BY p.id DESC",
            Sort::Oldest => " ORDER BY p.id ASC",
            Sort::Title => " ORDER BY p.title ASC",
        });
        q.push(" LIMIT ").push_bind(f.per_page as i64);
        q.push(" OFFSET ").push_bind(((f.page - 1) * f.per_page) as i64);
        let posts = q.build_query_as::<Post>().fetch_all(&self.pool).await?;

        let mut c = QueryBuilder::<Sqlite>::new("SELECT COUNT(*) FROM posts p WHERE p.deleted_at IS NULL");
        if let Some(tag) = &f.tag {
            c.push(" AND p.id IN (SELECT pt.post_id FROM post_tags pt JOIN tags t ON t.id = pt.tag_id WHERE t.name = ")
                .push_bind(tag.clone())
                .push(")");
        }
        let total: i64 = c.build_query_scalar().fetch_one(&self.pool).await?;
        Ok::<_, sqlx::Error>((posts, total))
        }).await
    }

    async fn search(&self, q: &str) -> sqlx::Result<Vec<Hit>> {
        let q = fts_query(q);
        sqlx::query_as!(
            Hit,
            r#"SELECT p.id as "id!", p.title, snippet(posts_fts, 1, '[', ']', '...', 8) as "snippet!: String"
               FROM posts_fts JOIN posts p ON p.id = posts_fts.rowid
               WHERE posts_fts MATCH ? AND p.deleted_at IS NULL
               ORDER BY bm25(posts_fts) LIMIT 20"#,
            q
        )
        .fetch_all(&self.pool)
        .await
    }

    async fn get(&self, id: i64) -> sqlx::Result<Option<Post>> {
        sqlx::query_as!(Post, "SELECT id, title, body, slug, created_at, author_id FROM posts WHERE id = ? AND deleted_at IS NULL", id)
            .fetch_optional(&self.pool)
            .await
    }

    async fn create(&self, author_id: i64, new: NewPost) -> sqlx::Result<PostWithTags> {
        let slug = slugify(&new.title);
        let mut tx = self.pool.begin().await?;
        let post = sqlx::query_as!(
            Post,
            r#"INSERT INTO posts (title, body, slug, author_id) VALUES (?, ?, ?, ?) RETURNING id, title, body, slug, created_at, author_id"#,
            new.title,
            new.body,
            slug,
            author_id
        )
        .fetch_one(&mut *tx)
        .await?;
        for tag in &new.tags {
            sqlx::query!("INSERT INTO tags (name) VALUES (?) ON CONFLICT (name) DO NOTHING", tag)
                .execute(&mut *tx)
                .await?;
            sqlx::query!(
                "INSERT INTO post_tags (post_id, tag_id) SELECT ?, id FROM tags WHERE name = ?",
                post.id,
                tag
            )
            .execute(&mut *tx)
            .await?;
        }
        tx.commit().await?;
        Ok(PostWithTags { post, tags: new.tags })
    }

    async fn update(&self, id: i64, new: NewPost) -> sqlx::Result<Option<Post>> {
        sqlx::query_as!(
            Post,
            r#"UPDATE posts SET title = ?, body = ? WHERE id = ? AND deleted_at IS NULL RETURNING id, title, body, slug, created_at, author_id"#,
            new.title,
            new.body,
            id
        )
        .fetch_optional(&self.pool)
        .await
    }

    async fn delete(&self, id: i64) -> sqlx::Result<bool> {
        let result = sqlx::query!("UPDATE posts SET deleted_at = CURRENT_TIMESTAMP WHERE id = ? AND deleted_at IS NULL", id)
            .execute(&self.pool)
            .await?;
        Ok(result.rows_affected() > 0)
    }

    async fn restore(&self, id: i64) -> sqlx::Result<bool> {
        let result = sqlx::query!("UPDATE posts SET deleted_at = NULL WHERE id = ? AND deleted_at IS NOT NULL", id)
            .execute(&self.pool)
            .await?;
        Ok(result.rows_affected() > 0)
    }

    async fn create_user(&self, email: &str, password_hash: &str) -> sqlx::Result<User> {
        sqlx::query_as!(
            User,
            r#"INSERT INTO users (email, password_hash) VALUES (?, ?) RETURNING id, email, role, created_at"#,
            email,
            password_hash
        )
        .fetch_one(&self.pool)
        .await
    }

    async fn find_user_by_email(&self, email: &str) -> sqlx::Result<Option<UserWithHash>> {
        sqlx::query_as!(
            UserWithHash,
            "SELECT id, email, password_hash, role FROM users WHERE email = ?",
            email
        )
        .fetch_optional(&self.pool)
        .await
    }

    async fn get_user(&self, id: i64) -> sqlx::Result<Option<User>> {
        sqlx::query_as!(User, "SELECT id, email, role, created_at FROM users WHERE id = ?", id)
            .fetch_optional(&self.pool)
            .await
    }

    async fn store_refresh(&self, hash: &str, user_id: i64, expires_at: i64) -> sqlx::Result<()> {
        sqlx::query!(
            "INSERT INTO refresh_tokens (token_hash, user_id, expires_at) VALUES (?, ?, ?)",
            hash,
            user_id,
            expires_at
        )
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    async fn take_refresh(&self, hash: &str, now: i64) -> sqlx::Result<Option<i64>> {
        sqlx::query_scalar!(
            r#"DELETE FROM refresh_tokens WHERE token_hash = ? AND expires_at > ? RETURNING user_id as "user_id!""#,
            hash,
            now
        )
        .fetch_optional(&self.pool)
        .await
    }

    async fn delete_refresh(&self, hash: &str) -> sqlx::Result<()> {
        sqlx::query!("DELETE FROM refresh_tokens WHERE token_hash = ?", hash).execute(&self.pool).await?;
        Ok(())
    }

    async fn delete_expired_refresh(&self, now: i64) -> sqlx::Result<u64> {
        let done = sqlx::query!("DELETE FROM refresh_tokens WHERE expires_at < ?", now).execute(&self.pool).await?;
        Ok(done.rows_affected())
    }

    async fn list_comments(&self, post_id: i64) -> sqlx::Result<Vec<Comment>> {
        sqlx::query_as!(
            Comment,
            "SELECT id, post_id, body, created_at FROM comments WHERE post_id = ? ORDER BY id",
            post_id
        )
        .fetch_all(&self.pool)
        .await
    }

    async fn add_comment(&self, post_id: i64, new: NewComment) -> sqlx::Result<Comment> {
        sqlx::query_as!(
            Comment,
            r#"INSERT INTO comments (post_id, body) VALUES (?, ?) RETURNING id as "id!", post_id, body, created_at"#,
            post_id,
            new.body
        )
        .fetch_one(&self.pool)
        .await
    }

    async fn list_users(&self) -> sqlx::Result<Vec<User>> {
        sqlx::query_as!(User, "SELECT id, email, role, created_at FROM users ORDER BY id")
            .fetch_all(&self.pool)
            .await
    }

    async fn ping(&self) -> sqlx::Result<()> {
        sqlx::query("SELECT 1").execute(&self.pool).await?;
        Ok(())
    }

    async fn add_image(&self, post_id: i64, url: &str) -> sqlx::Result<Image> {
        sqlx::query_as!(
            Image,
            r#"INSERT INTO images (post_id, url) VALUES (?, ?) RETURNING id as "id!", post_id, url"#,
            post_id,
            url
        )
        .fetch_one(&self.pool)
        .await
    }
}

fn fts_query(input: &str) -> String {
    input
        .split_whitespace()
        .map(|word| format!("\"{}\"", word.replace('"', "\"\"")))
        .collect::<Vec<_>>()
        .join(" ")
}

src/timed.rs

rust

use std::future::Future;
use pin_project_lite::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};

pin_project! {
    pub struct Timed<F> {
        #[pin]
        inner: F,
        start: Option<Instant>,
        name: &'static str,
    }
}

impl<F> Timed<F> {
    pub fn new(name: &'static str, inner: F) -> Timed<F> {
        Timed { inner, start: None, name }
    }
}

impl<F: Future> Future for Timed<F> {
    type Output = F::Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<F::Output> {
        let this = self.project();
        let start = *this.start.get_or_insert_with(Instant::now);
        let result = this.inner.poll(cx);
        if result.is_ready() {
            let took = start.elapsed();
            if took > Duration::from_millis(50) {
                tracing::warn!(name = *this.name, ?took, "slow future");
            }
        }
        result
    }
}

Recap

A Future has one method, poll: Ready(value) or Pending
Returning Pending without arranging a wake-up hangs the task forever
async fn compiles to a state machine that can point into itself, so it must not move: Pin
Pin::new only works for Unpin types; async blocks aren't Unpin
pin-project-lite gives safe pin projection: #[pin] fields and self.project()
A wrapper future like Timed can measure any async work

Next episode: unsafe Rust, and calling C code from Rust. Watch the series on YouTube.