fix: pin Journal source pagination to saved coverage content

crates/subjektiv/src/journal/mod.rs

@@ -31,6 +31,11 @@
 //! A replacement may checkpoint old saved record receipts only for identical selected source bytes.
 //! Tokens incorporate prior saved content conditions as well as current content, preventing A→B→A
 //! edits from resurrecting an old operation premise after intervening saved effects.
+//! Source-ledger pagination uses an opaque, scope/Subject/Journal-bound content cursor: late gap
+//! admission, accepted-fragment extension, reflection, or availability changes reject continuation.
+//! The fingerprint and bounded projection share one transaction; hashing streams bounded rows.
+//! Hosts resuming at the processing cursor can use a bounded source point lookup instead of
+//! traversing already reflected pages. Receipt/entry keysets remain append-stable.
 
 mod store;
 #[cfg(test)]

crates/subjektiv/src/journal/store.rs

@@ -3,6 +3,37 @@ use crate::{issued_id, now};
 use rusqlite::{Connection, OptionalExtension, params};
 use sha2::{Digest, Sha256};
 
+#[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct SourcePageCursor {
+    scope_id: String,
+    subject_id: String,
+    journal_id: String,
+    source_fingerprint: String,
+    after_order: u64,
+}
+
+// Hash rows incrementally rather than materializing an unbounded coverage ledger. The caller
+// holds one storage transaction for both this content condition and the bounded page projection.
+fn source_fingerprint(tx: &Connection, journal_id: &str) -> JournalResult<String> {
+    let mut hash = Sha256::new();
+    let mut stmt=tx.prepare("SELECT source_order, source_json, accepted_json, unavailable_json, reflected_bytes FROM journal_sources WHERE journal_id=?1 ORDER BY source_order")?;
+    let mut rows = stmt.query([journal_id])?;
+    while let Some(row) = rows.next()? {
+        let value = (
+            row.get::<_, u64>(0)?,
+            row.get::<_, String>(1)?,
+            row.get::<_, String>(2)?,
+            row.get::<_, String>(3)?,
+            row.get::<_, u64>(4)?,
+        );
+        let bytes = serde_json::to_vec(&value)?;
+        hash.update((bytes.len() as u64).to_le_bytes());
+        hash.update(bytes);
+    }
+    Ok(hash.finalize().iter().map(|b| format!("{b:02x}")).collect())
+}
+
 fn invalid(message: &str) -> JournalError {
     JournalError::InvalidInput(message.into())
 }
@@ -185,6 +216,7 @@ fn refresh_state(tx: &Connection, j: &mut Journal) -> JournalResult<()> {
 }
 fn check_source(range: &JournalSourceRange, session: &str) -> JournalResult<()> {
     let s = &range.source;
+    label(&s.session_id)?;
     let suffix = s
         .pointer
         .entry_ref
@@ -741,23 +773,49 @@ impl SubjektivStore {
             fit_page(items,more)
         })
     }
