如何在async-graphql与axum间共享状态？代码报错求助

问题背景 #

使用Axum和async-graphql构建GraphQL服务器时，需要将Axum接收到的身份认证HTTP Header传递给GraphQL模块，但当前状态共享方案导致生命周期报错，无法实现需求。

项目配置 #

Cargo.toml #

[package]
name = "app"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = { version = "0.6.1", features = ["headers"] }
tokio = { version = "1.19.2", features = ["rt-multi-thread", "macros"] }
serde = { version = "1.0.136", features = ["derive"] }
async-graphql = { version = "5.0.4", features = ["chrono"] }
sqlx = { version = "0.6.2", features = [ "runtime-actix-native-tls", "postgres", "chrono" ] }
dotenv = "0.15.0"
tower-http = { version = "0.3.5", features = ["cors", "trace"] }
tokio-stream = "0.1.11"
chrono = "0.4.23"
jsonwebtoken = "8.2.0"
thiserror = "1.0.38"
async-trait = "0.1.60"

核心代码 #

main.rs #

mod db;
mod repositories;
mod resolvers;

use async_graphql::{
    http::{playground_source, GraphQLPlaygroundConfig},
    Request, Response, Schema,
};
use axum::{
    extract::{Extension, State},
    http::{
        header::{ACCEPT, AUTHORIZATION},
        HeaderValue, Method, Request as AxumRequest,
    },
    middleware::Next,
    response::{Html, IntoResponse, Response as AxumResponse},
    routing::get,
    Json, Router,
};
use dotenv::dotenv;
use resolvers::{QueryRoot, Subscription};
use std::net::SocketAddr;
use tower_http::cors::CorsLayer;

use crate::{db::DB, resolvers::Mutation};

pub type MainSchema = Schema<QueryRoot, Mutation, Subscription>;

async fn graphql_handler(schema: Extension<MainSchema>, req: Json<Request>) -> Json<Response> {
    schema.execute(req.0).await.into()
}

async fn graphql_playground() -> impl IntoResponse {
    Html(playground_source(GraphQLPlaygroundConfig::new("/")))
}

#[derive(Clone, Debug)]
pub struct AppState {
    db: DB,
    token: Option<String>,
}

impl AppState {
    fn set_token(mut self, token: String) {
        self.token = Some(token)
    }
}

async fn propagate_header<B>(
    State(state): State<&AppState>,
    req: AxumRequest<B>,
    next: Next<B>,
) -> AxumResponse {
    let token = req.headers().get("Authorization");

    if token.is_some() {
        // TODO: Put token in state.
    };

    next.run(req).await
}

#[tokio::main]
async fn main() {
    dotenv().ok();
    let server = async {
        let db = DB::new().await;
        let state = AppState { db, token: None };
        let schema = Schema::build(QueryRoot, Mutation, Subscription)
            // .limit_depth(5)
            .data(&state)
            .finish();

        let cors_layer = CorsLayer::new()
            .allow_origin("*".parse::<HeaderValue>().unwrap())
            .allow_methods([Method::GET, Method::POST, Method::OPTIONS])
            .allow_headers(vec![AUTHORIZATION, ACCEPT]);

        let app = Router::new()
            .route("/", get(graphql_playground).post(graphql_handler))
            .layer(cors_layer)
            .layer(axum::middleware::from_fn_with_state(
                &state,
                propagate_header,
            ))
            .layer(Extension(schema));

        let addr = SocketAddr::from(([0, 0, 0, 0], 8009));
        axum::Server::bind(&addr)
            .serve(app.into_make_service())
            .await
            .unwrap();
    };

    tokio::join!(server);
}

报错信息 #

error[E0597]: `state` does not live long enough
  --> src/main.rs:71:19
   |
69 |           let schema = Schema::build(QueryRoot, Mutation, Subscription)
   |  ______________________-
70 | |             // .limit_depth(5)
71 | |             .data(&state)
   | |___________________^^^^^^- argument requires that `state` is borrowed for `'static`
   |                     |
   |                     borrowed value does not live long enough
