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    #[error("unsupported fork op version (expected {expected}, got {got})")]
166    Version { expected: u32, got: u32 },
167    /// This plane does not own the mutation partition for the fork.
168    #[error("not the partition leader for this fork")]
169    NotLeader,
170}
171
172/// The canonical fork-id safelist, shared by every fork-serving backend. A
173/// fork id is caller-chosen and a backend that overlays a fork inlines it into
174/// a copy-on-write query as a quoted identifier, so the charset must be a
175/// strict safelist, not just a length bound: this is the one anti-injection
176/// rule. A valid id is non-empty, at most [`MAX_FORK_ID_BYTES`] bytes, and made
177/// only of ASCII letters, digits, `-`, `_`, and `.`. A backend that binds the
178/// id as a parameter may skip the call, but one that inlines it gets the defense for
179/// free by calling this before use.
180pub fn validate_fork_id(fork_id: &str) -> Result<(), InvalidError> {
181    crate::validate::validate_safelisted_name("fork id", fork_id, MAX_FORK_ID_BYTES)
182}
183
184#[cfg(all(test, feature = "cbor"))]
185mod tests {
186    use super::*;
187    use crate::codes::FORK_OP_VERSION;
188    use crate::framing::{decode_named, encode_named};
189
190    #[test]
191    fn given_fork_ids_when_validated_then_should_enforce_charset_and_length() {
192        assert!(validate_fork_id("experiment-2026-q2").is_ok());
193        assert!(validate_fork_id("run_7.v2").is_ok());
194        assert!(validate_fork_id("").is_err(), "empty");
195        assert!(validate_fork_id("bad id").is_err(), "space");
196        assert!(
197            validate_fork_id("o'brien; drop table").is_err(),
198            "sql metachars rejected"
199        );
200        assert!(
201            validate_fork_id("name/../etc").is_err(),
202            "slash and dot-dot"
203        );
204        assert!(validate_fork_id(&"f".repeat(MAX_FORK_ID_BYTES)).is_ok());
205        assert!(validate_fork_id(&"f".repeat(MAX_FORK_ID_BYTES + 1)).is_err());
206    }
207
208    #[test]
209    fn given_fork_create_when_round_tripped_then_should_preserve_kind() {
210        let request = ForkCreate {
211            v: FORK_OP_VERSION,
212            fork_id: "agent-run-7".to_owned(),
213            parent: None,
214            kind: ForkKind::Severed,
215            tables: vec!["orders".to_owned()],
216        };
217        let bytes = encode_named(&request).expect("serializes");
218        let back: ForkCreate = decode_named(&bytes).expect("deserializes");
219        assert_eq!(back.fork_id, "agent-run-7");
220        assert_eq!(back.kind, ForkKind::Severed);
221    }
222
223    #[test]
224    fn given_fork_reply_created_when_round_tripped_then_should_preserve_info() {
225        let reply = ForkReply::Ok(ForkOutcome::Created(ForkInfo {
226            fork_id: "f1".to_owned(),
227            parent: None,
228            kind: ForkKind::Continuous,
229            user_id: 5,
230            status: ForkStatus::Open,
231            created_at_micros: 1,
232            row_count: 0,
233        }));
234        let bytes = encode_named(&reply).expect("serializes");
235        let back: ForkReply = decode_named(&bytes).expect("deserializes");
236        let ForkReply::Ok(ForkOutcome::Created(info)) = back else {
237            panic!("expected Created");
238        };
239        assert_eq!(info.user_id, 5);
240        assert_eq!(info.kind, ForkKind::Continuous);
241    }
242
243    #[test]
244    fn given_fork_kind_when_serialized_then_should_be_snake_case() {
245        // Must match LaserData Cloud's `rename_all = "snake_case"` on the wire.
246        assert_eq!(
247            serde_json::to_string(&ForkKind::Severed).expect("serializes"),
248            "\"severed\""
249        );
250        assert_eq!(
251            serde_json::to_string(&ForkKind::Continuous).expect("serializes"),
252            "\"continuous\""
253        );
254    }
255}
laser_wire/fork.rs

laser_wire/
fork.rs