+    /// Coverage changes can fill an earlier gap or extend an already listed entry. Continuation
+    /// pins the complete ledger content condition, rejecting changes instead of losing those bytes.
     pub fn list_journal_sources(
         &self,
         host: &JournalHostContext,
         journal_id: &str,
-        after: Option<u64>,
+        cursor: Option<&str>,
         limit: usize,
     ) -> JournalResult<JournalPage<JournalSourceStatus>> {
         let limit = page_limit(limit)?;
-        self.database.try_with_connection(|tx| {
+        self.database.try_transaction(|tx| {
             check_host(self,tx,host,false)?;journal(tx,host,journal_id,false)?;
+            let fingerprint=source_fingerprint(tx,journal_id)?;
+            let after=if let Some(cursor)=cursor {
+                if cursor.len()>2048 {return Err(invalid("source cursor exceeds byte limit"));}
+                let cursor:SourcePageCursor=serde_json::from_str(cursor).map_err(|_|invalid("invalid source cursor"))?;
+                if cursor.scope_id!=host.scope_id || cursor.subject_id!=host.subject_id || cursor.journal_id!=journal_id || cursor.after_order>i64::MAX as u64 {return Err(invalid("source cursor belongs to a different scope/Subject/Journal or invalid order"));}
+                if cursor.source_fingerprint!=fingerprint {return Err(conflict("source coverage changed; restart pagination from current state"));}
+                Some(cursor.after_order)
+            } else {None};
             let mut stmt=tx.prepare("SELECT source_order FROM journal_sources WHERE journal_id=?1 AND (?2 IS NULL OR source_order>?2) ORDER BY source_order LIMIT ?3")?;
             let orders=stmt.query_map(params![journal_id,after,limit+1],|r|r.get::<_,u64>(0))?.collect::<Result<Vec<_>,_>>()?;
             let more=orders.len()>limit;let mut items=Vec::new();
-            for order in orders.into_iter().take(limit) {items.push((order.to_string(),source(tx,journal_id,order)?));}
+            for order in orders.into_iter().take(limit) {
+                let next=serde_json::to_string(&SourcePageCursor {scope_id:host.scope_id.clone(),subject_id:host.subject_id.clone(),journal_id:journal_id.into(),source_fingerprint:fingerprint.clone(),after_order:order})?;
+                items.push((next,source(tx,journal_id,order)?));
+            }
             fit_page(items,more)
         })
     }
+    /// Bounded point lookup for Host resume from `process_order`, without scanning reflected pages.
+    pub fn read_journal_source(
+        &self,
+        host: &JournalHostContext,
+        journal_id: &str,
+        order: u64,
+    ) -> JournalResult<JournalSourceStatus> {
+        self.database.try_with_connection(|tx| {
+            check_host(self, tx, host, false)?;
+            journal(tx, host, journal_id, false)?;
+            source(tx, journal_id, order)
+        })
+    }
     /// Returns metadata only. IDs and entry positions are append-stable keysets, so appends/rewrites cannot shift unread pages.
     pub fn list_journals(
         &self,

crates/subjektiv/src/journal/tests.rs

@@ -1747,7 +1747,7 @@ fn metadata_and_operation_keysets_do_not_skip_appends_or_sort_by_request_id() {
         .list_journal_sources(
             &fixture.host,
             &fixture.journal_id,
-            Some(source_page.next.unwrap().parse().unwrap()),
+            source_page.next.as_deref(),
             2,
         )
         .unwrap();
@@ -2444,3 +2444,180 @@ fn journal_migration_failure_rolls_back_ddl_and_retries_without_changing_existin
         "post-migration-session"
     );
 }
+
+#[test]
+fn late_gap_admission_and_partial_updates_reject_stale_source_page_cursors() {
+    let fixture = Fixture::with_sizes(&[8]);
+    let second = source_range("source-session", 1, 8);
+    let third = source_range("source-session", 2, 8);
+    let fourth = source_range("source-session", 3, 8);
+    fixture
+        .store()
+        .accept_journal_sources(&fixture.host, "", &[third, fourth])
+        .unwrap();
+    let first = fixture
+        .store()
+        .list_journal_sources(&fixture.host, &fixture.journal_id, None, 2)
+        .unwrap();
+    assert_eq!(
+        first
+            .items
+            .iter()
+            .map(|s| s.source.order)
+            .collect::<Vec<_>>(),
+        vec![0, 2]
+    );
+    let cursor = first.next.unwrap();
+    fixture
+        .store()
+        .accept_journal_sources(&fixture.host, "", &[second])
+        .unwrap();
+    let next =
+        fixture
+            .store()
+            .list_journal_sources(&fixture.host, &fixture.journal_id, Some(&cursor), 2);
+    assert!(
+        matches!(next, Err(JournalError::Conflict(_))),
+        "gap completion must not silently skip newly admitted order 1: {next:?}"
+    );
+    let fresh = fixture
+        .store()
+        .list_journal_sources(&fixture.host, &fixture.journal_id, None, 2)
+        .unwrap();
+    assert_eq!(
+        fresh
+            .items
+            .iter()
+            .map(|s| s.source.order)
+            .collect::<Vec<_>>(),
+        vec![0, 1]
+    );
+    let next = fixture
+        .store()
+        .list_journal_sources(&fixture.host, &fixture.journal_id, fresh.next.as_deref(), 2)
+        .unwrap();
+    assert_eq!(
+        next.items
+            .iter()
+            .map(|s| s.source.order)
+            .collect::<Vec<_>>(),
+        vec![2, 3]
+    );
+    // Duplicate notifications do not change the content condition.
+    fixture
+        .store()
+        .accept_journal_sources(&fixture.host, "", &fixture.ranges)
+        .unwrap();
+    assert!(
+        fixture
+            .store()
+            .list_journal_sources(&fixture.host, &fixture.journal_id, fresh.next.as_deref(), 2)
+            .is_ok()
+    );
+    let selected = fixture.ranges[0].clone();
+    let (_, job) = fixture.generation("reflect-during-list", vec![selected.clone()]);
+    fixture.execute(
+        &job,
+        "no-change",
+        JournalAction::ConfirmUnchanged,
+        vec![selected.clone()],
+    );
+    assert!(
+        fixture
+            .store()
+            .list_journal_sources(&fixture.host, &fixture.journal_id, fresh.next.as_deref(), 2)
+            .is_ok()
+    );
+    fixture.execute(
+        &job,
+        "checkpoint",
+        checkpoint(&["no-change"]),
+        vec![selected],
+    );
+    assert!(matches!(
+        fixture.store().list_journal_sources(
+            &fixture.host,
+            &fixture.journal_id,
+            fresh.next.as_deref(),
+            2
+        ),
+        Err(JournalError::Conflict(_))
+    ));
+    assert_eq!(
+        fixture
+            .store()
+            .read_journal_source(&fixture.host, &fixture.journal_id, 0)
+            .unwrap()
+            .reflected_bytes,
+        8
+    );
+
+    let host = session_host(fixture.store(), &fixture.host.subject_id, "partial-session");
+    let range = source_range("partial-session", 0, 16);
+    let later = source_range("partial-session", 1, 8);
+    let id = fixture
+        .store()
+        .accept_journal_sources(&host, "Partial source", &[fragment(&range, 0, 4), later])
+        .unwrap();
+    let page = fixture
+        .store()
+        .list_journal_sources(&host, &id, None, 1)
+        .unwrap();
+    fixture
+        .store()
+        .accept_journal_sources(&host, "", &[fragment(&range, 4, 8)])
+        .unwrap();
+    assert!(matches!(
+        fixture
+            .store()
+            .list_journal_sources(&host, &id, page.next.as_deref(), 1),
+        Err(JournalError::Conflict(_))
+    ));
+    let page = fixture
+        .store()
+        .list_journal_sources(&host, &id, None, 1)
+        .unwrap();
+    fixture
+        .store()
+        .mark_journal_sources_unavailable(&host, &id, &[fragment(&range, 0, 8)])
+        .unwrap();
+    assert!(matches!(
+        fixture
+            .store()
+            .list_journal_sources(&host, &id, page.next.as_deref(), 1),
+        Err(JournalError::Conflict(_))
+    ));
+    let page = fixture
+        .store()
+        .list_journal_sources(&host, &id, None, 1)
+        .unwrap();
+    fixture
+        .store()
+        .mark_journal_sources_available(&host, &id, &[fragment(&range, 0, 4)])
+        .unwrap();
+    assert!(matches!(
+        fixture
+            .store()
+            .list_journal_sources(&host, &id, page.next.as_deref(), 1),
+        Err(JournalError::Conflict(_))
+    ));
+    let page = fixture
+        .store()
+        .list_journal_sources(&host, &id, None, 1)
+        .unwrap();
+    assert!(matches!(
+        fixture
+            .store()
+            .list_journal_sources(&host, &fixture.journal_id, page.next.as_deref(), 1),
+        Err(JournalError::InvalidInput(_))
+    ));
+    let mut cursor: serde_json::Value =
+        serde_json::from_str(page.next.as_deref().unwrap()).unwrap();
+    cursor["unexpected_field"] = true.into();
+    assert!(matches!(
+        fixture
+            .store()
+            .list_journal_sources(&host, &id, Some(&cursor.to_string()), 1),
+        Err(JournalError::InvalidInput(_))
+    ));
+}