Function start_rpc_server

apps/hash-graph/src/subcommand/server.rs:256–312

(
    address: RpcAddress,
    dependencies: Dependencies<S, ()>,
    lifecycle: &ServerLifecycle,
)

}

fn start_rpc_server<S>(
    address: RpcAddress,
    dependencies: Dependencies<S, ()>,
    lifecycle: &ServerLifecycle,
) -> Result<(), Report<GraphError>>
where
    S: StorePool + Send + Sync + 'static,
    for<'p> S::Store<'p>: PrincipalStore,
{
    let server = Server::new(harpc_server::ServerConfig::default()).change_context(GraphError)?;
    let cancellation_token = server.cancellation_token();

    let (router, task) = hash_graph_api::rpc::rpc_router(
        Dependencies {
            store: dependencies.store,
            temporal_client: dependencies.temporal_client,
            codec: JsonCodec,
        },
        server.events(),
    );

    lifecycle.spawn("RPC task", async move {
        task.await;
        Ok(())
    });

    // Bridge: when our shutdown token fires, cancel the harpc server's internal token
    let shutdown = lifecycle.shutdown.clone();
    lifecycle.spawn("RPC shutdown bridge", async move {
        shutdown.cancelled().await;
        cancellation_token.cancel();
        Ok(())
    });

    let socket_address: SocketAddr = SocketAddr::try_from(address).change_context(GraphError)?;
    let mut address = Multiaddr::empty();
    match socket_address {
        SocketAddr::V4(v4) => {
            address.push(Protocol::Ip4(*v4.ip()));
            address.push(Protocol::Tcp(v4.port()));
        }
        SocketAddr::V6(v6) => {
            address.push(Protocol::Ip6(*v6.ip()));
            address.push(Protocol::Tcp(v6.port()));
        }
    }

    #[expect(clippy::significant_drop_tightening, reason = "false positive")]
    lifecycle.spawn("RPC server", async move {
        let stream = server.listen(address).await.change_context(GraphError)?;

        harpc_server::serve::serve(stream, router).await;
        Ok(())
    });

    Ok(())
}