Function consume_remote

nodedb/src/event/cdc/consume.rs:236–313

Forward a consume request to the remote partition leader via the gateway. Routes the stream SELECT SQL through `gateway.execute_sql`, which plans it locally and dispatches it as an `ExecuteRequest` over QUIC to the correct leader node. The `leader_node` parameter is accepted for caller compatibility but is ignored — the gateway handles node selection.

(
    state: &SharedState,
    params: &ConsumeParams<'_>,
    _leader_node: u64,
)

/// leader node. The `leader_node` parameter is accepted for caller
/// compatibility but is ignored — the gateway handles node selection.
pub async fn consume_remote(
    state: &SharedState,
    params: &ConsumeParams<'_>,
    _leader_node: u64,
) -> Result<ConsumeResult, ConsumeError> {
    let gateway = state
        .gateway
        .as_ref()
        .ok_or(ConsumeError::NoClusterTransport)?;

    let sql = build_consume_sql(params);
    let tenant_id = params.tenant_id;

    let gw_ctx = crate::control::gateway::core::QueryContext {
        tenant_id: crate::types::TenantId::new(tenant_id),
        trace_id: nodedb_types::TraceId::generate(),
        database_id: nodedb_types::id::DatabaseId::DEFAULT,
    };

    let query_ctx = crate::control::planner::context::QueryContext::for_state(state);

    let payloads = gateway
        .execute_sql(&gw_ctx, &sql, &[], || {
            let tasks = tokio::task::block_in_place(|| {
                tokio::runtime::Handle::current().block_on(query_ctx.plan_sql(
                    &sql,
                    crate::types::TenantId::new(tenant_id),
                    crate::types::DatabaseId::DEFAULT,
                ))
            })
            .map_err(|e| crate::Error::PlanError {
                detail: e.to_string(),
            })?;
            // Take the first task's plan (stream reads are single-task).
            tasks
                .into_iter()
                .next()
                .map(|t| t.plan)
                .ok_or_else(|| crate::Error::PlanError {
                    detail: "stream SELECT produced no physical tasks".into(),
                })
        })
        .await
        .map_err(|e| ConsumeError::RemoteError(e.to_string()))?;

    // Deserialize events from the response payloads.
    // Payloads contain msgpack-serialised Vec<CdcEvent>.
    let events: Vec<Arc<CdcEvent>> = if let Some(payload) = payloads.first() {
        zerompk::from_msgpack::<Vec<CdcEvent>>(payload)
            .unwrap_or_default()
            .into_iter()
            .map(Arc::new)
            .collect()
    } else {
        Vec::new()
    };

    // Compute per-partition max LSN for the returned batch.