laser_wire/
fork.rs

1use crate::error::InvalidError;
2use crate::limits::MAX_FORK_ID_BYTES;
3use serde::{Deserialize, Serialize};
4use std::collections::BTreeMap;
5
6/// How a fork relates to the trunk it branched from.
7#[derive(
8    Clone,
9    Copy,
10    Debug,
11    Default,
12    PartialEq,
13    Eq,
14    Serialize,
15    Deserialize,
16    strum::Display,
17    strum::EnumString,
18    strum::VariantArray,
19)]
20#[serde(rename_all = "snake_case")]
21#[strum(serialize_all = "snake_case")]
22#[non_exhaustive]
23pub enum ForkKind {
24    /// Frozen snapshot: sees trunk rows only up to the offsets captured at
25    /// creation, plus the fork's own rows. Later trunk appends are invisible.
26    Severed,
27    /// Live branch: sees the trunk as it grows, plus the fork's own rows overlaid.
28    #[default]
29    Continuous,
30}
31
32/// Lifecycle of a fork.
33#[derive(
34    Clone,
35    Copy,
36    Debug,
37    Default,
38    PartialEq,
39    Eq,
40    Serialize,
41    Deserialize,
42    strum::Display,
43    strum::EnumString,
44    strum::VariantArray,
45)]
46#[serde(rename_all = "snake_case")]
47#[strum(serialize_all = "snake_case")]
48#[non_exhaustive]
49pub enum ForkStatus {
50    #[default]
51    Open,
52    Promoted,
53    Squashed,
54}
55
56/// A fork's metadata, returned by `create` and `list`.
57#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
58pub struct ForkInfo {
59    pub fork_id: String,
60    #[serde(default, skip_serializing_if = "Option::is_none")]
61    pub parent: Option<String>,
62    pub kind: ForkKind,
63    pub user_id: u32,
64    pub status: ForkStatus,
65    pub created_at_micros: u64,
66    pub row_count: u64,
67}
68
69/// Wire form of the `AGDX_FORK_CREATE` request. Wire-stability-bound, built
70/// through the SDK's fork handle in application code.
71#[derive(Clone, Debug, Serialize, Deserialize)]
72pub struct ForkCreate {
73    pub v: u32,
74    pub fork_id: String,
75    #[serde(default, skip_serializing_if = "Option::is_none")]
76    pub parent: Option<String>,
77    #[serde(default)]
78    pub kind: ForkKind,
79    #[serde(default, skip_serializing_if = "Vec::is_empty")]
80    pub tables: Vec<String>,
81    /// The trunk positions a severed fork is frozen at, stamped by the server
82    /// that logs the request so every replica folds the same fork. Clients
83    /// leave it unset.
84    #[serde(default, skip_serializing_if = "Option::is_none")]
85    pub points: Option<Vec<ForkPoint>>,
86}
87
88/// One trunk position a severed fork is frozen at: the last materialized
89/// offset of one partition of one table.
90#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
91pub struct ForkPoint {
92    pub table: String,
93    pub partition_id: u32,
94    pub offset: u64,
95    /// The creation time of the source topic incarnation the offset belongs to, set by the server that logs the create.
96    #[serde(default, skip_serializing_if = "Option::is_none")]
97    pub source_incarnation: Option<u64>,
98}
99
100/// Wire form of the `AGDX_FORK_DELETE` (squash) request.
101#[derive(Clone, Debug, Serialize, Deserialize)]
102pub struct ForkDelete {
103    pub v: u32,
104    pub fork_id: String,
105}
106
107/// Wire form of the `AGDX_FORK_PROMOTE` request.
108#[derive(Clone, Debug, Serialize, Deserialize)]
109pub struct ForkPromote {
110    pub v: u32,
111    pub fork_id: String,
112}
113
114/// Wire form of the `AGDX_FORK_LIST` request.
115#[derive(Clone, Debug, Serialize, Deserialize)]
116pub struct ForkList {
117    pub v: u32,
118    /// The stream whose resources the listing covers. Required in stream
119    /// tenancy, where the server authorizes the stream and lists only its
120    /// resources.
121    #[serde(default, skip_serializing_if = "Option::is_none")]
122    pub stream: Option<String>,
123}
124
125/// Wire form of the `AGDX_FORK_PUT` request. Wire-stability-bound, built through
126/// the SDK's fork handle in application code.
127#[derive(Clone, Debug, Serialize, Deserialize)]
128pub struct ForkPut {
129    pub v: u32,
130    pub fork_id: String,
131    pub table: String,
132    pub partition_id: u32,
133    pub offset: u64,
134    #[serde(default)]
135    pub projection_id: String,
136    #[serde(default)]
137    pub projection_version: u32,
138    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
139    pub fields: BTreeMap<String, String>,
140    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
141    pub metadata: BTreeMap<String, String>,
142    #[serde(
143        default,
144        skip_serializing_if = "Option::is_none",
145        with = "crate::encoding::opt_bin_bytes"
146    )]
147    pub payload: Option<Vec<u8>>,
148    #[serde(default, skip_serializing_if = "Option::is_none")]
149    pub embedding: Option<String>,
150    #[serde(default)]
151    pub tombstone: bool,
152}
153
154/// The result of a fork op: `Ok` with the outcome, or `Err` with a failure.
155#[derive(Clone, Debug, Serialize, Deserialize)]
156#[non_exhaustive]
157pub enum ForkReply {
158    Ok(ForkOutcome),
159    Err(ForkError),
160}
161
162/// The successful outcome of a fork command, shaped per op.
163#[derive(Clone, Debug, Serialize, Deserialize)]
164#[non_exhaustive]
165pub enum ForkOutcome {
166    Created(ForkInfo),
167    Deleted(bool),
168    Promoted { rows: usize },
169    List(Vec<ForkInfo>),
170    Written,
171}
172
173/// Why a fork operation failed.
174#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
175#[non_exhaustive]
176pub enum ForkError {
177    #[error("forks not supported: {0}")]
178    Unsupported(String),
179    #[error("fork not found: {0}")]
180    NotFound(String),
181    #[error("invalid fork: {0}")]
182    InvalidFork(String),
183    #[error("fork conflict: {0}")]
184    Conflict(String),
185    #[error("fork backend error: {0}")]
186    Backend(String),
187    /// The request could not be served right now and the identical request may
188    /// succeed later: the store was momentarily out of reach, a connection was
189    /// refused or dropped, or a concurrency conflict was aborted. Retry with
190    /// backoff rather than treating it as a fault.
191    #[error("temporarily unavailable: {0}")]
192    Unavailable(String),
193    #[error("unsupported fork op version (expected {expected}, got {got})")]
194    Version { expected: u32, got: u32 },
195    /// This managed backend does not own the mutation partition for the fork.
196    #[error("not the partition leader for this fork")]
197    NotLeader,
198}
199
200/// The canonical fork-id safelist, shared by every fork-serving backend. A
201/// fork id is caller-chosen and a backend that overlays a fork inlines it into
202/// a copy-on-write query as a quoted identifier, so the charset must be a
203/// strict safelist, not just a length bound: this is the one anti-injection
204/// rule. A valid id is non-empty, at most [`MAX_FORK_ID_BYTES`] bytes, and made
205/// only of ASCII letters, digits, `-`, `_`, and `.`. A stream-scoped id
206/// `stream:<stream>/<local>` is valid when the whole id fits the same bound and
207/// both parts pass the same charset rule. A backend that binds the id as a
208/// parameter may skip the call, but one that inlines it gets the defense for
209/// free by calling this before use.
210pub fn validate_fork_id(fork_id: &str) -> Result<(), InvalidError> {
211    let Some((stream, local)) = crate::authz::split_scoped_resource(fork_id) else {
212        return crate::validate::validate_safelisted_name("fork id", fork_id, MAX_FORK_ID_BYTES);
213    };
214    if fork_id.len() > MAX_FORK_ID_BYTES {
215        return Err(InvalidError::new(format!(
216            "fork id is {}B, exceeds cap {MAX_FORK_ID_BYTES}B",
217            fork_id.len()
218        )));
219    }
220    crate::validate::validate_safelisted_name("fork id stream", stream, MAX_FORK_ID_BYTES)?;
221    crate::validate::validate_safelisted_name("fork id", local, MAX_FORK_ID_BYTES)
222}
223
224#[cfg(all(test, feature = "cbor"))]
225mod tests {
226    use super::*;
227    use crate::codes::FORK_OP_VERSION;
228    use crate::framing::{decode_named, encode_named};
229
230    #[test]
231    fn given_fork_ids_when_validated_then_should_enforce_charset_and_length() {
232        assert!(validate_fork_id("experiment-2026-q2").is_ok());
233        assert!(validate_fork_id("run_7.v2").is_ok());
234        assert!(validate_fork_id("").is_err(), "empty");
235        assert!(validate_fork_id("bad id").is_err(), "space");
236        assert!(
237            validate_fork_id("o'brien; drop table").is_err(),
238            "sql metachars rejected"
239        );
240        assert!(
241            validate_fork_id("name/../etc").is_err(),
242            "slash and dot-dot"
243        );
244        assert!(validate_fork_id(&"f".repeat(MAX_FORK_ID_BYTES)).is_ok());
245        assert!(validate_fork_id(&"f".repeat(MAX_FORK_ID_BYTES + 1)).is_err());
246    }
247
248    #[test]
249    fn given_stream_scoped_fork_ids_when_validated_then_should_safelist_both_parts() {
250        assert!(validate_fork_id("stream:acme/experiment-2026-q2").is_ok());
251        assert!(
252            validate_fork_id("stream:acme/a/b").is_err(),
253            "slash in the local part"
254        );
255        assert!(
256            validate_fork_id("stream:ac me/run").is_err(),
257            "space in the stream"
258        );
259        assert!(
260            validate_fork_id("stream:acme/").is_err(),
261            "empty local part"
262        );
263        assert!(
264            validate_fork_id("stream:acme").is_err(),
265            "the stream resource itself"
266        );
267        let local = "f".repeat(MAX_FORK_ID_BYTES - "stream:acme/".len());
268        assert!(validate_fork_id(&format!("stream:acme/{local}")).is_ok());
269        assert!(validate_fork_id(&format!("stream:acme/{local}f")).is_err());
270    }
271
272    #[test]
273    fn given_fork_create_when_round_tripped_then_should_preserve_kind() {
274        let request = ForkCreate {
275            v: FORK_OP_VERSION,
276            fork_id: "agent-run-7".to_owned(),
277            parent: None,
278            kind: ForkKind::Severed,
279            tables: vec!["readings".to_owned()],
280            points: None,
281        };
282        let bytes = encode_named(&request).expect("serializes");
283        let back: ForkCreate = decode_named(&bytes).expect("deserializes");
284        assert_eq!(back.fork_id, "agent-run-7");
285        assert_eq!(back.kind, ForkKind::Severed);
286        assert_eq!(back.points, None);
287    }
288
289    #[test]
290    fn given_fork_create_with_points_when_round_tripped_then_should_preserve_them() {
291        let points = vec![
292            ForkPoint {
293                table: "readings".to_owned(),
294                partition_id: 2,
295                offset: 41,
296                source_incarnation: Some(1_700_000_000_000_000),
297            },
298            ForkPoint {
299                table: "readings".to_owned(),
300                partition_id: 3,
301                offset: 7,
302                source_incarnation: None,
303            },
304        ];
305        let request = ForkCreate {
306            v: FORK_OP_VERSION,
307            fork_id: "agent-run-8".to_owned(),
308            parent: None,
309            kind: ForkKind::Severed,
310            tables: Vec::new(),
311            points: Some(points.clone()),
312        };
313        let bytes = encode_named(&request).expect("serializes");
314        let back: ForkCreate = decode_named(&bytes).expect("deserializes");
315        assert_eq!(back.points, Some(points));
316    }
317
318    #[test]
319    fn given_fork_reply_created_when_round_tripped_then_should_preserve_info() {
320        let reply = ForkReply::Ok(ForkOutcome::Created(ForkInfo {
321            fork_id: "f1".to_owned(),
322            parent: None,
323            kind: ForkKind::Continuous,
324            user_id: 5,
325            status: ForkStatus::Open,
326            created_at_micros: 1,
327            row_count: 0,
328        }));
329        let bytes = encode_named(&reply).expect("serializes");
330        let back: ForkReply = decode_named(&bytes).expect("deserializes");
331        let ForkReply::Ok(ForkOutcome::Created(info)) = back else {
332            panic!("expected Created");
333        };
334        assert_eq!(info.user_id, 5);
335        assert_eq!(info.kind, ForkKind::Continuous);
336    }
337
338    #[test]
339    fn given_fork_kind_when_serialized_then_should_be_snake_case() {
340        // Must match LaserData Cloud's `rename_all = "snake_case"` on the wire.
341        assert_eq!(
342            serde_json::to_string(&ForkKind::Severed).expect("serializes"),
343            "\"severed\""
344        );
345        assert_eq!(
346            serde_json::to_string(&ForkKind::Continuous).expect("serializes"),
347            "\"continuous\""
348        );
349    }
350}
laser_wire/fork.rs

laser_wire/
fork.rs
