appcore_sync_sqlite/
object_pages.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: object_pages.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/10/12 00:00:00 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/10/12 00:00:00 by dnettoRaw
8//      ###########      S: 1.0.2-rc
9// =============================================================================
10
11//! Read-only bounded current-state pages and incremental invalidations.
12use crate::objects::{head, read_object, validate_id};
13use crate::{
14    ObjectHeadV1, OpaqueObjectRecordV1, SqliteOpaqueObjectStoreV1, SqliteSyncError as E,
15    SqliteSyncResult as R,
16};
17use rusqlite::params;
18/// Incremental page cursor plus current values, not historical values as of the cursor.
19/// Values are observed atomically within this page; a newer object revision may exceed `through`.
20#[derive(Clone, Debug, PartialEq, Eq)]
21pub struct ObjectChangesPageV1 {
22    /// Generation/revision observed in the same read transaction.
23    pub head: ObjectHeadV1,
24    /// Resume after this change revision. Always advances or reports completion.
25    pub cursor: u64,
26    /// Whether the requested range has been fully covered.
27    pub complete: bool,
28    /// Current hash-checked objects for the selected changes, including tombstones.
29    pub records: Vec<OpaqueObjectRecordV1>,
30}
31/// Bounded lexicographic scan of current objects, including tombstones.
32#[derive(Clone, Debug, PartialEq, Eq)]
33pub struct ObjectsPageV1 {
34    /// Generation/revision observed by this page, not a frozen multi-page snapshot.
35    pub head: ObjectHeadV1,
36    /// Last admitted key when more records exist; otherwise no continuation.
37    pub next_key: Option<String>,
38    /// Current hash-checked records. Catch up changes from the head captured before scanning.
39    pub records: Vec<OpaqueObjectRecordV1>,
40}
41impl SqliteOpaqueObjectStoreV1 {
42    /// Reads current values for a fixed change range, deduplicated by key within the range.
43    /// `generation` must match the prior head; source replacement fails closed.
44    /// Metadata is admitted before BLOB reads. Too small a budget fails instead of returning a stalled page.
45    pub fn changes(
46        &self,
47        installation: &str,
48        generation: &str,
49        after: u64,
50        through: u64,
51        limit: usize,
52        max_bytes: usize,
53    ) -> R<ObjectChangesPageV1> {
54        validate_page(installation, limit, max_bytes)?;
55        self.0.with_connection(|connection| {
56            let transaction=connection.transaction().map_err(E::database)?;
57            let current=head(&transaction)?;
58            if current.generation!=generation || after>through || through>current.revision {return Err(E::CorruptRecord("opaque object precondition"));}
59            let mut statement=transaction.prepare("SELECT max(c.revision),c.object_key,length(o.envelope)+length(o.metadata) FROM appcore_object_change c JOIN appcore_opaque_object o ON o.installation_id=c.installation_id AND o.object_key=c.object_key WHERE c.installation_id=?1 AND c.revision>?2 AND c.revision<=?3 GROUP BY c.object_key ORDER BY max(c.revision) LIMIT ?4").map_err(E::database)?;
60            let rows=statement.query_map(params![installation,after,through,limit+1],|row|Ok((row.get::<_,u64>(0)?,row.get::<_,String>(1)?,row.get::<_,usize>(2)?))).map_err(E::database)?;
61            let selected=rows.collect::<Result<Vec<_>,_>>().map_err(E::database)?;
62            let mut bytes=0usize;let mut records=Vec::new();let mut cursor=after;let mut complete=true;
63            for (position,key,length) in selected {
64                if records.len()==limit || bytes.saturating_add(length)>max_bytes {complete=false;break;}
65                bytes+=length;cursor=position;
66                records.push(read_object(&transaction,installation,&key)?.ok_or(E::IntegrityFailed)?);
67            }
68            if !complete && records.is_empty() {return Err(E::CapacityExceeded("object page bytes"));}
69            if complete {cursor=through;}
70            Ok(ObjectChangesPageV1 {head:current,cursor,complete,records})
71        })
72    }
73    /// Scans current records; callers must catch up changes after the initial scan head.
74    pub fn scan(
75        &self,
76        installation: &str,
77        after: Option<&str>,
78        limit: usize,
79        max_bytes: usize,
80    ) -> R<ObjectsPageV1> {
81        validate_page(installation, limit, max_bytes)?;
82        if let Some(key) = after {
83            validate_id(key)?;
84        }
85        self.0.with_connection(|connection| {
86            let transaction=connection.transaction().map_err(E::database)?;
87            let current=head(&transaction)?;
88            let mut statement=transaction.prepare("SELECT object_key,length(envelope)+length(metadata) FROM appcore_opaque_object WHERE installation_id=?1 AND object_key>?2 ORDER BY object_key LIMIT ?3").map_err(E::database)?;
89            let rows=statement.query_map(params![installation,after.unwrap_or(""),limit+1],|row|Ok((row.get::<_,String>(0)?,row.get::<_,usize>(1)?))).map_err(E::database)?;
90            let selected=rows.collect::<Result<Vec<_>,_>>().map_err(E::database)?;
91            let mut bytes=0usize;let mut records=Vec::new();let mut last=None;let mut more=false;
92            for (key,length) in selected {
93                if records.len()==limit || bytes.saturating_add(length)>max_bytes {more=true;break;}
94                bytes+=length;last=Some(key.clone());records.push(read_object(&transaction,installation,&key)?.ok_or(E::IntegrityFailed)?);
95            }
96            if more && records.is_empty() {return Err(E::CapacityExceeded("object page bytes"));}
97            Ok(ObjectsPageV1 {head:current,next_key:if more {last}else{None},records})
98        })
99    }
100}
101fn validate_page(installation: &str, limit: usize, max_bytes: usize) -> R<()> {
102    validate_id(installation)?;
103    if limit == 0 || limit > 1000 || max_bytes == 0 || max_bytes > 16 * 1024 * 1024 {
104        return Err(E::CapacityExceeded("object page"));
105    }
106    Ok(())
107}
appcore_sync_sqlite/object_pages.rs

appcore_sync_sqlite/
object_pages.rs
