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: usize,
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}
82
83/// Wire form of the `AGDX_FORK_DELETE` (squash) request.
84#[derive(Clone, Debug, Serialize, Deserialize)]
85pub struct ForkDelete {
86    pub v: u32,
87    pub fork_id: String,
88}
89
90/// Wire form of the `AGDX_FORK_PROMOTE` request.
91#[derive(Clone, Debug, Serialize, Deserialize)]
92pub struct ForkPromote {
93    pub v: u32,
94    pub fork_id: String,
95}
96
97/// Wire form of the `AGDX_FORK_LIST` request.
98#[derive(Clone, Debug, Serialize, Deserialize)]
99pub struct ForkList {
100    pub v: u32,
101}
102
103/// Wire form of the `AGDX_FORK_PUT` request. Wire-stability-bound, built through
104/// the SDK's fork handle in application code.
105#[derive(Clone, Debug, Serialize, Deserialize)]
106pub struct ForkPut {
107    pub v: u32,
108    pub fork_id: String,
109    pub table: String,
110    pub partition_id: u32,
111    pub offset: u64,
112    #[serde(default)]
113    pub projection_id: String,
114    #[serde(default)]
115    pub projection_version: u32,
116    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
117    pub fields: BTreeMap<String, String>,
118    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
119    pub metadata: BTreeMap<String, String>,
120    #[serde(
121        default,
122        skip_serializing_if = "Option::is_none",
123        with = "crate::encoding::opt_bin_bytes"
124    )]
125    pub payload: Option<Vec<u8>>,
126    #[serde(default, skip_serializing_if = "Option::is_none")]
127    pub embedding: Option<String>,
128    #[serde(default)]
129    pub tombstone: bool,
130}
131
132/// The result of a fork op: `Ok` with the outcome, or `Err` with a failure.
133#[derive(Clone, Debug, Serialize, Deserialize)]
134#[non_exhaustive]
135pub enum ForkReply {
136    Ok(ForkOutcome),
137    Err(ForkError),
138}
139
140/// The successful outcome of a fork command, shaped per op.
141#[derive(Clone, Debug, Serialize, Deserialize)]
142#[non_exhaustive]
143pub enum ForkOutcome {
144    Created(ForkInfo),
145    Deleted(bool),
146    Promoted { rows: usize },
147    List(Vec<ForkInfo>),
148    Written,
149}
150
151/// Why a fork operation failed.
152#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
153#[non_exhaustive]
154pub enum ForkError {
155    #[error("forks not supported: {0}")]
156    Unsupported(String),
157    #[error("fork not found: {0}")]
158    NotFound(String),
159    #[error("invalid fork: {0}")]
160    InvalidFork(String),
161    #[error("fork conflict: {0}")]
162    Conflict(String),
163    #[error("fork backend error: {0}")]
164    Backend(String),
165    /// The request could not be served right now and the identical request may
166    /// succeed later: the store was momentarily out of reach, a connection was
167    /// refused or dropped, or a concurrency conflict was aborted. Retry with
168    /// backoff rather than treating it as a fault.
169    #[error("temporarily unavailable: {0}")]
170    Unavailable(String),
171    #[error("unsupported fork op version (expected {expected}, got {got})")]
172    Version { expected: u32, got: u32 },
173    /// This managed backend does not own the mutation partition for the fork.
174    #[error("not the partition leader for this fork")]
175    NotLeader,
176}
177
178/// The canonical fork-id safelist, shared by every fork-serving backend. A
179/// fork id is caller-chosen and a backend that overlays a fork inlines it into
180/// a copy-on-write query as a quoted identifier, so the charset must be a
181/// strict safelist, not just a length bound: this is the one anti-injection
182/// rule. A valid id is non-empty, at most [`MAX_FORK_ID_BYTES`] bytes, and made
183/// only of ASCII letters, digits, `-`, `_`, and `.`. A stream-scoped id
184/// `stream:<stream>/<local>` is valid when the whole id fits the same bound and
185/// both parts pass the same charset rule. A backend that binds the id as a
186/// parameter may skip the call, but one that inlines it gets the defense for
187/// free by calling this before use.
188pub fn validate_fork_id(fork_id: &str) -> Result<(), InvalidError> {
189    let Some((stream, local)) = crate::authz::split_scoped_resource(fork_id) else {
190        return crate::validate::validate_safelisted_name("fork id", fork_id, MAX_FORK_ID_BYTES);
191    };
192    if fork_id.len() > MAX_FORK_ID_BYTES {
193        return Err(InvalidError::new(format!(
194            "fork id is {}B, exceeds cap {MAX_FORK_ID_BYTES}B",
195            fork_id.len()
196        )));
197    }
198    crate::validate::validate_safelisted_name("fork id stream", stream, MAX_FORK_ID_BYTES)?;
199    crate::validate::validate_safelisted_name("fork id", local, MAX_FORK_ID_BYTES)
200}
201
202#[cfg(all(test, feature = "cbor"))]
203mod tests {
204    use super::*;
205    use crate::codes::FORK_OP_VERSION;
206    use crate::framing::{decode_named, encode_named};
207
208    #[test]
209    fn given_fork_ids_when_validated_then_should_enforce_charset_and_length() {
210        assert!(validate_fork_id("experiment-2026-q2").is_ok());
211        assert!(validate_fork_id("run_7.v2").is_ok());
212        assert!(validate_fork_id("").is_err(), "empty");
213        assert!(validate_fork_id("bad id").is_err(), "space");
214        assert!(
215            validate_fork_id("o'brien; drop table").is_err(),
216            "sql metachars rejected"
217        );
218        assert!(
219            validate_fork_id("name/../etc").is_err(),
220            "slash and dot-dot"
221        );
222        assert!(validate_fork_id(&"f".repeat(MAX_FORK_ID_BYTES)).is_ok());
223        assert!(validate_fork_id(&"f".repeat(MAX_FORK_ID_BYTES + 1)).is_err());
224    }
225
226    #[test]
227    fn given_stream_scoped_fork_ids_when_validated_then_should_safelist_both_parts() {
228        assert!(validate_fork_id("stream:acme/experiment-2026-q2").is_ok());
229        assert!(
230            validate_fork_id("stream:acme/a/b").is_err(),
231            "slash in the local part"
232        );
233        assert!(
234            validate_fork_id("stream:ac me/run").is_err(),
235            "space in the stream"
236        );
237        assert!(
238            validate_fork_id("stream:acme/").is_err(),
239            "empty local part"
240        );
241        assert!(
242            validate_fork_id("stream:acme").is_err(),
243            "the stream resource itself"
244        );
245        let local = "f".repeat(MAX_FORK_ID_BYTES - "stream:acme/".len());
246        assert!(validate_fork_id(&format!("stream:acme/{local}")).is_ok());
247        assert!(validate_fork_id(&format!("stream:acme/{local}f")).is_err());
248    }
249
250    #[test]
251    fn given_fork_create_when_round_tripped_then_should_preserve_kind() {
252        let request = ForkCreate {
253            v: FORK_OP_VERSION,
254            fork_id: "agent-run-7".to_owned(),
255            parent: None,
256            kind: ForkKind::Severed,
257            tables: vec!["readings".to_owned()],
258        };
259        let bytes = encode_named(&request).expect("serializes");
260        let back: ForkCreate = decode_named(&bytes).expect("deserializes");
261        assert_eq!(back.fork_id, "agent-run-7");
262        assert_eq!(back.kind, ForkKind::Severed);
263    }
264
265    #[test]
266    fn given_fork_reply_created_when_round_tripped_then_should_preserve_info() {
267        let reply = ForkReply::Ok(ForkOutcome::Created(ForkInfo {
268            fork_id: "f1".to_owned(),
269            parent: None,
270            kind: ForkKind::Continuous,
271            user_id: 5,
272            status: ForkStatus::Open,
273            created_at_micros: 1,
274            row_count: 0,
275        }));
276        let bytes = encode_named(&reply).expect("serializes");
277        let back: ForkReply = decode_named(&bytes).expect("deserializes");
278        let ForkReply::Ok(ForkOutcome::Created(info)) = back else {
279            panic!("expected Created");
280        };
281        assert_eq!(info.user_id, 5);
282        assert_eq!(info.kind, ForkKind::Continuous);
283    }
284
285    #[test]
286    fn given_fork_kind_when_serialized_then_should_be_snake_case() {
287        // Must match LaserData Cloud's `rename_all = "snake_case"` on the wire.
288        assert_eq!(
289            serde_json::to_string(&ForkKind::Severed).expect("serializes"),
290            "\"severed\""
291        );
292        assert_eq!(
293            serde_json::to_string(&ForkKind::Continuous).expect("serializes"),
294            "\"continuous\""
295        );
296    }
297}
laser_wire/fork.rs

laser_wire/
fork.rs
