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

laser_wire/
fork.rs
