appcore_sync_sqlite/
tombstone.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: tombstone.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/08/26 00:00:00 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/26 00:00:00 by dnettoRaw
8//      ###########      S: 2.0.0
9// =============================================================================
10
11//! Defines bounded tombstone contracts and behavior for this crate.
12
13use crate::{SqliteSyncError, SqliteSyncResult, SqliteSyncStore};
14use rusqlite::{params, OptionalExtension, TransactionBehavior};
15
16const MAX_NAMESPACE_BYTES: usize = 128;
17const MAX_OPAQUE_KEY_BYTES: usize = 512;
18
19/// One opaque Runtime sync deletion marker.
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct SqliteSyncTombstone {
22    /// Provider-neutral namespace owned by synchronization infrastructure.
23    pub namespace: String,
24    /// Opaque deletion identity; its business meaning is not interpreted.
25    pub opaque_key: String,
26    /// Replication sequence at which deletion occurred.
27    pub deleted_sequence: u64,
28    /// SHA-256 hash bound to the deleted opaque payload.
29    pub payload_hash: String,
30    /// Expiry time in Unix epoch milliseconds.
31    pub expires_at_ms: u64,
32}
33
34/// Bounded `SQLite` tombstone storage for conservative deletion replication.
35#[derive(Debug, Clone)]
36pub struct SqliteSyncTombstoneStore {
37    store: SqliteSyncStore,
38}
39
40impl SqliteSyncTombstoneStore {
41    pub(crate) fn new(store: SqliteSyncStore) -> Self {
42        Self { store }
43    }
44
45    /// Atomically inserts or advances one opaque deletion marker.
46    pub fn record(&self, tombstone: &SqliteSyncTombstone) -> SqliteSyncResult<bool> {
47        validate_tombstone(tombstone)?;
48        self.store.with_connection(|connection| {
49            let transaction = connection
50                .transaction_with_behavior(TransactionBehavior::Immediate)
51                .map_err(SqliteSyncError::database)?;
52            let existing: Option<(i64, String, i64)> = transaction
53                .query_row(
54                    "SELECT deleted_sequence, payload_hash, expires_at_ms
55                     FROM appcore_sync_tombstone WHERE namespace = ?1 AND opaque_key = ?2",
56                    params![tombstone.namespace, tombstone.opaque_key],
57                    |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
58                )
59                .optional()
60                .map_err(SqliteSyncError::database)?;
61            let incoming_sequence = to_i64(tombstone.deleted_sequence, "tombstone sequence")?;
62            let incoming_expiry = to_i64(tombstone.expires_at_ms, "tombstone expiry")?;
63            if let Some((sequence, hash, expiry)) = existing {
64                if incoming_sequence < sequence {
65                    return Ok(false);
66                }
67                if incoming_sequence == sequence {
68                    return if hash == tombstone.payload_hash && expiry == incoming_expiry {
69                        Ok(false)
70                    } else {
71                        Err(SqliteSyncError::CorruptRecord("tombstone conflict"))
72                    };
73                }
74            } else if count_tombstones(&transaction)? >= self.store.config().max_tombstones {
75                return Err(SqliteSyncError::CapacityExceeded("tombstone"));
76            }
77            let changed = transaction
78                .execute(
79                    "INSERT INTO appcore_sync_tombstone
80                     (namespace, opaque_key, deleted_sequence, payload_hash, expires_at_ms)
81                     VALUES (?1, ?2, ?3, ?4, ?5)
82                     ON CONFLICT(namespace, opaque_key) DO UPDATE SET
83                     deleted_sequence = excluded.deleted_sequence,
84                     payload_hash = excluded.payload_hash,
85                     expires_at_ms = excluded.expires_at_ms
86                     WHERE excluded.deleted_sequence >= appcore_sync_tombstone.deleted_sequence",
87                    params![
88                        tombstone.namespace,
89                        tombstone.opaque_key,
90                        incoming_sequence,
91                        tombstone.payload_hash,
92                        incoming_expiry
93                    ],
94                )
95                .map_err(SqliteSyncError::database)?;
96            transaction.commit().map_err(SqliteSyncError::database)?;
97            Ok(changed == 1)
98        })
99    }
100
101    /// Returns at most `limit` unexpired markers in deterministic order.
102    pub fn active(&self, now_ms: u64, limit: usize) -> SqliteSyncResult<Vec<SqliteSyncTombstone>> {
103        if limit == 0 || limit > self.store.config().max_tombstones {
104            return Err(SqliteSyncError::CapacityExceeded("tombstone read"));
105        }
106        self.store.with_connection(|connection| {
107            let mut statement = connection
108                .prepare(
109                    "SELECT namespace, opaque_key, deleted_sequence, payload_hash, expires_at_ms
110                     FROM appcore_sync_tombstone WHERE expires_at_ms > ?1
111                     ORDER BY namespace, opaque_key LIMIT ?2",
112                )
113                .map_err(SqliteSyncError::database)?;
114            let rows = statement
115                .query_map(
116                    params![
117                        to_i64(now_ms, "current time")?,
118                        to_i64(limit as u64, "limit")?
119                    ],
120                    |row| {
121                        Ok((
122                            row.get::<_, String>(0)?,
123                            row.get::<_, String>(1)?,
124                            row.get::<_, i64>(2)?,
125                            row.get::<_, String>(3)?,
126                            row.get::<_, i64>(4)?,
127                        ))
128                    },
129                )
130                .map_err(SqliteSyncError::database)?;
131            let mut tombstones = Vec::new();
132            for row in rows {
133                let (namespace, opaque_key, sequence, payload_hash, expiry) =
134                    row.map_err(SqliteSyncError::database)?;
135                tombstones.push(SqliteSyncTombstone {
136                    namespace,
137                    opaque_key,
138                    deleted_sequence: to_u64(sequence, "tombstone sequence")?,
139                    payload_hash,
140                    expires_at_ms: to_u64(expiry, "tombstone expiry")?,
141                });
142            }
143            Ok(tombstones)
144        })
145    }
146
147    /// Deletes at most `limit` expired markers and returns the deletion count.
148    pub fn prune_expired(&self, now_ms: u64, limit: usize) -> SqliteSyncResult<usize> {
149        if limit == 0 || limit > self.store.config().max_tombstones {
150            return Err(SqliteSyncError::CapacityExceeded("tombstone prune"));
151        }
152        self.store.with_connection(|connection| {
153            connection
154                .execute(
155                    "DELETE FROM appcore_sync_tombstone WHERE (namespace, opaque_key) IN (
156                         SELECT namespace, opaque_key FROM appcore_sync_tombstone
157                         WHERE expires_at_ms <= ?1 ORDER BY expires_at_ms LIMIT ?2
158                     )",
159                    params![
160                        to_i64(now_ms, "current time")?,
161                        to_i64(limit as u64, "limit")?
162                    ],
163                )
164                .map_err(SqliteSyncError::database)
165        })
166    }
167
168    /// Returns the current retained marker count.
169    pub fn len(&self) -> SqliteSyncResult<usize> {
170        self.store
171            .with_connection(|connection| count_tombstones(connection))
172    }
173
174    /// Reports whether no markers are retained.
175    pub fn is_empty(&self) -> SqliteSyncResult<bool> {
176        self.len().map(|length| length == 0)
177    }
178}
179
180fn count_tombstones(connection: &rusqlite::Connection) -> SqliteSyncResult<usize> {
181    let count: i64 = connection
182        .query_row("SELECT COUNT(*) FROM appcore_sync_tombstone", [], |row| {
183            row.get(0)
184        })
185        .map_err(SqliteSyncError::database)?;
186    usize::try_from(count).map_err(|_| SqliteSyncError::CorruptRecord("tombstone count"))
187}
188
189pub(crate) fn validate_tombstone(tombstone: &SqliteSyncTombstone) -> SqliteSyncResult<()> {
190    if !valid_identifier(&tombstone.namespace, MAX_NAMESPACE_BYTES)
191        || !valid_identifier(&tombstone.opaque_key, MAX_OPAQUE_KEY_BYTES)
192        || tombstone.deleted_sequence == 0
193        || tombstone.expires_at_ms == 0
194        || tombstone.payload_hash.len() != 64
195        || !tombstone
196            .payload_hash
197            .bytes()
198            .all(|byte| byte.is_ascii_hexdigit())
199    {
200        return Err(SqliteSyncError::CorruptRecord("tombstone input"));
201    }
202    Ok(())
203}
204
205fn valid_identifier(value: &str, max_bytes: usize) -> bool {
206    !value.is_empty()
207        && value.len() <= max_bytes
208        && value
209            .chars()
210            .all(|character| character.is_ascii_graphic() && !character.is_ascii_control())
211}
212
213fn to_i64(value: u64, resource: &'static str) -> SqliteSyncResult<i64> {
214    i64::try_from(value).map_err(|_| SqliteSyncError::CapacityExceeded(resource))
215}
216
217fn to_u64(value: i64, resource: &'static str) -> SqliteSyncResult<u64> {
218    u64::try_from(value).map_err(|_| SqliteSyncError::CorruptRecord(resource))
219}
appcore_sync_sqlite/tombstone.rs

appcore_sync_sqlite/
tombstone.rs
