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            self.config.outbox_table(),
67            self.config.outbox_database(),
68        )?;
69        if !inserted {
70            let exists: bool = self
71                .transaction
72                .query_row(
73                    &format!(
74                        "SELECT EXISTS(SELECT 1 FROM {} WHERE batch_id = ?1)",
75                        self.config.outbox_table()
76                    ),
77                    [&message.batch_id],
78                    |row| row.get(0),
79                )
80                .map_err(SqliteSyncError::database)?;
81            if !exists {
82                return Err(SqliteSyncError::CapacityExceeded("outbox"));
83            }
84        }
85        Ok(inserted)
86    }
87    /// Advances a checkpoint only within an authenticated observed history head.
88    /// Head authenticity and materialization are the caller's responsibility.
89    pub fn checkpoint(
90        &mut self,
91        peer: &str,
92        sequence: u64,
93        hash: &str,
94        observed_head: u64,
95    ) -> SqliteSyncResult<()> {
96        self.admit()?;
97        appcore_sync::validate_checkpoint_head_v1(sequence, observed_head)
98            .map_err(|_| SqliteSyncError::CorruptRecord("future checkpoint"))?;
99        crate::checkpoint::validate_peer_id(peer)
100            .and_then(|()| crate::checkpoint::validate_hash(hash))
101            .map_err(|_| SqliteSyncError::CorruptRecord("checkpoint"))?;
102        let sequence =
103            i64::try_from(sequence).map_err(|_| SqliteSyncError::CapacityExceeded("checkpoint"))?;
104        crate::checkpoint::set_in_transaction(
105            self.transaction,
106            peer,
107            sequence,
108            hash,
109            self.config.max_checkpoints,
110        )
111    }
112    /// Journals opaque bytes before filesystem publication in the same transaction
113    /// as enqueue/log/checkpoint. Replay materializes committed intents only.
114    pub fn stage_object_write(
115        &mut self,
116        intent: &crate::ObjectWriteIntentV1,
117    ) -> SqliteSyncResult<bool> {
118        self.admit()?;
119        crate::materialization::stage(self.transaction, intent)
120    }
121    /// Applies an exact ordered outbox receipt in this transaction, so checkpoint
122    /// advancement and acknowledgement may share one commit boundary.
123    pub fn acknowledge(
124        &mut self,
125        receipt: &appcore_sync::SyncOutboxReceipt,
126    ) -> SqliteSyncResult<usize> {
127        self.admit()?;
128        crate::outbox::acknowledge_in_transaction(
129            self.transaction,
130            receipt,
131            self.config.outbox_table(),
132        )
133    }
134    /// Mutates a canonical opaque object atomically with this work's sync operations.
135    /// Exact mutation replay returns its original revision without replacing later data.
136    /// Reusing an identity with different bytes fails. Propagate every returned error.
137    pub fn put_object_v1(
138        &mut self,
139        object: &crate::OpaqueObjectV1,
140        mutation_id: &str,
141        expected: crate::ObjectRevisionV1,
142    ) -> SqliteSyncResult<crate::ObjectWriteResultV1> {
143        self.admit()?;
144        crate::objects::mutate(self.transaction, object, mutation_id, expected)
145    }
146    fn admit(&mut self) -> SqliteSyncResult<()> {
147        if self.operations >= 1_024 {
148            return Err(SqliteSyncError::CapacityExceeded("transaction operation"));
149        }
150        self.operations += 1;
151        Ok(())
152    }
153}
appcore_sync_sqlite/work.rs

appcore_sync_sqlite/
work.rs
