appcore_sync_sqlite/
work.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: work.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//! One explicit `SQLite` transaction across log, outbox and checkpoints.
12use crate::{SqliteSyncError, SqliteSyncResult, SqliteSyncStore};
13use appcore_sync::SyncMessage;
14use rusqlite::{Transaction, TransactionBehavior};
15
16/// Bounded provider-owned unit of work; never exposes SQL or a connection.
17/// Callbacks must be short and must not reenter this store or wait for network I/O.
18/// Filesystem writes are outside this transaction and require a recovery journal.
19pub struct SqliteSyncWork<'a> {
20    transaction: &'a Transaction<'a>,
21    config: &'a crate::SqliteSyncConfig,
22    operations: usize,
23}
24impl SqliteSyncStore {
25    /// Commits all operations only when the callback succeeds. An error or unwind
26    /// drops the transaction and rolls back every operation. At most 1,024
27    /// operations may be submitted. The deployment owns callback execution bounds.
28    pub fn transaction<T>(
29        &self,
30        action: impl FnOnce(&mut SqliteSyncWork<'_>) -> SqliteSyncResult<T>,
31    ) -> SqliteSyncResult<T> {
32        self.with_connection(|connection| {
33            let transaction = connection
34                .transaction_with_behavior(TransactionBehavior::Immediate)
35                .map_err(SqliteSyncError::database)?;
36            let result = action(&mut SqliteSyncWork {
37                transaction: &transaction,
38                config: self.config(),
39                operations: 0,
40            })?;
41            transaction.commit().map_err(SqliteSyncError::database)?;
42            Ok(result)
43        })
44    }
45}
46impl SqliteSyncWork<'_> {
47    /// Appends an opaque record. Identical positive sequence replay is a no-op.
48    pub fn append(&mut self, payload: &[u8], sequence: u64) -> SqliteSyncResult<usize> {
49        self.admit()?;
50        crate::log::append_in_transaction(self.transaction, payload, sequence)
51    }
52    /// Enqueues one bounded message. Identical pending replay returns false;
53    /// a full queue returns an error, ensuring the whole unit of work rolls back.
54    pub fn enqueue(&mut self, message: &SyncMessage) -> SqliteSyncResult<bool> {
55        self.admit()?;
56        crate::outbox::validate_batch_id(&message.batch_id)
57            .map_err(|_| SqliteSyncError::CorruptRecord("outbox identity"))?;
58        let bytes =
59            crate::outbox_blob::encoded_message_bytes(message, self.config.max_outbox_record_bytes)
60                .map_err(|_| SqliteSyncError::CapacityExceeded("outbox record"))?;
61        let inserted = crate::outbox::enqueue_in_transaction(
62            self.transaction,
63            message,
64            bytes,
65            self.config.max_outbox_entries,
66        )?;
67        if !inserted {
68            let exists: bool = self
69                .transaction
70                .query_row(
71                    "SELECT EXISTS(SELECT 1 FROM appcore_sync_outbox WHERE batch_id = ?1)",
72                    [&message.batch_id],
73                    |row| row.get(0),
74                )
75                .map_err(SqliteSyncError::database)?;
76            if !exists {
77                return Err(SqliteSyncError::CapacityExceeded("outbox"));
78            }
79        }
80        Ok(inserted)
81    }
82    /// Advances a checkpoint only within an authenticated observed history head.
83    /// Head authenticity and materialization are the caller's responsibility.
84    pub fn checkpoint(
85        &mut self,
86        peer: &str,
87        sequence: u64,
88        hash: &str,
89        observed_head: u64,
90    ) -> SqliteSyncResult<()> {
91        self.admit()?;
92        appcore_sync::validate_checkpoint_head_v1(sequence, observed_head)
93            .map_err(|_| SqliteSyncError::CorruptRecord("future checkpoint"))?;
94        crate::checkpoint::validate_peer_id(peer)
95            .and_then(|()| crate::checkpoint::validate_hash(hash))
96            .map_err(|_| SqliteSyncError::CorruptRecord("checkpoint"))?;
97        let sequence =
98            i64::try_from(sequence).map_err(|_| SqliteSyncError::CapacityExceeded("checkpoint"))?;
99        crate::checkpoint::set_in_transaction(
100            self.transaction,
101            peer,
102            sequence,
103            hash,
104            self.config.max_checkpoints,
105        )
106    }
107    /// Journals opaque bytes before filesystem publication in the same transaction
108    /// as enqueue/log/checkpoint. Replay materializes committed intents only.
109    pub fn stage_object_write(
110        &mut self,
111        intent: &crate::ObjectWriteIntentV1,
112    ) -> SqliteSyncResult<bool> {
113        self.admit()?;
114        crate::materialization::stage(self.transaction, intent)
115    }
116    /// Applies an exact ordered outbox receipt in this transaction, so checkpoint
117    /// advancement and acknowledgement may share one commit boundary.
118    pub fn acknowledge(
119        &mut self,
120        receipt: &appcore_sync::SyncOutboxReceipt,
121    ) -> SqliteSyncResult<usize> {
122        self.admit()?;
123        crate::outbox::acknowledge_in_transaction(self.transaction, receipt)
124    }
125    /// Mutates a canonical opaque object atomically with this work's sync operations.
126    /// Exact mutation replay returns its original revision without replacing later data.
127    /// Reusing an identity with different bytes fails. Propagate every returned error.
128    pub fn put_object_v1(
129        &mut self,
130        object: &crate::OpaqueObjectV1,
131        mutation_id: &str,
132        expected: crate::ObjectRevisionV1,
133    ) -> SqliteSyncResult<crate::ObjectWriteResultV1> {
134        self.admit()?;
135        crate::objects::mutate(self.transaction, object, mutation_id, expected)
136    }
137    fn admit(&mut self) -> SqliteSyncResult<()> {
138        if self.operations >= 1_024 {
139            return Err(SqliteSyncError::CapacityExceeded("transaction operation"));
140        }
141        self.operations += 1;
142        Ok(())
143    }
144}
appcore_sync_sqlite/work.rs

appcore_sync_sqlite/
work.rs
