nodedb_cluster/
forward.rs

1// SPDX-License-Identifier: BUSL-1.1
2
3//! Physical-plan execution trait for leader-based request routing.
4//!
5//! [`PlanExecutor`]: the physical-plan execution path introduced in C-β.
6//! The legacy [`RequestForwarder`] SQL-string path was deleted in C-δ.6.
7
8use crate::rpc_codec::{ExecuteRequest, ExecuteResponse, TypedClusterError};
9
10// ── Physical-plan execution (C-β) ────────────────────────────────────────────
11
12/// Sink for streamed result chunks driven by [`PlanExecutor::execute_plan_streaming`].
13///
14/// The transport server provides a `ChunkSink` whose `send_chunk` writes one
15/// `RPC_EXECUTE_STREAM_CHUNK` envelope to the QUIC stream and awaits the write
16/// inline, so QUIC flow control back-pressures the producer (the local stream
17/// pull) when the coordinator falls behind.
18///
19/// `send_chunk` returning `Err` means the client / coordinator is gone (the
20/// stream was reset or finished early): the producer MUST stop pulling and
21/// return without writing a terminal frame.
22///
23/// `read_version_lsn` is that frame's per-collection read version (its scanned
24/// collection's `coll_write_lsn` at read time), distinct from the core-global
25/// `watermark_lsn`. The shuffle fan-out sink max-folds it so the producer can
26/// report the observed read version for cross-shard OCC validation; the plain
27/// `ExecuteStream` sink (which only carries `watermark_lsn` on the wire) ignores
28/// it.
29pub trait ChunkSink: Send {
30    fn send_chunk(
31        &mut self,
32        payload: Vec<u8>,
33        watermark_lsn: u64,
34        read_version_lsn: u64,
35    ) -> impl std::future::Future<Output = crate::error::Result<()>> + Send;
36}
37
38/// Trait for executing a pre-planned `PhysicalPlan` on the local Data Plane.
39///
40/// Implemented in `nodedb/src/control/exec_receiver.rs` by `LocalPlanExecutor`.
41/// The cluster RPC handler calls this when it receives an `ExecuteRequest`.
42///
43/// Responsibilities:
44/// 1. Validate that `deadline_remaining_ms > 0`.
45/// 2. For each `DescriptorVersionEntry`, verify the local descriptor version matches.
46/// 3. Decode `plan_bytes` via `nodedb::bridge::physical_plan::wire::decode`.
47/// 4. Dispatch through the local SPSC bridge.
48/// 5. Collect response payloads.
49/// 6. Map errors to `TypedClusterError`.
50pub trait PlanExecutor: Send + Sync + 'static {
51    fn execute_plan(
52        &self,
53        req: ExecuteRequest,
54    ) -> impl std::future::Future<Output = ExecuteResponse> + Send;
55
56    /// Streaming sibling of [`execute_plan`](Self::execute_plan).
57    ///
58    /// Executes the plan and pushes each result frame to `sink` as it arrives.
59    /// Returns `None` on a clean EOF (all chunks delivered), or `Some(err)` on a
60    /// terminal failure — validation rejection, decode failure, deadline, or an
61    /// error pulled from the underlying result stream. A `send_chunk` error
62    /// (coordinator gone) is NOT a terminal error: the producer stops and
63    /// returns `None`, since there is no live peer to receive a terminal frame.
64    fn execute_plan_streaming(
65        &self,
66        req: ExecuteRequest,
67        sink: impl ChunkSink,
68    ) -> impl std::future::Future<Output = Option<TypedClusterError>> + Send;
69}
70
71/// No-op executor for single-node mode or testing.
72pub struct NoopPlanExecutor;
73
74impl PlanExecutor for NoopPlanExecutor {
75    async fn execute_plan(&self, _req: ExecuteRequest) -> ExecuteResponse {
76        ExecuteResponse::err(TypedClusterError::Internal {
77            code: 0,
78            message: "plan execution not available (single-node mode)".into(),
79        })
80    }
81
82    async fn execute_plan_streaming(
83        &self,
84        _req: ExecuteRequest,
85        _sink: impl ChunkSink,
86    ) -> Option<TypedClusterError> {
87        Some(TypedClusterError::Internal {
88            code: 0,
89            message: "streaming plan execution not available (single-node mode)".into(),
90        })
91    }
92}
nodedb_cluster/forward.rs

nodedb_cluster/
forward.rs
