gonzalo_core/
record.rs

1//! The universal persisted unit and its classification.
2
3use crate::{ContentHash, Identity, RecordKey, Revision};
4use serde::{Deserialize, Serialize};
5use std::collections::BTreeMap;
6
7/// What a record represents. Drives the merge strategy.
8#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
9pub enum RecordKind {
10    MemoryTier,
11    Topic,
12    Session,
13    Checkpoint,
14    /// A tracked work item imported from an external ticket platform.
15    Ticket,
16    /// An append-only comment/event on a ticket.
17    TicketEvent,
18    /// A per-view code-graph manifest: `(repo, view_id) -> { path -> content_hash }`.
19    /// Regenerable from source; reconciled last-writer-wins. See ADR 0012.
20    GraphManifest,
21    /// The manifest of a durable vector index: which shards exist, which blob
22    /// holds each, and the embedding space they are in. Unlike
23    /// [`GraphManifest`](Self::GraphManifest) it is **not** regenerable — with
24    /// caller-supplied embeddings gonzalo never sees the model — so it merges
25    /// [`Opaque`](MergeClass::Opaque). See ADR 0027.
26    VectorManifest,
27    /// A human with fleet roles. Not ADR 0015's `Principal`, which is a gonzalo
28    /// bearer token. See ADR 0022.
29    Person,
30    /// Binds one external account (chat user id, OIDC subject) to a `Person`.
31    /// See ADR 0022.
32    IdentityBinding,
33    /// A person's fleet role, fleet-wide or for one workspace. See ADR 0022,
34    /// amended by ADR 0023.
35    RoleGrant,
36    /// A chat channel's configuration: what it follows, a notify preset, and
37    /// a role ceiling. See ADR 0022, amended by ADR 0023.
38    ChannelConfig,
39    /// A one-time, expiring token that links an account to a person. Stores only
40    /// the token's hash, and is marked consumed rather than deleted. See ADR 0022.
41    LinkToken,
42    /// A write-once record of who did what, to what, and with what result.
43    /// See ADR 0022.
44    AuditEntry,
45    /// A deletion marker. Hidden from consumer reads (`get`/`list`); replicated
46    /// by sync and pull through raw reads; physically removed only by
47    /// `Store::purge`. See ADR 0021.
48    Tombstone,
49}
50
51/// How concurrent edits to a record of a given kind are reconciled.
52#[derive(Clone, Copy, Debug, PartialEq, Eq)]
53pub enum MergeClass {
54    /// Edits union/concatenate (auto-memory topics, session transcripts).
55    AppendOnly,
56    /// Field-level 3-way merge against the common base.
57    Structured,
58    /// No safe automatic merge; surface to the caller.
59    Opaque,
60    /// Regenerable / don't-merge (e.g. per-view code-graph manifests, ADR 0012).
61    /// The body can be re-derived from source, and views are single-writer, so a
62    /// divergence is rare and reconciled deterministically in favor of side A
63    /// (the `ours` argument to `merge`, which has no `Meta` to compare) rather
64    /// than a content merge — never a surfaced conflict.
65    Derived,
66}
67
68impl RecordKind {
69    pub fn merge_class(self) -> MergeClass {
70        match self {
71            RecordKind::Topic | RecordKind::Session | RecordKind::TicketEvent => {
72                MergeClass::AppendOnly
73            }
74            RecordKind::MemoryTier
75            | RecordKind::Ticket
76            | RecordKind::Person
77            | RecordKind::IdentityBinding
78            | RecordKind::RoleGrant
79            | RecordKind::ChannelConfig => MergeClass::Structured,
80            RecordKind::Checkpoint => MergeClass::Opaque,
81            // Both diverge only through a double redemption or an audit-key
82            // collision, which must surface rather than merge (ADR 0022).
83            RecordKind::LinkToken | RecordKind::AuditEntry => MergeClass::Opaque,
84            RecordKind::GraphManifest => MergeClass::Derived,
85            // Not regenerable: a lost side is lost vectors, so surface the
86            // divergence rather than resolving it silently.
87            RecordKind::VectorManifest => MergeClass::Opaque,
88            // Sync and pull reconcile tombstones before any body merge runs;
89            // the most conservative class guards a path that forgets to.
90            RecordKind::Tombstone => MergeClass::Opaque,
91        }
92    }
93}
94
95/// A record body. `Inline` stores bytes directly in the record; `Blob`
96/// references content held out-of-line in a content-addressed [`BlobStore`],
97/// so byte-identical bodies (e.g. code-graph slices shared across worktrees)
98/// are stored once. See ADR 0012.
99///
100/// [`BlobStore`]: crate::store::BlobStore
101#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
102pub enum Body {
103    Inline(Vec<u8>),
104    /// Content stored out-of-line under `hash` in a [`BlobStore`]; `len` is the
105    /// referenced content's byte length. The record itself carries only the
106    /// reference — the bytes are fetched via `BlobStore::get_blob`.
107    ///
108    /// [`BlobStore`]: crate::store::BlobStore
109    Blob {
110        hash: ContentHash,
111        len: u64,
112    },
113}
114
115impl Body {
116    /// Build a blob body referencing `content` by its content hash. The content
117    /// itself is written separately via `BlobStore::put_blob`.
118    pub fn blob(content: &[u8]) -> Self {
119        Body::Blob {
120            hash: ContentHash::of(content),
121            len: content.len() as u64,
122        }
123    }
124
125    /// The bytes used for content hashing and merging. For a `Blob` these are
126    /// the reference's hash bytes, not the referenced content — identical
127    /// content yields an identical reference, so the record's revision is
128    /// stable under content-addressed dedup.
129    pub fn bytes(&self) -> &[u8] {
130        match self {
131            Body::Inline(b) => b,
132            Body::Blob { hash, .. } => hash.0.as_bytes(),
133        }
134    }
135}
136
137/// Provenance and labels for a record.
138#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
139pub struct Meta {
140    pub author: Identity,
141    pub origin_system: String,
142    /// When this record was first written, in ms since the Unix epoch. Stamped
143    /// by the store and carried across every edit; a recreation over a
144    /// tombstone starts a new one, because that is a new record at an old key
145    /// (gonzalo#293).
146    pub created: i64,
147    /// When this record was last written, in ms since the Unix epoch. Stamped
148    /// by the store, including on a delete, where it equals the tombstone's
149    /// `deleted_at`. A replication write keeps the source's times rather than
150    /// restamping, so syncing is not an edit.
151    pub updated: i64,
152    pub labels: BTreeMap<String, String>,
153}
154
155/// The universal persisted unit.
156#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
157pub struct Record {
158    pub key: RecordKey,
159    pub kind: RecordKind,
160    pub revision: Revision,
161    pub parent: Option<Revision>,
162    pub body: Body,
163    pub meta: Meta,
164    pub links: Vec<RecordKey>,
165    /// Recent revisions this record descends from, newest first, bounded by the
166    /// store's ancestor cap. Advisory: a missing or truncated list makes sync
167    /// report a conflict instead of guessing. Empty on legacy records.
168    #[serde(default, skip_serializing_if = "Vec::is_empty")]
169    pub ancestors: Vec<Revision>,
170    /// Tombstones only: when the delete happened, in ms since the Unix epoch.
171    /// Stamped by the store. `None` means collection never purges it.
172    #[serde(default, skip_serializing_if = "Option::is_none")]
173    pub deleted_at: Option<i64>,
174    /// Tombstones only: the blob the deleted record's body referenced, if it
175    /// had one. A tombstone's own body is empty, so without this the hash
176    /// would be lost the moment the record was deleted and blob collection
177    /// could not tell a blob that is merely deleted from one nothing points at.
178    ///
179    /// **A tombstone pins its blob**: the blob survives until the tombstone
180    /// itself is collected, so a record restored from a peer, or recreated
181    /// from the same content, still has its bytes (gonzalo#292).
182    #[serde(default, skip_serializing_if = "Option::is_none")]
183    pub deleted_blob: Option<ContentHash>,
184    /// Tombstones only: the kind of the record that was deleted.
185    ///
186    /// Set when the deleted body is **retained** — the manifest kinds, whose
187    /// bodies name blobs out of line — so GC knows which parser to use for the
188    /// references inside, and `undelete` knows what kind to recreate. A
189    /// manifest's body is the only record of which blob was which shard, so
190    /// discarding it would leave the pinned bytes unnameable (ADR 0029).
191    ///
192    /// `None` on a tombstone whose body was discarded, and on every tombstone
193    /// written before this field existed.
194    #[serde(default, skip_serializing_if = "Option::is_none")]
195    pub deleted_kind: Option<RecordKind>,
196}
197
198impl Meta {
199    /// Provenance for a write by `author` from `origin_system`, with no labels.
200    ///
201    /// `created` and `updated` are left at `0` and the store stamps them on
202    /// write (gonzalo#293), so there is nothing useful to pass here. Set the
203    /// fields directly for labels.
204    pub fn new(author: Identity, origin_system: impl Into<String>) -> Self {
205        Self {
206            author,
207            origin_system: origin_system.into(),
208            created: 0,
209            updated: 0,
210            labels: BTreeMap::new(),
211        }
212    }
213}
214
215impl Record {
216    /// A record that does not exist yet: the body's first revision, no parent
217    /// and no history. Write it with `put(record, None)` (gonzalo#305).
218    ///
219    /// ```
220    /// use gonzalo_core::{Body, Identity, Meta, Record, RecordKey, RecordKind, Revision};
221    ///
222    /// let key = RecordKey::new("caliban", "topics", "rust");
223    /// let meta = Meta::new(Identity::new("ada"), "example");
224    /// let v0 = Record::create(key, RecordKind::Topic, Body::Inline(b"first".to_vec()), meta);
225    /// assert_eq!(v0.revision, Revision::initial(b"first"));
226    /// assert_eq!(v0.parent, None);
227    ///
228    /// // The next version descends from it: `put(v1, Some(v0.revision))` commits.
229    /// let meta = Meta::new(Identity::new("ada"), "example");
230    /// let v1 = v0.update(Body::Inline(b"second".to_vec()), meta);
231    /// assert_eq!(v1.revision, v0.revision.next(b"second"));
232    /// assert_eq!(v1.parent, Some(v0.revision));
233    /// ```
234    pub fn create(key: RecordKey, kind: RecordKind, body: Body, meta: Meta) -> Self {
235        Self {
236            revision: Revision::initial(body.bytes()),
237            key,
238            kind,
239            parent: None,
240            body,
241            meta,
242            links: Vec::new(),
243            ancestors: Vec::new(),
244            deleted_at: None,
245            deleted_blob: None,
246            deleted_kind: None,
247        }
248    }
249
250    /// The next version of this record carrying `body`, written by `meta`.
251    ///
252    /// Its revision follows this one and its `parent` is this revision, so
253    /// `put(next, Some(current.revision))` commits it and a stale read
254    /// conflicts — which is the pairing that is easy to get wrong by hand
255    /// (gonzalo#305). `key`, `kind` and `links` carry over.
256    ///
257    /// `ancestors` is left empty: the store folds this record's revision and
258    /// history in when it commits, bounded by its ancestor cap. `deleted_at`
259    /// is cleared, since only a store writes a tombstone.
260    pub fn update(&self, body: Body, meta: Meta) -> Self {
261        Self {
262            key: self.key.clone(),
263            kind: self.kind,
264            revision: self.revision.next(body.bytes()),
265            parent: Some(self.revision.clone()),
266            body,
267            meta,
268            links: self.links.clone(),
269            ancestors: Vec::new(),
270            deleted_at: None,
271            deleted_blob: None,
272            deleted_kind: None,
273        }
274    }
275
276    /// Whether this record is a deletion marker.
277    pub fn is_tombstone(&self) -> bool {
278        self.kind == RecordKind::Tombstone
279    }
280}
281
282#[cfg(test)]
283mod tests {
284    use super::*;
285
286    // ---- builders (#305) ----
287
288    fn meta_by(author: &str) -> Meta {
289        Meta::new(Identity::new(author), "test")
290    }
291
292    #[test]
293    fn meta_new_carries_author_and_origin_and_nothing_else() {
294        let m = meta_by("ada");
295        assert_eq!(m.author, Identity::new("ada"));
296        assert_eq!(m.origin_system, "test");
297        assert_eq!((m.created, m.updated), (0, 0));
298        assert!(m.labels.is_empty());
299    }
300
301    #[test]
302    fn create_is_a_first_revision_with_no_history() {
303        let key = RecordKey::new("ns", "col", "id");
304        let body = Body::Inline(b"hello".to_vec());
305        let r = Record::create(key.clone(), RecordKind::Topic, body.clone(), meta_by("ada"));
306
307        assert_eq!(r.key, key);
308        assert_eq!(r.kind, RecordKind::Topic);
309        assert_eq!(r.revision, Revision::initial(b"hello"));
310        assert_eq!(r.parent, None);
311        assert_eq!(r.body, body);
312        assert!(r.links.is_empty() && r.ancestors.is_empty());
313        assert_eq!(r.deleted_at, None);
314    }
315
316    #[test]
317    fn update_follows_the_revision_it_was_built_from() {
318        let key = RecordKey::new("ns", "col", "id");
319        let mut v0 = Record::create(
320            key.clone(),
321            RecordKind::Topic,
322            Body::Inline(b"v0".to_vec()),
323            meta_by("ada"),
324        );
325        v0.links = vec![RecordKey::new("ns", "col", "other")];
326
327        let v1 = v0.update(Body::Inline(b"v1".to_vec()), meta_by("grace"));
328
329        assert_eq!(v1.revision, v0.revision.next(b"v1"));
330        assert_eq!(v1.parent, Some(v0.revision.clone()));
331        assert_eq!(v1.body, Body::Inline(b"v1".to_vec()));
332        assert_eq!(
333            v1.meta.author,
334            Identity::new("grace"),
335            "the updater writes it"
336        );
337        assert_eq!((v1.key, v1.kind), (key, RecordKind::Topic));
338        assert_eq!(v1.links, v0.links, "links carry over");
339        assert!(
340            v1.ancestors.is_empty(),
341            "the store folds history in on commit"
342        );
343    }
344
345    #[tokio::test]
346    async fn built_records_commit_through_occ() {
347        use crate::{PutResult, Store, memstore::MemStore};
348
349        let store = MemStore::new();
350        let key = RecordKey::new("ns", "col", "id");
351        let v0 = Record::create(
352            key.clone(),
353            RecordKind::Topic,
354            Body::Inline(b"v0".to_vec()),
355            meta_by("ada"),
356        );
357        assert!(matches!(
358            store.put(v0.clone(), None).await.unwrap(),
359            PutResult::Committed(_)
360        ));
361
362        // Update what was read back, the way a consumer does.
363        let current = store.get(&key).await.unwrap().unwrap();
364        let v1 = current.update(Body::Inline(b"v1".to_vec()), meta_by("ada"));
365        assert!(matches!(
366            store.put(v1, Some(current.revision.clone())).await.unwrap(),
367            PutResult::Committed(_)
368        ));
369
370        // A second update built from the now-stale read conflicts, as OCC should.
371        let stale = current.update(Body::Inline(b"v1b".to_vec()), meta_by("ada"));
372        assert!(matches!(
373            store
374                .put(stale, Some(current.revision.clone()))
375                .await
376                .unwrap(),
377            PutResult::Conflict(_)
378        ));
379    }
380
381    #[test]
382    fn merge_class_is_assigned_per_kind() {
383        assert_eq!(RecordKind::Topic.merge_class(), MergeClass::AppendOnly);
384        assert_eq!(RecordKind::Session.merge_class(), MergeClass::AppendOnly);
385        assert_eq!(RecordKind::MemoryTier.merge_class(), MergeClass::Structured);
386        assert_eq!(RecordKind::Checkpoint.merge_class(), MergeClass::Opaque);
387        assert_eq!(RecordKind::Ticket.merge_class(), MergeClass::Structured);
388        assert_eq!(
389            RecordKind::TicketEvent.merge_class(),
390            MergeClass::AppendOnly
391        );
392        assert_eq!(RecordKind::GraphManifest.merge_class(), MergeClass::Derived);
393        assert_eq!(RecordKind::Person.merge_class(), MergeClass::Structured);
394        assert_eq!(
395            RecordKind::IdentityBinding.merge_class(),
396            MergeClass::Structured
397        );
398        assert_eq!(RecordKind::RoleGrant.merge_class(), MergeClass::Structured);
399        assert_eq!(
400            RecordKind::ChannelConfig.merge_class(),
401            MergeClass::Structured
402        );
403        assert_eq!(RecordKind::LinkToken.merge_class(), MergeClass::Opaque);
404        assert_eq!(RecordKind::AuditEntry.merge_class(), MergeClass::Opaque);
405    }
406
407    #[test]
408    fn fleet_kinds_serialize_as_their_names() {
409        for (kind, name) in [
410            (RecordKind::Person, "\"Person\""),
411            (RecordKind::IdentityBinding, "\"IdentityBinding\""),
412            (RecordKind::RoleGrant, "\"RoleGrant\""),
413            (RecordKind::ChannelConfig, "\"ChannelConfig\""),
414            (RecordKind::LinkToken, "\"LinkToken\""),
415            (RecordKind::AuditEntry, "\"AuditEntry\""),
416        ] {
417            assert_eq!(serde_json::to_string(&kind).unwrap(), name);
418            let back: RecordKind = serde_json::from_str(name).unwrap();
419            assert_eq!(back, kind);
420        }
421    }
422
423    #[test]
424    fn body_exposes_bytes() {
425        assert_eq!(Body::Inline(b"hi".to_vec()).bytes(), b"hi");
426    }
427
428    #[test]
429    fn blob_body_references_content_by_hash() {
430        let body = Body::blob(b"fn main() {}");
431        match &body {
432            Body::Blob { hash, len } => {
433                assert_eq!(*hash, crate::ContentHash::of(b"fn main() {}"));
434                assert_eq!(*len, 12);
435            }
436            _ => panic!("expected Body::Blob"),
437        }
438    }
439
440    #[test]
441    fn blob_body_bytes_are_stable_per_content() {
442        // Identical content -> identical body bytes -> identical revision (the
443        // record-level face of content-addressed dedup).
444        assert_eq!(Body::blob(b"same").bytes(), Body::blob(b"same").bytes());
445        assert_ne!(Body::blob(b"same").bytes(), Body::blob(b"diff").bytes());
446    }
447
448    fn plain_record() -> Record {
449        let body = Body::Inline(b"hello".to_vec());
450        Record {
451            key: RecordKey::new("ns", "col", "id"),
452            kind: RecordKind::Topic,
453            revision: Revision::initial(body.bytes()),
454            parent: None,
455            body,
456            meta: Meta {
457                author: Identity::new("t"),
458                origin_system: "test".into(),
459                created: 0,
460                updated: 0,
461                labels: BTreeMap::new(),
462            },
463            links: Vec::new(),
464            ancestors: Vec::new(),
465            deleted_at: None,
466            deleted_blob: None,
467            deleted_kind: None,
468        }
469    }
470
471    #[test]
472    fn tombstone_kind_is_opaque_and_detected() {
473        assert_eq!(RecordKind::Tombstone.merge_class(), MergeClass::Opaque);
474        let mut r = plain_record();
475        assert!(!r.is_tombstone());
476        r.kind = RecordKind::Tombstone;
477        assert!(r.is_tombstone());
478    }
479
480    #[test]
481    fn tombstone_kind_serializes_as_its_name() {
482        assert_eq!(
483            serde_json::to_string(&RecordKind::Tombstone).unwrap(),
484            "\"Tombstone\""
485        );
486    }
487
488    #[test]
489    fn empty_new_fields_are_omitted_from_json() {
490        let json = serde_json::to_value(plain_record()).unwrap();
491        let obj = json.as_object().unwrap();
492        assert!(!obj.contains_key("ancestors"));
493        assert!(!obj.contains_key("deleted_at"));
494    }
495
496    #[test]
497    fn legacy_json_without_new_fields_deserializes() {
498        let mut json = serde_json::to_value(plain_record()).unwrap();
499        let obj = json.as_object_mut().unwrap();
500        obj.remove("ancestors");
501        obj.remove("deleted_at");
502        let back: Record = serde_json::from_value(json).unwrap();
503        assert_eq!(back, plain_record());
504    }
505
506    #[test]
507    fn populated_new_fields_roundtrip() {
508        let mut r = plain_record();
509        r.kind = RecordKind::Tombstone;
510        r.ancestors = vec![Revision::initial(b"a"), Revision::initial(b"b")];
511        r.deleted_at = Some(1_700_000_000_000);
512        let back: Record = serde_json::from_str(&serde_json::to_string(&r).unwrap()).unwrap();
513        assert_eq!(back, r);
514    }
515}
gonzalo_core/record.rs

gonzalo_core/
record.rs
