From c2df3041f9ce100a72dc71a140071605f767a226 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 14 Oct 2026 22:18:02 +0900 Subject: [PATCH] fix: pin Journal source pagination to saved coverage content --- crates/subjektiv/src/journal/mod.rs | 5 + crates/subjektiv/src/journal/store.rs | 64 ++++++++- crates/subjektiv/src/journal/tests.rs | 179 +++++++++++++++++++++++++- 3 files changed, 244 insertions(+), 4 deletions(-) diff --git a/crates/subjektiv/src/journal/mod.rs b/crates/subjektiv/src/journal/mod.rs index 707c1d3a2..d3cd57a80 100644 --- a/crates/subjektiv/src/journal/mod.rs +++ b/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)] diff --git a/crates/subjektiv/src/journal/store.rs b/crates/subjektiv/src/journal/store.rs index d24c51154..a5b8d7dfa 100644 --- a/crates/subjektiv/src/journal/store.rs +++ b/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 { + 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, + cursor: Option<&str>, limit: usize, ) -> JournalResult> { 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::,_>>()?; 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 { + 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, diff --git a/crates/subjektiv/src/journal/tests.rs b/crates/subjektiv/src/journal/tests.rs index b04b3888b..433b44bf7 100644 --- a/crates/subjektiv/src/journal/tests.rs +++ b/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![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![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![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(_)) + )); +}