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

appcore_sync_sqlite/
tombstone.rs
