diff --git a/CHANGELOG.md b/CHANGELOG.md index df19040..18b83d4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -146,6 +146,8 @@ All notable changes to Craqle are documented here. - Back up Irokle's Fjall database before its first open with this version. Irokle upgrades schema 1 to schema 2 in place and resumes an interrupted migration. Older Irokle binaries reject schema 2; rollback requires the pre-upgrade backup. +- Topic cursors stored by 0.2 carry no branch fence. Reconciliation replays such a + topic from its start; records a graph clock already covers apply as duplicates. - Derived-index repair preserves surviving RDF source state. It cannot reconstruct source events lost by older bugs without an authoritative history or healthy replica. diff --git a/src/lib.rs b/src/lib.rs index 5e6d7a6..6324607 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6220,6 +6220,43 @@ mod tests { assert!(pair.replica.contains_graph(&graph).unwrap()); } + /// A version 1 cursor written before branch fences replays its topic as duplicates. + #[test] + fn legacy_cursor_replays() { + let pair = replica_pair(); + let graph = GraphId::new("urn:test:legacy-topic-cursor"); + pair.origin + .create_crate(&writer_auth(), crate_request(&graph, "legacy")) + .unwrap(); + pair.replica.reconcile_irokle().unwrap(); + let topic = pair.origin.irokle_topic_id(&graph).unwrap().unwrap(); + let cursor = |node: &CraqleNode| { + let bytes = node.store.applied_topic_clock(topic.as_bytes()).unwrap(); + crate::sync::applied_clock(topic, &bytes.unwrap()).unwrap() + }; + let clock = cursor(&pair.replica); + let before = pair + .replica + .export_rocrate(&AllowAllAuthorizer, &graph) + .unwrap(); + let legacy = crate::sync::legacy_cursor(topic, clock.clone()); + pair.replica + .store + .set_topic_clock(topic.as_bytes(), &legacy) + .unwrap(); + + pair.replica.reconcile_irokle().unwrap(); + let stored = pair.replica.store.applied_topic_clock(topic.as_bytes()); + assert_ne!( + stored.unwrap(), + Some(legacy), + "replay stores a fenced cursor" + ); + assert_eq!(cursor(&pair.replica), clock); + let after = pair.replica.export_rocrate(&AllowAllAuthorizer, &graph); + assert_eq!(after.unwrap(), before); + } + #[test] fn locked_revocation_denies() { let pair = replica_pair(); diff --git a/src/sync.rs b/src/sync.rs index d32b9f4..e0d227a 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -736,7 +736,33 @@ pub(crate) fn applied_clock( topic: irokle::TopicId, bytes: &[u8], ) -> SyncResult { - Ok(decode_topic_cursor(topic, bytes)?.clock) + Ok(stored_cursor(topic, bytes)?.map_or_else(Default::default, |cursor| cursor.clock)) +} + +/// A verified version 1 cursor has no branch fence, so the topic replays from its start. +/// Records a graph clock already covers apply as duplicates; corrupt cursors still fail. +fn stored_cursor(topic: irokle::TopicId, bytes: &[u8]) -> SyncResult> { + match decode_topic_cursor(topic, bytes) { + Ok(cursor) => Ok(Some(cursor)), + Err(CraqleSyncError::ExpiredCursor { .. }) => Ok(None), + Err(error) => Err(error), + } +} + +/// Encodes a version 1 cursor as stores written before branch fences hold it. +#[cfg(test)] +pub(crate) fn legacy_cursor(topic: irokle::TopicId, clock: irokle::ActorClock) -> Vec { + let payload = LegacyCursorPayload { + version: 1, + topic, + clock, + }; + let payload_bytes = postcard::to_allocvec(&payload).unwrap(); + postcard::to_allocvec(&LegacyCursorEnvelope { + payload, + checksum: *blake3::hash(&payload_bytes).as_bytes(), + }) + .unwrap() } pub fn topic_cursor_digest(bytes: &[u8]) -> [u8; 32] { @@ -1459,9 +1485,10 @@ impl CraqleGraphSync for IrokleGraphSync { "injected history failure".to_owned(), ))); } - let stored = cursor - .map(|bytes| decode_topic_cursor(topic_id, bytes)) - .transpose()?; + let stored = match cursor { + Some(bytes) => stored_cursor(topic_id, bytes)?, + None => None, + }; let page = self.node.storage().read_snapshot(|read| { let view = read.topic_view(&topic_id, None)?.ok_or_else(|| { irokle::Error::Storage(format!("missing topic state for {topic_id}")) @@ -2762,17 +2789,7 @@ mod tests { Err(CraqleSyncError::CorruptCursor { .. }) )); - let payload = LegacyCursorPayload { - version: 1, - topic, - clock: irokle::ActorClock::default(), - }; - let payload_bytes = postcard::to_allocvec(&payload).unwrap(); - let legacy = postcard::to_allocvec(&LegacyCursorEnvelope { - payload, - checksum: *blake3::hash(&payload_bytes).as_bytes(), - }) - .unwrap(); + let legacy = legacy_cursor(topic, irokle::ActorClock::default()); assert!(matches!( decode_topic_cursor(topic, &legacy), Err(CraqleSyncError::ExpiredCursor { .. })