waggle_core/
reconstruct.rs

1//! Reconstruct: the normative replay (design doc `04 §6`). Dedup by
2//! `(token, seq)` (R-3), order per-token by seq (the weakest ordering that
3//! keeps every query exact — doc `04 §2`), fold everything in one pass, and
4//! serialize deterministically (R-1).
5
6use std::collections::BTreeMap;
7
8use serde::Serialize;
9
10use crate::fold::{replay, FunnelFold, LineageFold, ManifestFold};
11use crate::log::LogRecord;
12use crate::manifest::AttributionManifest;
13use crate::slug::Stage;
14use crate::token::Token;
15
16/// The world at a log prefix: manifests, funnels, lineage. Serialization is
17/// deterministic (`BTreeMap` everywhere) — R-1's "byte-identical" is
18/// checked against the serialized form.
19#[derive(Debug, Default, Serialize)]
20pub struct WorldState {
21    /// Token → manifest state.
22    pub manifests: BTreeMap<Token, AttributionManifest>,
23    /// Token → stage → count.
24    pub funnels: BTreeMap<Token, BTreeMap<Stage, u64>>,
25    /// Parent → children (mint order).
26    pub lineage: BTreeMap<Token, Vec<Token>>,
27}
28
29/// Rebuild the world from records, in any arrival order, with duplicates.
30///
31/// Properties (tested in `tests/event_sourcing.rs`):
32/// - **R-1**: any arrival order respecting nothing at all — reconstruct
33///   sorts per-token by seq — yields byte-identical serialized state;
34/// - **R-3**: duplicates of `(token, seq, kind-discriminant)` are dropped;
35/// - **R-2**: reconstruct(prefix) + replay(suffix) ≡ reconstruct(all) —
36///   `WorldState` is a snapshot, and [`apply_suffix`] is the replay.
37#[must_use]
38pub fn reconstruct(records: impl IntoIterator<Item = LogRecord>) -> WorldState {
39    // Group per token, dedup by (seq, kind), order by seq. Kind joins the
40    // key because Minted is seq 0 by definition while an event could carry
41    // seq 0 from a permissive producer — never conflate birth with traffic.
42    let mut per_token: BTreeMap<Token, BTreeMap<(u32, u8), LogRecord>> = BTreeMap::new();
43    for rec in records {
44        let kind = match &rec {
45            LogRecord::Minted { .. } => 0u8,
46            LogRecord::Mutation { .. } => 1,
47            LogRecord::Event(_) => 2,
48        };
49        per_token
50            .entry(rec.token())
51            .or_default()
52            .entry((rec.seq().0, kind))
53            .or_insert(rec);
54    }
55
56    let ordered = per_token.into_values().flat_map(BTreeMap::into_values);
57    let (manifests, (funnels, lineage)) = replay(
58        ordered,
59        (
60            ManifestFold::default(),
61            (FunnelFold::default(), LineageFold::default()),
62        ),
63    );
64    WorldState {
65        manifests: manifests.manifests,
66        funnels: funnels.per_token,
67        lineage: lineage.children,
68    }
69}
70
71/// Continue a snapshot with later records (R-2's replay half). The suffix
72/// must already be per-token seq-ordered relative to the snapshot — stores
73/// guarantee this by construction (C-3).
74#[must_use]
75pub fn apply_suffix(
76    snapshot: WorldState,
77    suffix: impl IntoIterator<Item = LogRecord>,
78) -> WorldState {
79    let mut manifests = ManifestFold {
80        manifests: snapshot.manifests,
81    };
82    let mut funnels = FunnelFold {
83        per_token: snapshot.funnels,
84    };
85    let mut lineage = LineageFold {
86        children: snapshot.lineage,
87    };
88    for rec in suffix {
89        use crate::fold::Fold as _;
90        manifests.apply(&rec);
91        funnels.apply(&rec);
92        lineage.apply(&rec);
93    }
94    WorldState {
95        manifests: manifests.manifests,
96        funnels: funnels.per_token,
97        lineage: lineage.children,
98    }
99}
100
101#[cfg(test)]
102mod tests {
103    use super::*;
104    use crate::event::{ActorClass, Event, Seq};
105    use crate::{CanonicalUrl, Channel, MintOptions, MintSpec, ResolverContext, Sharer, Timestamp};
106
107    #[test]
108    fn seq_zero_event_never_conflates_with_minted() {
109        let mut entropy = |buf: &mut [u8]| {
110            buf.fill(9);
111            Ok(())
112        };
113        let m = crate::mint(
114            MintSpec::new(
115                CanonicalUrl::new("ws://a/b").unwrap(),
116                Sharer::new("lead").unwrap(),
117                Channel::subagent_general(),
118            ),
119            &MintOptions::default(),
120            &mut entropy,
121            Timestamp::from_unix_ms(1),
122        )
123        .unwrap();
124        let token = m.token;
125        let records = vec![
126            LogRecord::Event(Event {
127                token,
128                stage: crate::Stage::impression(),
129                actor: ActorClass::from_context(&ResolverContext::human()),
130                at: Timestamp::from_unix_ms(2),
131                seq: Seq(0), // permissive producer
132                variant: None,
133                regions: None,
134            }),
135            LogRecord::Minted {
136                manifest: Box::new(m),
137            },
138        ];
139        let world = reconstruct(records);
140        assert!(world.manifests.contains_key(&token));
141        assert_eq!(world.funnels[&token][&crate::Stage::impression()], 1);
142    }
143}
waggle_core/reconstruct.rs

waggle_core/
reconstruct.rs