...
93 |       };
   |       - `state` dropped here while still borrowed

解决方案 #

错误根源 #

1. 生命周期不匹配：Schema::build().data()要求传入的数据具有'static生命周期（因为Schema会长期驻留内存），但当前state是main函数内的局部变量，生命周期仅到server闭包结束，无法满足'static要求。
2. 状态设计错误：试图用全局共享的AppState存储请求级别的token，异步环境下多请求并发会导致数据竞争，完全不可行。

正确实现思路 #

不要通过全局状态传递请求级别的认证信息，而是在处理GraphQL请求时直接从Axum的请求中提取Header，注入到GraphQL的请求上下文里。

步骤1：修改GraphQL请求处理器，注入token

将graphql_handler修改为能提取Axum请求的Header，并将token注入到GraphQL Request的上下文：

async fn graphql_handler(
    schema: Extension<MainSchema>,
    req: Json<Request>,
    axum_req: AxumRequest<()>,
) -> Json<Response> {
    // 提取Authorization Header
    let token = axum_req.headers()
        .get(AUTHORIZATION)
        .and_then(|hv| hv.to_str().ok())
        .map(|s| s.to_string());

    // 将token注入GraphQL请求的上下文
    let mut graphql_req = req.0;
    graphql_req = graphql_req.data(token);

    schema.execute(graphql_req).await.into()
}

或者用Axum自带的Header提取器更简洁：

use axum::headers::{Authorization, Bearer};

async fn graphql_handler(
    schema: Extension<MainSchema>,
    req: Json<Request>,
    #[axum::extract(headers)] auth: Option<Authorization<Bearer>>,
) -> Json<Response> {
    let token = auth.map(|a| a.token().to_string());
    let mut graphql_req = req.0;
    graphql_req = graphql_req.data(token);

    schema.execute(graphql_req).await.into()
}

步骤2：在Resolver中获取token

在GraphQL的Resolver函数中，通过Context获取注入的token：

use async_graphql::Context;

// 示例Resolver
#[async_graphql::Object]
impl QueryRoot {
    async fn current_user(&self, ctx: &Context<'_>) -> Result<User, Error> {
        // 从上下文取出token
        let token = ctx.data_opt::<Option<String>>()
            .flatten()
            .ok_or(Error::Unauthorized)?;

        // 验证token并查询用户逻辑...
        Ok(User { id: 1, name: "test".into() })
    }
}

步骤3：修正DB的生命周期问题

如果需要将DB实例传入Schema，需要将其包装为Arc（线程安全引用计数），使其满足'static生命周期：

use std::sync::Arc;

#[tokio::main]
async fn main() {
    dotenv().ok();
    let server = async {
        let db = Arc::new(DB::new().await);
        // 构建Schema时传入Arc包装的DB
        let schema = Schema::build(QueryRoot, Mutation, Subscription)
            .data(db.clone())
            .finish();

        let cors_layer = CorsLayer::new()
            .allow_origin("*".parse::<HeaderValue>().unwrap())
            .allow_methods([Method::GET, Method::POST, Method::OPTIONS])
            .allow_headers(vec![AUTHORIZATION, ACCEPT]);

        let app = Router::new()
            .route("/", get(graphql_playground).post(graphql_handler))
            .layer(cors_layer)
            .layer(Extension(schema));

        let addr = SocketAddr::from(([0, 0, 0, 0], 8009));
        axum::Server::bind(&addr)
            .serve(app.into_make_service())
            .await
            .unwrap();
    };

    tokio::join!(server);
}

然后在Resolver中获取DB：

async fn some_query(&self, ctx: &Context<'_>) -> Result<...> {
    let db = ctx.data::<Arc<DB>>()?;
    // 使用db执行查询...
}

额外说明 #

• 移除原来的AppState和propagate_header中间件，因为请求级别的token不需要全局存储。
• 异步环境下，全局可变状态必须用Mutex或RwLock包裹，但请求级别的数据完全不需要全局共享，直接通过上下文传递是最安全高效的方式。

内容的提问来源于stack exchange，提问作者kjt