cratestack_client_rust/rpc/
batch.rs

1// -----------------------------------------------------------------------------
2// Typed batch surface
3//
4// Lets callers compose heterogeneous batches of typed RPC ops into a
5// single `POST /rpc/batch` round-trip without dropping to the raw
6// `RpcRequest` / `RpcResponseFrame` wire types. Typical use through
7// the macro-generated client:
8//
9//   let mut batch = client.batch();
10//   let h_widgets = client.widgets().list(&list_input).queue(&mut batch);
11//   let h_ping    = client.procedures().ping(&args).queue(&mut batch);
12//   let h_created = client.widgets().create(&new).queue(&mut batch);
13//
14//   let mut results = batch.send().await?;            // one HTTP call
15//   let widgets:  Vec<Widget> = results.take(h_widgets)?;
16//   let echoed:   PingArgs    = results.take(h_ping)?;
17//   let created:  Widget      = results.take(h_created)?;
18//
19// `BatchableCall<C, O>` is what every macro-generated unary RPC method
20// now returns. It implements `IntoFuture`, so `.await` on it fires the
21// call immediately exactly like before — `.queue(&mut batch)` is the
22// opt-in deferral path. No `_batched` API duplication; same method,
23// two consumption modes.
24//
25// Sequence-streaming methods (`call_streaming` under the hood) stay as
26// `async fn -> Result<RpcStream<O>, _>` and do NOT participate in
27// batches — `/rpc/batch` is unary by construction.
28// -----------------------------------------------------------------------------
29
30use cratestack_core::CratestackError;
31
32use crate::codec::HttpClientCodec;
33use crate::remote_refusal::Challenges;
34use crate::rpc::batch_call::BatchHandle;
35use crate::rpc::client::RpcClient;
36use crate::rpc::error::{RpcClientError, RpcRemoteError, http_status_for_rpc_code};
37
38/// Accumulates queued [`BatchableCall`]s into a single
39/// `POST /rpc/batch` round-trip. Build via [`RpcClient::batch_builder`]
40/// or the macro-generated `Client::batch()`.
41///
42/// Send-on-drop is intentionally *not* implemented — the batch only
43/// fires when you call `.send().await`. Drops without sending are
44/// silent (queued calls just discarded).
45#[must_use = "BatchBuilder does nothing until `.send().await`ed"]
46pub struct BatchBuilder<C> {
47    rpc: RpcClient<C>,
48    frames: Vec<cratestack_core::rpc::RpcRequest>,
49    /// Frames whose input failed to encode pre-send — recorded by id
50    /// so [`BatchResults::take`] can surface the error per-handle
51    /// instead of poisoning the whole batch.
52    encode_errors: std::collections::HashMap<u64, CratestackError>,
53    next_id: u64,
54}
55
56impl<C> BatchBuilder<C>
57where
58    C: HttpClientCodec + Clone + Send + 'static,
59{
60    pub(crate) fn new(rpc: RpcClient<C>) -> Self {
61        Self {
62            rpc,
63            frames: Vec::new(),
64            encode_errors: std::collections::HashMap::new(),
65            next_id: 0,
66        }
67    }
68
69    /// Number of queued frames (including ones whose input failed
70    /// to encode — those still consume an id and will surface their
71    /// error on the matching `take`).
72    pub fn len(&self) -> usize {
73        self.frames.len() + self.encode_errors.len()
74    }
75
76    pub fn is_empty(&self) -> bool {
77        self.len() == 0
78    }
79
80    fn next_id(&mut self) -> u64 {
81        let id = self.next_id;
82        self.next_id += 1;
83        id
84    }
85
86    pub(crate) fn push_frame(&mut self, op_id: String, input: serde_json::Value) -> u64 {
87        let id = self.next_id();
88        self.frames.push(cratestack_core::rpc::RpcRequest {
89            id,
90            op: op_id,
91            input,
92            idem: None,
93        });
94        id
95    }
96
97    pub(crate) fn push_failed_frame(&mut self, error: CratestackError) -> u64 {
98        let id = self.next_id();
99        self.encode_errors.insert(id, error);
100        id
101    }
102
103    /// Fire the batch as a single `POST /rpc/batch` and return a
104    /// [`BatchResults`] keyed by handle for per-op output extraction.
105    ///
106    /// The outer `Result` only fails on transport / batch-envelope
107    /// errors (the whole batch couldn't be sent or the response
108    /// couldn't be parsed). Per-frame failures — both pre-send input
109    /// encoding errors and server-side `RpcErrorBody` — are deferred
110    /// to the matching `BatchResults::take(handle)?` call.
111    pub async fn send(self) -> Result<BatchResults, RpcClientError> {
112        let encode_errors = self.encode_errors;
113        let frames = if self.frames.is_empty() {
114            std::collections::HashMap::new()
115        } else {
116            let response_frames = self.rpc.batch(&self.frames).await?;
117            response_frames.into_iter().map(|f| (f.id, f)).collect()
118        };
119        Ok(BatchResults {
120            frames,
121            encode_errors,
122        })
123    }
124}
125
126/// Per-handle results from a sent batch. Each handle can be `take`n
127/// exactly once.
128pub struct BatchResults {
129    frames: std::collections::HashMap<u64, cratestack_core::rpc::RpcResponseFrame>,
130    encode_errors: std::collections::HashMap<u64, CratestackError>,
131}
132
133impl BatchResults {
134    /// Extract the typed output for one queued op. Returns:
135    ///
136    /// - `Ok(output)` — the server emitted an `output` for this frame
137    ///   and it decoded into `O`.
138    /// - `Err(RpcClientError::Codec(_))` — the input failed to encode
139    ///   before send, or the output failed to decode.
140    /// - `Err(RpcClientError::Remote(RpcRemoteError { body, .. }))` —
141    ///   the server emitted an `error` frame for this op. The
142    ///   `status` field is derived from the gRPC-style code in the
143    ///   body since `/rpc/batch` returns 200 at the HTTP level
144    ///   regardless of per-frame outcomes.
145    /// - `Err(RpcClientError::InvalidResponse(_))` — the server
146    ///   omitted this frame entirely or the frame had neither
147    ///   `output` nor `error` set.
148    pub fn take<O>(&mut self, handle: BatchHandle<O>) -> Result<O, RpcClientError>
149    where
150        O: serde::de::DeserializeOwned,
151    {
152        if let Some(error) = self.encode_errors.remove(&handle.id) {
153            return Err(RpcClientError::Codec(error));
154        }
155        let frame = self.frames.remove(&handle.id).ok_or_else(|| {
156            RpcClientError::InvalidResponse(format!(
157                "batch response missing frame for id {}",
158                handle.id,
159            ))
160        })?;
161        match (frame.output, frame.error) {
162            (Some(output), None) => serde_json::from_value::<O>(output).map_err(|error| {
163                RpcClientError::Codec(CratestackError::Codec(format!(
164                    "decode batch output for id {}: {error}",
165                    handle.id,
166                )))
167            }),
168            (None, Some(body)) => Err(RpcClientError::Remote(RpcRemoteError {
169                status: http_status_for_rpc_code(&body.code),
170                body,
171                challenges: Challenges::default(),
172            })),
173            (Some(_), Some(_)) | (None, None) => Err(RpcClientError::InvalidResponse(format!(
174                "batch frame {} has both `output` and `error` set, or neither",
175                handle.id,
176            ))),
177        }
178    }
179}
cratestack_client_rust/rpc/batch.rs

cratestack_client_rust/rpc/
batch.rs
