fix: replay topics with version 1 cursors instead of failing open

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.
 

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();

src/sync.rs

@@ -736,7 +736,33 @@ pub(crate) fn applied_clock(
     topic: irokle::TopicId,
     bytes: &[u8],
 ) -> SyncResult<irokle::ActorClock> {
-    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<Option<TopicCursorPayload>> {
+    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<u8> {
+    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<S: irokle::Storage> CraqleGraphSync for IrokleGraphSync<S> {
                 "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 { .. })