prikk_store/lifecycle_cache/
incremental.rs

1//! Incremental baseline lifecycle-state cache (DC-64).
2//!
3//! Accelerates the commit path's baseline reconstruction by applying only the *newest* block's
4//! patch operations to a persisted predecessor state, instead of replaying a lineage's entire
5//! history on every commit. Scoped to the commit path only — see the design document at
6//! `rfcs/handoffs/DC-64-baseline-reconstruction-cost/incremental-baseline-cache-design-v1.md`,
7//! required reading before changing anything here, including why this is safe despite the trust
8//! ladder in `cache_ladder.rs` requiring full-replay certification for a superficially similar
9//! problem (that ladder guards a different, merge-only decision this cache never makes).
10//!
11//! Rebuildable and never authoritative (NFR-PERF-04): any problem loading the persisted cache — a
12//! missing file, a checksum mismatch, a decode failure — is treated as an absent cache, never a hard
13//! error, and always falls through to the unmodified `replay_derived_state` full-replay path.
14
15use prikk_error::Result;
16use prikk_object::{CanonicalWriter, ObjectId, WireType};
17
18use crate::byte_cursor::ByteCursor;
19use crate::fsutil::{read_file_if_exists, write_file_atomically};
20use crate::layout::RepositoryLayout;
21use crate::node_lifecycle::{LiveNode, NodeContent, NodeLifecycleState, Tombstone};
22use crate::object_store::ObjectReader;
23use crate::path::RepoPath;
24
25use super::{ReplayDerivedLifecycleState, replay, replay_derived_state};
26
27const CACHE_FILE_NAME: &str = "lifecycle-state.v1";
28const CACHE_MAGIC: &[u8] = b"PRIKK-LIFECYCLE-INCREMENTAL-CACHE-v1\0";
29const CACHE_SCHEMA_VERSION: u32 = 1;
30
31/// After this many consecutive incremental steps on one lineage, the next commit is forced through
32/// an unmodified full replay regardless of cache eligibility. This is the only control on how long a
33/// persistence fault that survives the checksum and `from_replay`'s structural check could live
34/// before an independent reconstruction overwrites the cache with ground truth. See the design
35/// document §5 for the exposure/amortized-overhead reasoning behind this exact value.
36const REANCHOR_BOUND: u32 = 64;
37
38struct IncrementalCache {
39    baseline_block_id: ObjectId,
40    horizon_id: ObjectId,
41    steps_since_reanchor: u32,
42    state: NodeLifecycleState,
43}
44
45/// One disagreement between the persisted incremental cache and an independent full replay of the
46/// block it currently claims to represent — the persistence-fault case the checksum and
47/// `from_replay`'s structural check do not, by themselves, catch (design document §6).
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub struct LifecycleCacheDivergence {
50    /// The block the cache claimed to represent when the divergence was found.
51    pub baseline_block_id: ObjectId,
52    /// Human-readable detail: either a content disagreement or that independent verification
53    /// itself could not complete (e.g. the claimed block became unreadable).
54    pub detail: String,
55}
56
57/// Check the persisted cache, if any, against an independent full replay of the block it claims to
58/// represent. Never errors: a replay failure is itself reported as a divergence rather than
59/// propagated, so `verify` always completes with a full picture rather than aborting on this check.
60/// This answers a narrower question than "is the cache eligible for the next commit" — it verifies
61/// only what the cache is *currently* offering, on the DC-56 `verify_divergence` pattern.
62pub(crate) fn verify_divergence(
63    reader: &impl ObjectReader,
64    layout: &RepositoryLayout,
65) -> Vec<LifecycleCacheDivergence> {
66    let Some(cached) = load(layout) else {
67        return Vec::new();
68    };
69    match replay_derived_state(reader, cached.baseline_block_id, cached.horizon_id) {
70        Ok(replayed) if replayed.state() == &cached.state => Vec::new(),
71        Ok(_) => vec![LifecycleCacheDivergence {
72            baseline_block_id: cached.baseline_block_id,
73            detail: "cached lifecycle state disagrees with an independent full replay".to_string(),
74        }],
75        Err(err) => vec![LifecycleCacheDivergence {
76            baseline_block_id: cached.baseline_block_id,
77            detail: format!("cached lifecycle state could not be independently verified: {err}"),
78        }],
79    }
80}
81
82/// Resolve the baseline lifecycle state for `baseline_block_id`/`horizon_id`, using an incremental
83/// step from a cached predecessor when eligible and falling back to an unmodified full replay
84/// otherwise. This is the sole entry point `node_authoring.rs` calls in place of
85/// `replay_derived_state` directly; the return contract is identical.
86pub(crate) fn resolve_baseline_state(
87    layout: &RepositoryLayout,
88    reader: &impl ObjectReader,
89    baseline_block_id: ObjectId,
90    horizon_id: ObjectId,
91) -> Result<ReplayDerivedLifecycleState> {
92    if let Some(cached) = load(layout) {
93        if cached.horizon_id == horizon_id && cached.steps_since_reanchor < REANCHOR_BOUND {
94            if let Some(state) = try_incremental_step(reader, &cached, baseline_block_id)? {
95                let result = ReplayDerivedLifecycleState::from_replay(baseline_block_id, state)?;
96                persist(
97                    layout,
98                    baseline_block_id,
99                    horizon_id,
100                    cached.steps_since_reanchor + 1,
101                    result.state(),
102                );
103                return Ok(result);
104            }
105        }
106    }
107    let result = replay_derived_state(reader, baseline_block_id, horizon_id)?;
108    persist(layout, baseline_block_id, horizon_id, 0, result.state());
109    Ok(result)
110}
111
112/// Attempt the incremental step. `Ok(None)` means "not eligible" — parent mismatch, multi-parent, the
113/// new block could not be read (the full-replay path re-derives it with its own proper error class),
114/// or — DC-65 — the block's operations need a `TextFile` node's materialized content that this
115/// step's fresh, single-block text cache cannot supply. That last case is not a cache-trust failure:
116/// applying one block's operations against a cached predecessor uses the identical
117/// `apply_state_effect` fold full replay uses, but a fold that spans only one block cannot
118/// materialize a node whose current content is itself an *earlier*, already-cached-away block's
119/// `EditText` result — full replay's `TextCache` accumulates across the whole lineage and never has
120/// this gap. Structurally falling back to full replay for this one commit is the correct, general
121/// fix (rather than a narrower per-node fallback), consistent with the DC-65 invariant that any
122/// consumer needing a `TextFile` node's actual bytes must be able to materialize them, never assume
123/// a stored object. See the design document §9a. A genuine application failure of any other class
124/// propagates as `Err`, still not folded into the fallback path — see §3.
125fn try_incremental_step(
126    reader: &impl ObjectReader,
127    cached: &IncrementalCache,
128    baseline_block_id: ObjectId,
129) -> Result<Option<NodeLifecycleState>> {
130    let Ok(block) = replay::read_block(reader, baseline_block_id) else {
131        return Ok(None);
132    };
133    if block.parent_block_ids.as_slice() != [cached.baseline_block_id] {
134        return Ok(None);
135    }
136    let mut state = cached.state.clone();
137    match replay::apply_one_block(reader, &block, &mut state, false) {
138        Ok(()) => Ok(Some(state)),
139        Err(replay::LifecycleReplayError::MissingBlobForLifecycleEffect { .. }) => Ok(None),
140        Err(other) => Err(other.into()),
141    }
142}
143
144/// Persist the refreshed cache. Best-effort: a save failure does not fail the commit that just
145/// succeeded — losing the opportunity to accelerate the *next* commit is a performance regression,
146/// not a correctness one, and the cache is rebuildable by design.
147fn persist(
148    layout: &RepositoryLayout,
149    baseline_block_id: ObjectId,
150    horizon_id: ObjectId,
151    steps_since_reanchor: u32,
152    state: &NodeLifecycleState,
153) {
154    let cache = IncrementalCache {
155        baseline_block_id,
156        horizon_id,
157        steps_since_reanchor,
158        state: state.clone(),
159    };
160    let _ = save(layout, &cache);
161}
162
163fn cache_path(layout: &RepositoryLayout) -> std::path::PathBuf {
164    layout.cache_dir().join(CACHE_FILE_NAME)
165}
166
167fn load(layout: &RepositoryLayout) -> Option<IncrementalCache> {
168    let relative = layout.repository_relative(&cache_path(layout)).ok()?;
169    let bytes = read_file_if_exists(layout.repository_mutation_root(), &relative).ok()??;
170    decode(&bytes)
171}
172
173fn save(layout: &RepositoryLayout, cache: &IncrementalCache) -> Result<()> {
174    let relative = layout.repository_relative(&cache_path(layout))?;
175    write_file_atomically(layout.repository_mutation_root(), &relative, &encode(cache))
176}
177
178fn encode(cache: &IncrementalCache) -> Vec<u8> {
179    let mut writer = CanonicalWriter::new();
180    let _ = writer.field_u32(1, CACHE_SCHEMA_VERSION);
181    let _ = writer.field_object_id(2, &cache.baseline_block_id);
182    let _ = writer.field_object_id(3, &cache.horizon_id);
183    let _ = writer.field_u32(4, cache.steps_since_reanchor);
184    for (node_id, node) in cache.state.live_nodes() {
185        if let Ok(record) = encode_node_record(node_id, &node.path, node.kind, &node.content) {
186            let _ = writer.field_raw(10, WireType::RecordListItem, &record);
187        }
188    }
189    for (node_id, tombstone) in cache.state.tombstones() {
190        if let Ok(record) =
191            encode_node_record(node_id, &tombstone.path, tombstone.kind, &tombstone.content)
192        {
193            let _ = writer.field_raw(11, WireType::RecordListItem, &record);
194        }
195    }
196    let body = writer.finish();
197    let checksum = prikk_hash::sha256(&body);
198
199    let mut out = Vec::with_capacity(CACHE_MAGIC.len() + 32 + body.len());
200    out.extend_from_slice(CACHE_MAGIC);
201    out.extend_from_slice(&checksum);
202    out.extend_from_slice(&body);
203    out
204}
205
206fn decode(bytes: &[u8]) -> Option<IncrementalCache> {
207    let after_magic = bytes.strip_prefix(CACHE_MAGIC)?;
208    if after_magic.len() < 32 {
209        return None;
210    }
211    let (checksum, body) = after_magic.split_at(32);
212    if prikk_hash::sha256(body) != checksum {
213        return None;
214    }
215
216    let mut cursor = ByteCursor::new(body);
217    let mut schema_version: Option<u32> = None;
218    let mut baseline_block_id: Option<ObjectId> = None;
219    let mut horizon_id: Option<ObjectId> = None;
220    let mut steps_since_reanchor: Option<u32> = None;
221    let mut state = NodeLifecycleState::new();
222
223    let mut last_tag: Option<u16> = None;
224    while let Some(field) = next_field(&mut cursor)? {
225        if let Some(previous) = last_tag {
226            if field.tag < previous {
227                return None;
228            }
229        }
230        last_tag = Some(field.tag);
231        match field.tag {
232            1 => {
233                if field.wire != WireType::U32 as u8 || schema_version.is_some() {
234                    return None;
235                }
236                schema_version = Some(u32::from_be_bytes(field.value.try_into().ok()?));
237            }
238            2 => {
239                if field.wire != WireType::ObjectId as u8 || baseline_block_id.is_some() {
240                    return None;
241                }
242                baseline_block_id = Some(ObjectId::from_bytes(field.value.try_into().ok()?));
243            }
244            3 => {
245                if field.wire != WireType::ObjectId as u8 || horizon_id.is_some() {
246                    return None;
247                }
248                horizon_id = Some(ObjectId::from_bytes(field.value.try_into().ok()?));
249            }
250            4 => {
251                if field.wire != WireType::U32 as u8 || steps_since_reanchor.is_some() {
252                    return None;
253                }
254                steps_since_reanchor = Some(u32::from_be_bytes(field.value.try_into().ok()?));
255            }
256            10 => {
257                if field.wire != WireType::RecordListItem as u8 {
258                    return None;
259                }
260                let (node_id, path, kind, content) = decode_node_record(field.value)?;
261                state
262                    .seed_live_node(
263                        node_id,
264                        LiveNode {
265                            path,
266                            kind,
267                            content,
268                        },
269                    )
270                    .ok()?;
271            }
272            11 => {
273                if field.wire != WireType::RecordListItem as u8 {
274                    return None;
275                }
276                let (node_id, path, kind, content) = decode_node_record(field.value)?;
277                state
278                    .seed_tombstone(
279                        node_id,
280                        Tombstone {
281                            kind,
282                            content,
283                            path,
284                        },
285                    )
286                    .ok()?;
287            }
288            _ => return None,
289        }
290    }
291
292    if schema_version? != CACHE_SCHEMA_VERSION {
293        return None;
294    }
295    Some(IncrementalCache {
296        baseline_block_id: baseline_block_id?,
297        horizon_id: horizon_id?,
298        steps_since_reanchor: steps_since_reanchor?,
299        state,
300    })
301}
302
303struct Field<'a> {
304    tag: u16,
305    wire: u8,
306    value: &'a [u8],
307}
308
309fn next_field<'a>(cursor: &mut ByteCursor<'a>) -> Option<Option<Field<'a>>> {
310    if cursor.is_finished() {
311        return Some(None);
312    }
313    let tag = cursor.read_u16().ok()?;
314    let wire = cursor.read_array::<1>().ok()?[0];
315    let len = usize::try_from(cursor.read_u64().ok()?).ok()?;
316    let value = cursor.read_exact(len).ok()?;
317    Some(Some(Field { tag, wire, value }))
318}
319
320fn encode_node_record(
321    node_id: &prikk_object::NodeId,
322    path: &RepoPath,
323    kind: prikk_object::NodeKind,
324    content: &NodeContent,
325) -> Result<Vec<u8>> {
326    let mut writer = CanonicalWriter::new();
327    writer.field_repo_path(1, path.as_str())?;
328    writer.field_bytes(2, node_id.as_bytes())?;
329    writer.field_enum_u16(3, kind.code())?;
330    match content {
331        NodeContent::File { blob_id, mode } => {
332            writer.field_object_id(4, blob_id)?;
333            writer.field_u32(5, *mode)?;
334        }
335        NodeContent::Symlink { target } => {
336            writer.field_string(6, target)?;
337        }
338    }
339    Ok(writer.finish())
340}
341
342fn decode_node_record(
343    bytes: &[u8],
344) -> Option<(
345    prikk_object::NodeId,
346    RepoPath,
347    prikk_object::NodeKind,
348    NodeContent,
349)> {
350    let mut cursor = ByteCursor::new(bytes);
351    let mut path: Option<RepoPath> = None;
352    let mut node_id: Option<prikk_object::NodeId> = None;
353    let mut kind: Option<prikk_object::NodeKind> = None;
354    let mut blob_id: Option<ObjectId> = None;
355    let mut mode: Option<u32> = None;
356    let mut target: Option<String> = None;
357
358    let mut last_tag: Option<u16> = None;
359    while let Some(field) = next_field(&mut cursor)? {
360        if let Some(previous) = last_tag {
361            if field.tag < previous {
362                return None;
363            }
364        }
365        last_tag = Some(field.tag);
366        match field.tag {
367            1 => {
368                if field.wire != WireType::RepoPath as u8 || path.is_some() {
369                    return None;
370                }
371                path = Some(RepoPath::parse(core::str::from_utf8(field.value).ok()?).ok()?);
372            }
373            2 => {
374                if field.wire != WireType::Bytes as u8 || node_id.is_some() {
375                    return None;
376                }
377                node_id =
378                    Some(prikk_object::NodeId::try_from_bytes(field.value.try_into().ok()?).ok()?);
379            }
380            3 => {
381                if field.wire != WireType::EnumU16 as u8 || kind.is_some() {
382                    return None;
383                }
384                let code = u16::from_be_bytes(field.value.try_into().ok()?);
385                kind = Some(prikk_object::NodeKind::from_code(code).ok()?);
386            }
387            4 => {
388                if field.wire != WireType::ObjectId as u8 || blob_id.is_some() {
389                    return None;
390                }
391                blob_id = Some(ObjectId::from_bytes(field.value.try_into().ok()?));
392            }
393            5 => {
394                if field.wire != WireType::U32 as u8 || mode.is_some() {
395                    return None;
396                }
397                mode = Some(u32::from_be_bytes(field.value.try_into().ok()?));
398            }
399            6 => {
400                if field.wire != WireType::String as u8 || target.is_some() {
401                    return None;
402                }
403                target = Some(core::str::from_utf8(field.value).ok()?.to_string());
404            }
405            _ => return None,
406        }
407    }
408
409    let path = path?;
410    let node_id = node_id?;
411    let kind = kind?;
412    let content = match kind {
413        prikk_object::NodeKind::TextFile | prikk_object::NodeKind::BinaryFile => {
414            if target.is_some() {
415                return None;
416            }
417            NodeContent::File {
418                blob_id: blob_id?,
419                mode: mode?,
420            }
421        }
422        prikk_object::NodeKind::Symlink => {
423            if blob_id.is_some() || mode.is_some() {
424                return None;
425            }
426            NodeContent::Symlink { target: target? }
427        }
428    };
429    Some((node_id, path, kind, content))
430}
431
432#[cfg(test)]
433mod tests;
prikk_store/lifecycle_cache/incremental.rs

prikk_store/lifecycle_cache/
incremental.rs
