Skip to main content

max / synckit

Expose the dead-letter hold to consumers D4 step 5 needs to render what the device is holding, which needs three things the crate did not offer: a view across every scope, the payload behind each held entry, and a retry that a UI can call. HeldEntry gains `scope` and `payload`. The payload is what lets a consumer name a held row in terms its user recognises ("Task: Call the bank") instead of a wire row id, which means nothing to anyone. A payload that no longer parses degrades to None rather than failing the read. list_all and counts_all cover every scope. A device syncs personal plus one scope per group, and a row held in a group scope is exactly as absent as one held in the personal scope, so a status surface scoped to personal would hide the failures this whole mechanism exists to show. SyncStore grows held_counts, held_entries, and retry_held so a consumer goes through the facade it already holds rather than opening its own connection and reimplementing what counts as retryable. 415 lib + 111 integration tests pass, clippy clean.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-27 23:07 UTC
Signed with PGP, not checked
Commit: bc387de75ec620f8b48cda2d8215b09ca0418cce
Parent: 6467dd5
2 files changed, +96 insertions, -23 deletions
M synckit-client/src/store/deferred.rs +71 -23
@@ -88,6 +88,8 @@
88 88 /// One entry in the hold, as listed for a human.
89 89 #[derive(Debug, Clone, PartialEq, Eq)]
90 90 pub struct HeldEntry {
91 + /// Scope the change belongs to: `""` for personal, otherwise the group id.
92 + pub scope: String,
91 93 /// Table the change targets.
92 94 pub table: String,
93 95 /// Wire row id of the change.
@@ -100,6 +102,10 @@
100 102 pub attempts: i64,
101 103 /// When the entry was first held (RFC 3339, UTC).
102 104 pub first_seen: String,
105 + /// The held change's row payload, so a consumer can name the row in terms
106 + /// its user recognises. `None` for a delete, or a payload that no longer
107 + /// parses.
108 + pub payload: Option<serde_json::Value>,
103 109 }
104 110
105 111 /// How much is being held for a scope, for a sync-status surface.
@@ -172,27 +178,55 @@
172 178 Ok(out)
173 179 }
174 180
181 + const LIST_COLUMNS: &str =
182 + "scope, table_name, row_id, cause, state, attempts, first_seen, entry FROM sync_deferred";
183 +
184 + fn read_held(r: &rusqlite::Row<'_>) -> rusqlite::Result<HeldEntry> {
185 + let state: String = r.get(4)?;
186 + Ok(HeldEntry {
187 + scope: r.get(0)?,
188 + table: r.get(1)?,
189 + row_id: r.get(2)?,
190 + cause: r.get(3)?,
191 + state: if state == "rejected" {
192 + HoldState::Rejected
193 + } else {
194 + HoldState::Deferred
195 + },
196 + attempts: r.get(5)?,
197 + first_seen: r.get(6)?,
198 + // The payload is handed back so a consumer can label the row with
199 + // something a person recognises (a task's title) instead of a wire row
200 + // id. A payload that will not parse degrades to None, never an error.
201 + payload: serde_json::from_str::<ChangeEntry>(&r.get::<_, String>(7)?)
202 + .ok()
203 + .and_then(|e| e.data),
204 + })
205 + }
206 +
175 207 /// List everything held for a scope, newest first, for a UI surface.
176 208 pub fn list(conn: &Connection, scope: &str) -> Result<Vec<HeldEntry>> {
177 - let mut stmt = conn.prepare(
178 - "SELECT table_name, row_id, cause, state, attempts, first_seen \
179 - FROM sync_deferred WHERE scope = ?1 ORDER BY last_seen DESC",
180 - )?;
181 - let rows = stmt.query_map([scope], |r| {
182 - let state: String = r.get(3)?;
183 - Ok(HeldEntry {
184 - table: r.get(0)?,
185 - row_id: r.get(1)?,
186 - cause: r.get(2)?,
187 - state: if state == "rejected" {
188 - HoldState::Rejected
189 - } else {
190 - HoldState::Deferred
191 - },
192 - attempts: r.get(4)?,
193 - first_seen: r.get(5)?,
194 - })
195 - })?;
209 + let mut stmt = conn.prepare(&format!(
210 + "SELECT {LIST_COLUMNS} WHERE scope = ?1 ORDER BY last_seen DESC"
211 + ))?;
212 + collect(stmt.query_map([scope], read_held)?)
213 + }
214 +
215 + /// List everything held across every scope, newest first.
216 + ///
217 + /// A device syncs its personal scope plus one per group, and a row held in a
218 + /// group scope is just as lost as one held in the personal scope, so a status
219 + /// surface that covered only personal would hide the failures this exists to
220 + /// show.
221 + pub fn list_all(conn: &Connection) -> Result<Vec<HeldEntry>> {
222 + let mut stmt = conn.prepare(&format!("SELECT {LIST_COLUMNS} ORDER BY last_seen DESC"))?;
223 + collect(stmt.query_map([], read_held)?)
224 + }
225 +
226 + fn collect<I>(rows: I) -> Result<Vec<HeldEntry>>
227 + where
228 + I: Iterator<Item = rusqlite::Result<HeldEntry>>,
229 + {
196 230 let mut out = Vec::new();
197 231 for row in rows {
198 232 out.push(row?);
@@ -200,14 +234,28 @@
200 234 Ok(out)
201 235 }
202 236
237 + fn read_tally(r: &rusqlite::Row<'_>) -> rusqlite::Result<(String, i64)> {
238 + Ok((r.get(0)?, r.get(1)?))
239 + }
240 +
203 241 /// Held counts for a scope.
204 242 pub fn counts(conn: &Connection, scope: &str) -> Result<HoldCounts> {
205 - let mut out = HoldCounts::default();
206 243 let mut stmt =
207 244 conn.prepare("SELECT state, COUNT(*) FROM sync_deferred WHERE scope = ?1 GROUP BY state")?;
208 - let rows = stmt.query_map([scope], |r| {
209 - Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?))
210 - })?;
245 + tally(stmt.query_map([scope], read_tally)?)
246 + }
247 +
248 + /// Held counts across every scope. See [`list_all`].
249 + pub fn counts_all(conn: &Connection) -> Result<HoldCounts> {
250 + let mut stmt = conn.prepare("SELECT state, COUNT(*) FROM sync_deferred GROUP BY state")?;
251 + tally(stmt.query_map([], read_tally)?)
252 + }
253 +
254 + fn tally<I>(rows: I) -> Result<HoldCounts>
255 + where
256 + I: Iterator<Item = rusqlite::Result<(String, i64)>>,
257 + {
258 + let mut out = HoldCounts::default();
211 259 for row in rows {
212 260 let (state, n) = row?;
213 261 let n = u64::try_from(n).unwrap_or(0);
M synckit-client/src/store/facade.rs +25
@@ -13,6 +13,7 @@
13 13 DbSource, clear_applying_remote, count_pending_changes, get_sync_state, get_sync_state_or,
14 14 set_sync_state,
15 15 };
16 + use super::deferred::{self, HeldEntry, HoldCounts};
16 17 use super::scheduler::{
17 18 SyncObserver, SyncState, backoff_delay, interval_elapsed, is_auth_lost,
18 19 is_subscription_required,
@@ -204,6 +205,30 @@
204 205 self.blocking(count_pending_changes).await
205 206 }
206 207
208 + /// How many remote changes the device is holding because they could not be
209 + /// applied, across every scope. See [`deferred`](super::deferred).
210 + pub async fn held_counts(&self) -> Result<HoldCounts> {
211 + self.blocking(deferred::counts_all).await
212 + }
213 +
214 + /// The held changes themselves, newest first, across every scope. Each
215 + /// carries its cause and its payload, so a caller can list them in terms its
216 + /// user recognises.
217 + pub async fn held_entries(&self) -> Result<Vec<HeldEntry>> {
218 + self.blocking(deferred::list_all).await
219 + }
220 +
221 + /// Put a held change back in the retry queue with a fresh attempt budget.
222 + ///
223 + /// This is the per-row Retry: an entry that ran out of attempts, or one a
224 + /// user has unblocked by hand, gets another run at the next pull. Returns
225 + /// whether such an entry existed.
226 + pub async fn retry_held(&self, scope: &str, table: &str, row_id: &str) -> Result<bool> {
227 + let (scope, table, row_id) = (scope.to_string(), table.to_string(), row_id.to_string());
228 + self.blocking(move |conn| deferred::requeue(conn, &scope, &table, &row_id))
229 + .await
230 + }
231 +
207 232 async fn maybe_snapshot(&self) -> Result<()> {
208 233 let db = self.db.clone();
209 234 let schema = self.schema.clone();