Skip to main content

max / goingson

fix(sync,backup): restore recurrence_rule on apply + transactional verbatim restore Two Axis-1 data-integrity fixes from ultra-fuzz Run #26. UF-1 (CRITICAL) + CHRONIC-C: sync apply dropped recurrence_rule. The 045/050 triggers emit recurrence_rule into the changelog for tasks and events, but apply.rs::table_columns() omitted it, so on pull the rich recurrence rule arrived NULL and degraded to the legacy Daily/Weekly/Monthly string -- silent cross-device corruption of "2nd Tuesday of every month"-style schedules. Add recurrence_rule to both whitelists and make the drift class uncompilable: table_columns() now reads from a single const SYNCED_COLUMNS, and a new round-trip test (trigger_columns_match_whitelist) reads the live trigger DDL from a migrated DB and asserts every json_object key a changelog trigger emits equals the apply whitelist, for all 16 synced tables. This would have failed on the recurrence_rule gap and guards the whole "two hand-maintained lists drift" class going forward. Backup restore (SERIOUS): the per-repository restore ran each entity on its own pooled connection (no shared transaction -> permanent half-state on mid-restore failure) and rebuilt task sub-collections through the normal create helpers, losing subtask completion state and ordering and resetting annotation timestamps to now. Replace it with goingson_db_sqlite::restore_all: one all-or-nothing transaction over the shared pool, verbatim id- and field-preserving inserts for every row including sub-collections. The buggy core orchestration (restore_from_backup) is removed so there is a single restore path; core keeps only the RestoreInput/RestoreResult DTOs. Integration tests now drive restore_all and assert subtask completion/ordering and annotation timestamps survive verbatim, plus idempotent re-restore. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-15 23:19 UTC
Commit: 65b732a17ece5ab9bc3f47e3e2fa65ad3da7e60e
Parent: a8217d3
8 files changed, +762 insertions, -222 deletions
M crates/core/src/backup_restore.rs +7 -141
@@ -1,17 +1,13 @@
1 - //! Backup restore orchestration logic.
1 + //! Backup restore data-transfer types.
2 2 //!
3 - //! Contains the entity iteration and existence-check-before-create pattern
4 - //! for restoring data from backups. The command layer handles file I/O;
5 - //! this module handles the restore logic using repository trait objects.
3 + //! The parsed-backup [`RestoreInput`] and the [`RestoreResult`] tally. The actual
4 + //! restore is performed transactionally by the storage backend
5 + //! (`goingson_db_sqlite::restore_all`) so the whole merge is one all-or-nothing unit
6 + //! and sub-collections are written verbatim; this module only defines the types that
7 + //! cross the command -> storage boundary.
6 8
7 - use crate::id_types::UserId;
8 -
9 - use crate::error::CoreError;
10 9 use crate::models::{Email, Event, Project, Task};
11 - use crate::{
12 - Contact, NewContactCustomField, NewContactEmail, NewContactPhone, NewSocialHandle,
13 - };
14 - use crate::repository::{ContactRepository, EmailRepository, EventRepository, ProjectRepository, TaskRepository};
10 + use crate::Contact;
15 11
16 12 /// Result of a restore operation.
17 13 #[derive(Debug, Default)]
@@ -41,136 +37,6 @@ pub struct RestoreInput {
41 37 pub contacts: Vec<Contact>,
42 38 }
43 39
44 - /// Restores entities from backup data, skipping those that already exist.
45 - ///
46 - /// Entities are inserted **verbatim, preserving their original UUIDs** (and
47 - /// status / completed_at / metadata) via the repositories' `restore` methods.
48 - /// Because IDs are preserved, the existence checks below find previously-restored
49 - /// rows, so restoring the same backup twice is idempotent — it does not duplicate
50 - /// the dataset. This is a merge operation: existing data is preserved, and the
51 - /// backup's foreign-key references (project_id, linked_task_id) stay valid
52 - /// without remapping.
53 - pub async fn restore_from_backup(
54 - user_id: UserId,
55 - input: &RestoreInput,
56 - projects: &dyn ProjectRepository,
57 - tasks: &dyn TaskRepository,
58 - events: &dyn EventRepository,
59 - emails: &dyn EmailRepository,
60 - contacts: &dyn ContactRepository,
61 - ) -> Result<RestoreResult, CoreError> {
62 - let mut result = RestoreResult::default();
63 -
64 - // Projects — preserve id + created_at + status.
65 - for project in &input.projects {
66 - if projects.get_by_id(project.id, user_id).await?.is_none() {
67 - projects.restore(user_id, project).await?;
68 - result.projects_restored += 1;
69 - }
70 - }
71 -
72 - // Tasks — preserve id + status + completed_at + all scheduling metadata.
73 - // project_id references stay valid because project ids are preserved.
74 - for task in &input.tasks {
75 - if tasks.get_by_id(task.id, user_id).await?.is_none() {
76 - tasks.restore(user_id, task).await?;
77 - result.tasks_restored += 1;
78 -
79 - // Restore annotations
80 - for annotation in &task.annotations {
81 - if tasks.add_annotation(task.id, user_id, &annotation.note).await?.is_some() {
82 - result.annotations_restored += 1;
83 - }
84 - }
85 -
86 - // Restore subtasks (text-only; linked subtasks handled in second pass)
87 - for subtask in &task.subtasks {
88 - if subtask.linked_task_id.is_none()
89 - && tasks.add_subtask(task.id, user_id, &subtask.text).await?.is_some()
90 - {
91 - result.subtasks_restored += 1;
92 - }
93 - }
94 - }
95 - }
96 -
97 - // Second pass: restore subtasks with linked_task_id (requires all tasks to
98 - // exist). IDs are preserved, so parent and linked ids are used directly.
99 - for task in &input.tasks {
100 - for subtask in &task.subtasks {
101 - if let Some(linked_id) = subtask.linked_task_id {
102 - if tasks.add_subtask_link(task.id, user_id, linked_id).await.ok().flatten().is_some() {
103 - result.subtasks_restored += 1;
104 - }
105 - }
106 - }
107 - }
108 -
109 - // Events — preserve id + full metadata (block_type, external_*, snoozed,
110 - // reminders, recurrence_parent_id).
111 - for event in &input.events {
112 - if events.get_by_id(event.id, user_id).await?.is_none() {
113 - events.restore(user_id, event).await?;
114 - result.events_restored += 1;
115 - }
116 - }
117 -
118 - // Emails — dedupe by message_id when present, else by preserved id.
119 - for email in &input.emails {
120 - let exists = if let Some(ref msg_id) = email.message_id {
121 - emails.exists_by_message_id(user_id, msg_id).await?
122 - } else {
123 - emails.get_by_id(email.id, user_id).await?.is_some()
124 - };
125 -
126 - if !exists {
127 - emails.restore(user_id, email).await?;
128 - result.emails_restored += 1;
129 - }
130 - }
131 -
132 - // Contacts — preserve id + created_at/updated_at; restore sub-collections.
133 - for contact in &input.contacts {
134 - if contacts.get_by_id(contact.id, user_id).await?.is_none() {
135 - contacts.restore(user_id, contact).await?;
136 - result.contacts_restored += 1;
137 -
138 - // Restore sub-collections (fresh ids; gated by the parent existence
139 - // check above so a second restore does not re-add them).
140 - for email in &contact.emails {
141 - let _ = contacts.add_email(contact.id, user_id, NewContactEmail {
142 - address: email.address.clone(),
143 - label: email.label.clone(),
144 - is_primary: email.is_primary,
145 - }).await;
146 - }
147 - for phone in &contact.phones {
148 - let _ = contacts.add_phone(contact.id, user_id, NewContactPhone {
149 - number: phone.number.clone(),
150 - label: phone.label.clone(),
151 - is_primary: phone.is_primary,
152 - }).await;
153 - }
154 - for handle in &contact.social_handles {
155 - let _ = contacts.add_social_handle(contact.id, user_id, NewSocialHandle {
156 - platform: handle.platform.clone(),
157 - handle: handle.handle.clone(),
158 - url: handle.url.clone(),
159 - }).await;
160 - }
161 - for field in &contact.custom_fields {
162 - let _ = contacts.add_custom_field(contact.id, user_id, NewContactCustomField {
163 - label: field.label.clone(),
164 - value: field.value.clone(),
165 - url: field.url.clone(),
166 - }).await;
167 - }
168 - }
169 - }
170 -
171 - Ok(result)
172 - }
173 -
174 40 #[cfg(test)]
175 41 mod tests {
176 42 use super::*;
M crates/db-sqlite/src/lib.rs +2
@@ -71,3 +71,5 @@ pub use repository::{
71 71 SqliteSyncAccountRepository,
72 72 SqliteWeeklyReviewRepository,
73 73 };
74 +
75 + pub use repository::restore::restore_all;
M crates/db-sqlite/src/repository/mod.rs +1
@@ -8,6 +8,7 @@ mod backup_settings_repo;
8 8 mod contact_repo;
9 9 mod daily_note_repo;
10 10 mod project_repo;
11 + pub mod restore;
11 12 mod annotation_repo;
12 13 mod subtask_repo;
13 14 mod task_repo;
A crates/db-sqlite/src/repository/restore.rs +431
@@ -0,0 +1,431 @@
1 + //! Transactional backup restore.
2 + //!
3 + //! A single all-or-nothing restore over one SQLite transaction: either the entire
4 + //! backup is merged into the database, or nothing is. This replaces the previous
5 + //! per-repository orchestration in `goingson_core::backup_restore`, which ran each
6 + //! entity through a separate pooled connection (so a mid-restore failure left a
7 + //! permanent half-state) and rebuilt task sub-collections through the normal create
8 + //! helpers -- losing subtask completion state and ordering, and annotation
9 + //! timestamps (ultra-fuzz Run #26 SERIOUS finding).
10 + //!
11 + //! Restore is a *merge*: every row is inserted verbatim with `INSERT OR IGNORE`,
12 + //! preserving its original id and every field, and existing rows are left untouched.
13 + //! Because ids are preserved, restoring the same backup twice is idempotent. The sync
14 + //! changelog triggers fire as normal, so restored rows propagate to other devices.
15 +
16 + use goingson_core::backup_restore::{RestoreInput, RestoreResult};
17 + use goingson_core::{CoreError, DbValue, Result, UserId};
18 + use sqlx::{Sqlite, SqlitePool, Transaction};
19 +
20 + use crate::utils::{format_datetime, format_datetime_opt};
21 +
22 + /// Restore a backup into the database as a single transaction.
23 + ///
24 + /// On any error the transaction is dropped without committing, so the database is
25 + /// left exactly as it was before the call.
26 + pub async fn restore_all(
27 + pool: &SqlitePool,
28 + user_id: UserId,
29 + input: &RestoreInput,
30 + ) -> Result<RestoreResult> {
31 + let mut tx = pool.begin().await.map_err(CoreError::database)?;
32 + let mut result = RestoreResult::default();
33 +
34 + restore_projects(&mut tx, user_id, input, &mut result).await?;
35 + restore_tasks_with_children(&mut tx, user_id, input, &mut result).await?;
36 + restore_events(&mut tx, user_id, input, &mut result).await?;
37 + restore_emails(&mut tx, user_id, input, &mut result).await?;
38 + restore_contacts_with_children(&mut tx, user_id, input, &mut result).await?;
39 +
40 + tx.commit().await.map_err(CoreError::database)?;
41 + Ok(result)
42 + }
43 +
44 + async fn restore_projects(
45 + tx: &mut Transaction<'_, Sqlite>,
46 + user_id: UserId,
47 + input: &RestoreInput,
48 + result: &mut RestoreResult,
49 + ) -> Result<()> {
50 + for project in &input.projects {
51 + let affected = sqlx::query(
52 + "INSERT OR IGNORE INTO projects (id, user_id, name, description, project_type, status, created_at) \
53 + VALUES (?, ?, ?, ?, ?, ?, ?)",
54 + )
55 + .bind(project.id.to_string())
56 + .bind(user_id.to_string())
57 + .bind(&project.name)
58 + .bind(&project.description)
59 + .bind(project.project_type.db_value())
60 + .bind(project.status.db_value())
61 + .bind(format_datetime(&project.created_at))
62 + .execute(&mut **tx)
63 + .await
64 + .map_err(CoreError::database)?
65 + .rows_affected();
66 + if affected > 0 {
67 + result.projects_restored += 1;
68 + }
69 + }
70 + Ok(())
71 + }
72 +
73 + async fn restore_tasks_with_children(
74 + tx: &mut Transaction<'_, Sqlite>,
75 + user_id: UserId,
76 + input: &RestoreInput,
77 + result: &mut RestoreResult,
78 + ) -> Result<()> {
79 + // First pass: task row + annotations + text-only subtasks. Children are restored
80 + // only when the parent task is newly inserted (a re-restore skips them).
81 + for task in &input.tasks {
82 + let tags_json = serde_json::to_string(&task.tags).unwrap_or_else(|_| "[]".to_string());
83 + let recurrence_rule_json = task
84 + .recurrence_rule
85 + .as_ref()
86 + .map(|r| serde_json::to_string(r).unwrap_or_default());
87 +
88 + let affected = sqlx::query(
89 + "INSERT OR IGNORE INTO tasks (\
90 + id, user_id, project_id, contact_id, milestone_id, description, status, \
91 + priority, due, tags, urgency, recurrence, recurrence_rule, recurrence_parent_id, \
92 + source_email_id, snoozed_until, waiting_for_response, waiting_since, expected_response_date, \
93 + scheduled_start, scheduled_duration, estimated_minutes, actual_minutes, \
94 + created_at, completed_at, is_focus, focus_set_at\
95 + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
96 + )
97 + .bind(task.id.to_string())
98 + .bind(user_id.to_string())
99 + .bind(task.project_id.map(|p| p.to_string()))
100 + .bind(task.contact_id.map(|c| c.to_string()))
101 + .bind(task.milestone_id.map(|m| m.to_string()))
102 + .bind(&task.description)
103 + .bind(task.status.db_value())
104 + .bind(task.priority.db_value())
105 + .bind(format_datetime_opt(task.due))
106 + .bind(&tags_json)
107 + .bind(task.urgency)
108 + .bind(task.recurrence.db_value())
109 + .bind(&recurrence_rule_json)
110 + .bind(task.recurrence_parent_id.map(|p| p.to_string()))
111 + .bind(task.source_email_id.map(|e| e.to_string()))
112 + .bind(format_datetime_opt(task.snoozed_until))
113 + .bind(if task.waiting_for_response { 1 } else { 0 })
114 + .bind(format_datetime_opt(task.waiting_since))
115 + .bind(format_datetime_opt(task.expected_response_date))
116 + .bind(format_datetime_opt(task.scheduled_start))
117 + .bind(task.scheduled_duration)
118 + .bind(task.estimated_minutes)
119 + .bind(task.actual_minutes)
120 + .bind(format_datetime(&task.created_at))
121 + .bind(task.completed_at.map(|d| format_datetime(&d)))
122 + .bind(if task.is_focus { 1 } else { 0 })
123 + .bind(task.focus_set_at.map(|d| format_datetime(&d)))
124 + .execute(&mut **tx)
125 + .await
126 + .map_err(CoreError::database)?
127 + .rows_affected();
128 +
129 + if affected == 0 {
130 + continue;
131 + }
132 + result.tasks_restored += 1;
133 +
134 + // Annotations -- preserve the original timestamp verbatim.
135 + for annotation in &task.annotations {
136 + let a = sqlx::query(
137 + "INSERT OR IGNORE INTO annotations (id, task_id, timestamp, note) VALUES (?, ?, ?, ?)",
138 + )
139 + .bind(annotation.id.to_string())
140 + .bind(task.id.to_string())
141 + .bind(format_datetime(&annotation.timestamp))
142 + .bind(&annotation.note)
143 + .execute(&mut **tx)
144 + .await
145 + .map_err(CoreError::database)?
146 + .rows_affected();
147 + if a > 0 {
148 + result.annotations_restored += 1;
149 + }
150 + }
151 +
152 + // Text-only subtasks -- preserve is_completed and position verbatim.
153 + // Linked subtasks are deferred to the second pass (their linked_task_id FK
154 + // requires every task row to exist first).
155 + for subtask in &task.subtasks {
156 + if subtask.linked_task_id.is_some() {
157 + continue;
158 + }
159 + let s = sqlx::query(
160 + "INSERT OR IGNORE INTO subtasks (id, task_id, text, is_completed, position) \
161 + VALUES (?, ?, ?, ?, ?)",
162 + )
163 + .bind(subtask.id.to_string())
164 + .bind(task.id.to_string())
165 + .bind(&subtask.text)
166 + .bind(if subtask.is_completed { 1 } else { 0 })
167 + .bind(subtask.position)
168 + .execute(&mut **tx)
169 + .await
170 + .map_err(CoreError::database)?
171 + .rows_affected();
172 + if s > 0 {
173 + result.subtasks_restored += 1;
174 + }
175 + }
176 + }
177 +
178 + // Second pass: linked subtasks, now that all task rows exist. Verbatim and
179 + // idempotent (INSERT OR IGNORE on the preserved id), so it is safe to run for
180 + // every task regardless of whether the parent was newly inserted.
181 + for task in &input.tasks {
182 + for subtask in &task.subtasks {
183 + let Some(linked_id) = subtask.linked_task_id else {
184 + continue;
185 + };
186 + let s = sqlx::query(
187 + "INSERT OR IGNORE INTO subtasks (id, task_id, text, is_completed, position, linked_task_id) \
188 + VALUES (?, ?, ?, ?, ?, ?)",
189 + )
190 + .bind(subtask.id.to_string())
191 + .bind(task.id.to_string())
192 + .bind(&subtask.text)
193 + .bind(if subtask.is_completed { 1 } else { 0 })
194 + .bind(subtask.position)
195 + .bind(linked_id.to_string())
196 + .execute(&mut **tx)
197 + .await
198 + .map_err(CoreError::database)?
199 + .rows_affected();
200 + if s > 0 {
201 + result.subtasks_restored += 1;
202 + }
203 + }
204 + }
205 +
206 + Ok(())
207 + }
208 +
209 + async fn restore_events(
210 + tx: &mut Transaction<'_, Sqlite>,
211 + user_id: UserId,
212 + input: &RestoreInput,
213 + result: &mut RestoreResult,
214 + ) -> Result<()> {
215 + for event in &input.events {
216 + let recurrence_rule_json = event
217 + .recurrence_rule
218 + .as_ref()
219 + .map(|r| serde_json::to_string(r).unwrap_or_default());
220 + let reminder_offsets_json = if event.reminder_offsets_seconds.is_empty() {
221 + None
222 + } else {
223 + Some(serde_json::to_string(&event.reminder_offsets_seconds).unwrap_or_default())
224 + };
225 +
226 + let affected = sqlx::query(
227 + "INSERT OR IGNORE INTO events (\
228 + id, user_id, project_id, title, description, start_time, end_time, location, \
229 + linked_task_id, recurrence, recurrence_rule, recurrence_parent_id, contact_id, \
230 + block_type, external_source, external_id, is_read_only, snoozed_until, reminder_offsets_seconds\
231 + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
232 + )
233 + .bind(event.id.to_string())
234 + .bind(user_id.to_string())
235 + .bind(event.project_id.map(|p| p.to_string()))
236 + .bind(&event.title)
237 + .bind(&event.description)
238 + .bind(format_datetime(&event.start_time))
239 + .bind(format_datetime_opt(event.end_time))
240 + .bind(&event.location)
241 + .bind(event.linked_task_id.map(|t| t.to_string()))
242 + .bind(event.recurrence.db_value())
243 + .bind(&recurrence_rule_json)
244 + .bind(event.recurrence_parent_id.map(|p| p.to_string()))
245 + .bind(event.contact_id.map(|c| c.to_string()))
246 + .bind(event.block_type.as_ref().map(|b| b.db_value()))
247 + .bind(&event.external_source)
248 + .bind(&event.external_id)
249 + .bind(if event.is_read_only { 1 } else { 0 })
250 + .bind(format_datetime_opt(event.snoozed_until))
251 + .bind(&reminder_offsets_json)
252 + .execute(&mut **tx)
253 + .await
254 + .map_err(CoreError::database)?
255 + .rows_affected();
256 + if affected > 0 {
257 + result.events_restored += 1;
258 + }
259 + }
260 + Ok(())
261 + }
262 +
263 + async fn restore_emails(
264 + tx: &mut Transaction<'_, Sqlite>,
265 + user_id: UserId,
266 + input: &RestoreInput,
267 + result: &mut RestoreResult,
268 + ) -> Result<()> {
269 + for email in &input.emails {
270 + // Dedupe by message_id when present (the same message can have a different
271 + // row id across devices); otherwise the preserved id + INSERT OR IGNORE
272 + // makes the restore idempotent.
273 + if let Some(msg_id) = &email.message_id {
274 + let existing: (i64,) = sqlx::query_as(
275 + "SELECT COUNT(*) FROM emails WHERE user_id = ? AND message_id = ?",
276 + )
277 + .bind(user_id.to_string())
278 + .bind(msg_id)
279 + .fetch_one(&mut **tx)
280 + .await
281 + .map_err(CoreError::database)?;
282 + if existing.0 > 0 {
283 + continue;
284 + }
285 + }
286 +
287 + let labels_json =
288 + serde_json::to_string(&email.labels).unwrap_or_else(|_| "[]".to_string());
289 + let affected = sqlx::query(
290 + "INSERT OR IGNORE INTO emails (\
291 + id, user_id, project_id, from_address, to_address, subject, body, html_body, \
292 + is_read, is_archived, received_at, message_id, in_reply_to, thread_id, is_outgoing, \
293 + labels, is_draft, cc_address, bcc_address, snoozed_until, waiting_for_response, \
294 + waiting_since, expected_response_date\
295 + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
296 + )
297 + .bind(email.id.to_string())
298 + .bind(user_id.to_string())
299 + .bind(email.project_id.map(|p| p.to_string()))
300 + .bind(&email.from)
301 + .bind(&email.to)
302 + .bind(&email.subject)
303 + .bind(&email.body)
304 + .bind(&email.html_body)
305 + .bind(if email.is_read { 1 } else { 0 })
306 + .bind(if email.is_archived { 1 } else { 0 })
307 + .bind(format_datetime(&email.received_at))
308 + .bind(&email.message_id)
309 + .bind(&email.in_reply_to)
310 + .bind(&email.thread_id)
311 + .bind(if email.is_outgoing { 1 } else { 0 })
312 + .bind(&labels_json)
313 + .bind(if email.is_draft { 1 } else { 0 })
314 + .bind(&email.cc_address)
315 + .bind(&email.bcc_address)
316 + .bind(format_datetime_opt(email.snoozed_until))
317 + .bind(if email.waiting_for_response { 1 } else { 0 })
318 + .bind(format_datetime_opt(email.waiting_since))
319 + .bind(format_datetime_opt(email.expected_response_date))
320 + .execute(&mut **tx)
321 + .await
322 + .map_err(CoreError::database)?
323 + .rows_affected();
324 + if affected > 0 {
325 + result.emails_restored += 1;
326 + }
327 + }
328 + Ok(())
329 + }
330 +
331 + async fn restore_contacts_with_children(
332 + tx: &mut Transaction<'_, Sqlite>,
333 + user_id: UserId,
334 + input: &RestoreInput,
335 + result: &mut RestoreResult,
336 + ) -> Result<()> {
337 + for contact in &input.contacts {
338 + let tags_json = serde_json::to_string(&contact.tags).unwrap_or_else(|_| "[]".to_string());
339 + let birthday_str = contact.birthday.map(|d| d.format("%Y-%m-%d").to_string());
340 +
341 + let affected = sqlx::query(
342 + "INSERT OR IGNORE INTO contacts (\
343 + id, user_id, display_name, nickname, company, title, notes, tags, birthday, \
344 + timezone, external_source, external_id, is_implicit, created_at, updated_at\
345 + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
346 + )
347 + .bind(contact.id.to_string())
348 + .bind(user_id.to_string())
349 + .bind(&contact.display_name)
350 + .bind(&contact.nickname)
351 + .bind(&contact.company)
352 + .bind(&contact.title)
353 + .bind(&contact.notes)
354 + .bind(&tags_json)
355 + .bind(&birthday_str)
356 + .bind(&contact.timezone)
357 + .bind(&contact.external_source)
358 + .bind(&contact.external_id)
359 + .bind(if contact.is_implicit { 1 } else { 0 })
360 + .bind(format_datetime(&contact.created_at))
361 + .bind(format_datetime(&contact.updated_at))
362 + .execute(&mut **tx)
363 + .await
364 + .map_err(CoreError::database)?
365 + .rows_affected();
366 +
367 + if affected == 0 {
368 + continue;
369 + }
370 + result.contacts_restored += 1;
371 +
372 + // Sub-collections, verbatim with their preserved ids.
373 + for email in &contact.emails {
374 + sqlx::query(
375 + "INSERT OR IGNORE INTO contact_emails (id, contact_id, address, label, is_primary) \
376 + VALUES (?, ?, ?, ?, ?)",
377 + )
378 + .bind(email.id.to_string())
379 + .bind(contact.id.to_string())
380 + .bind(&email.address)
381 + .bind(&email.label)
382 + .bind(if email.is_primary { 1 } else { 0 })
383 + .execute(&mut **tx)
384 + .await
385 + .map_err(CoreError::database)?;
386 + }
387 + for phone in &contact.phones {
388 + sqlx::query(
389 + "INSERT OR IGNORE INTO contact_phones (id, contact_id, number, label, is_primary) \
390 + VALUES (?, ?, ?, ?, ?)",
391 + )
392 + .bind(phone.id.to_string())
393 + .bind(contact.id.to_string())
394 + .bind(&phone.number)
395 + .bind(&phone.label)
396 + .bind(if phone.is_primary { 1 } else { 0 })
397 + .execute(&mut **tx)
398 + .await
399 + .map_err(CoreError::database)?;
400 + }
401 + for handle in &contact.social_handles {
402 + sqlx::query(
403 + "INSERT OR IGNORE INTO contact_social_handles (id, contact_id, platform, handle, url) \
404 + VALUES (?, ?, ?, ?, ?)",
405 + )
406 + .bind(handle.id.to_string())
407 + .bind(contact.id.to_string())
408 + .bind(&handle.platform)
409 + .bind(&handle.handle)
410 + .bind(&handle.url)
411 + .execute(&mut **tx)
412 + .await
413 + .map_err(CoreError::database)?;
414 + }
415 + for field in &contact.custom_fields {
416 + sqlx::query(
417 + "INSERT OR IGNORE INTO contact_custom_fields (id, contact_id, label, value, url) \
418 + VALUES (?, ?, ?, ?, ?)",
419 + )
420 + .bind(field.id.to_string())
421 + .bind(contact.id.to_string())
422 + .bind(&field.label)
423 + .bind(&field.value)
424 + .bind(&field.url)
425 + .execute(&mut **tx)
426 + .await
427 + .map_err(CoreError::database)?;
428 + }
429 + }
430 + Ok(())
431 + }
M crates/db-sqlite/tests/restore_tests.rs +135 -19
@@ -1,18 +1,19 @@
1 - //! Integration tests for backup restore (GO-1 launch fixes).
1 + //! Integration tests for transactional backup restore.
2 2 //!
3 - //! Covers: completed-status + completed_at preservation, event-metadata
4 - //! round-trip (the fields the normal create path never sets), and idempotent
5 - //! re-restore (restoring the same backup twice must not duplicate the dataset).
3 + //! Covers: completed-status + completed_at preservation, event-metadata round-trip,
4 + //! idempotent re-restore (restoring the same backup twice must not duplicate the
5 + //! dataset), and verbatim sub-collection fidelity -- subtask completion state and
6 + //! ordering, and annotation timestamps -- which the old create-helper restore path
7 + //! silently dropped (ultra-fuzz Run #26 SERIOUS finding).
6 8
7 9 mod common;
8 10
9 - use goingson_core::backup_restore::{restore_from_backup, RestoreInput};
11 + use goingson_core::backup_restore::RestoreInput;
10 12 use goingson_core::{
11 13 EventRepository, NewEvent, NewProject, NewTask, ProjectRepository, TaskRepository, TaskStatus,
12 14 };
13 15 use goingson_db_sqlite::{
14 - SqliteContactRepository, SqliteEmailRepository, SqliteEventRepository, SqliteProjectRepository,
15 - SqliteTaskRepository,
16 + restore_all, SqliteEventRepository, SqliteProjectRepository, SqliteTaskRepository,
16 17 };
17 18
18 19 #[tokio::test]
@@ -22,9 +23,6 @@ async fn restore_preserves_completed_status_and_is_idempotent() {
22 23
23 24 let projects = SqliteProjectRepository::new(pool.clone());
24 25 let tasks = SqliteTaskRepository::new(pool.clone());
25 - let events = SqliteEventRepository::new(pool.clone());
26 - let emails = SqliteEmailRepository::new(pool.clone());
27 - let contacts = SqliteContactRepository::new(pool.clone());
28 26
29 27 // Manufacture a project and a COMPLETED task.
30 28 let project = projects
@@ -71,9 +69,7 @@ async fn restore_preserves_completed_status_and_is_idempotent() {
71 69 };
72 70
73 71 // First restore recreates with status + completed_at + id preserved.
74 - let r1 = restore_from_backup(user_id, &input, &projects, &tasks, &events, &emails, &contacts)
75 - .await
76 - .unwrap();
72 + let r1 = restore_all(&pool, user_id, &input).await.unwrap();
77 73 assert_eq!(r1.projects_restored, 1);
78 74 assert_eq!(r1.tasks_restored, 1);
79 75 let restored = tasks
@@ -92,10 +88,8 @@ async fn restore_preserves_completed_status_and_is_idempotent() {
92 88 );
93 89 assert_eq!(restored.id, task.id, "original id preserved");
94 90
95 - // Second restore is a no-op — no duplication.
96 - let r2 = restore_from_backup(user_id, &input, &projects, &tasks, &events, &emails, &contacts)
97 - .await
98 - .unwrap();
91 + // Second restore is a no-op -- no duplication.
92 + let r2 = restore_all(&pool, user_id, &input).await.unwrap();
99 93 assert_eq!(r2.projects_restored, 0);
100 94 assert_eq!(r2.tasks_restored, 0);
101 95 assert_eq!(
@@ -111,6 +105,120 @@ async fn restore_preserves_completed_status_and_is_idempotent() {
111 105 }
112 106
113 107 #[tokio::test]
108 + async fn restore_preserves_subtask_completion_ordering_and_annotation_timestamp() {
109 + let pool = common::setup_test_db().await;
110 + let user_id = common::create_test_user(&pool).await;
111 + let tasks = SqliteTaskRepository::new(pool.clone());
112 +
113 + let task = tasks
114 + .create(user_id, NewTask::builder("Ship the thing").build())
115 + .await
116 + .unwrap();
117 +
118 + // Two subtasks; the second is completed. Positions 0 and 1.
119 + let first = tasks
120 + .add_subtask(task.id, user_id, "do A")
121 + .await
122 + .unwrap()
123 + .unwrap();
124 + let second = tasks
125 + .add_subtask(task.id, user_id, "do B")
126 + .await
127 + .unwrap()
128 + .unwrap();
129 + let second = tasks
130 + .toggle_subtask(second.id, user_id)
131 + .await
132 + .unwrap()
133 + .unwrap();
134 + assert!(second.is_completed, "precondition: second subtask completed");
135 + assert_eq!(first.position, 0);
136 + assert_eq!(second.position, 1);
137 +
138 + // One annotation, back-dated to a distinct timestamp so a reset-to-now
139 + // regression is observable.
140 + let annotation = tasks
141 + .add_annotation(task.id, user_id, "an old note")
142 + .await
143 + .unwrap()
144 + .unwrap();
145 + sqlx::query("UPDATE annotations SET timestamp = ? WHERE id = ?")
146 + .bind("2020-01-01 00:00:00")
147 + .bind(annotation.id.to_string())
148 + .execute(&pool)
149 + .await
150 + .unwrap();
151 +
152 + // The backup payload as it would be exported (fully hydrated task).
153 + let backup_task = tasks.get_by_id(task.id, user_id).await.unwrap().unwrap();
154 + assert_eq!(backup_task.subtasks.len(), 2);
155 + assert_eq!(backup_task.annotations.len(), 1);
156 +
157 + // Lose the task (cascades subtasks + annotations).
158 + sqlx::query("DELETE FROM tasks WHERE id = ?")
159 + .bind(task.id.to_string())
160 + .execute(&pool)
161 + .await
162 + .unwrap();
163 +
164 + let input = RestoreInput {
165 + projects: vec![],
166 + tasks: vec![backup_task.clone()],
167 + events: vec![],
168 + emails: vec![],
169 + contacts: vec![],
170 + };
171 + let r = restore_all(&pool, user_id, &input).await.unwrap();
172 + assert_eq!(r.tasks_restored, 1);
173 + assert_eq!(r.subtasks_restored, 2);
174 + assert_eq!(r.annotations_restored, 1);
175 +
176 + let restored = tasks.get_by_id(task.id, user_id).await.unwrap().unwrap();
177 +
178 + // Subtask completion state and ordering survive verbatim.
179 + let restored_first = restored
180 + .subtasks
181 + .iter()
182 + .find(|s| s.id == first.id)
183 + .expect("first subtask restored");
184 + let restored_second = restored
185 + .subtasks
186 + .iter()
187 + .find(|s| s.id == second.id)
188 + .expect("second subtask restored");
189 + assert!(
190 + !restored_first.is_completed,
191 + "incomplete subtask must stay incomplete"
192 + );
193 + assert!(
194 + restored_second.is_completed,
195 + "completed subtask must stay completed (regression: create-helper reset it)"
196 + );
197 + assert_eq!(restored_first.position, 0, "subtask order preserved");
198 + assert_eq!(restored_second.position, 1, "subtask order preserved");
199 +
200 + // Annotation timestamp survives verbatim (not reset to restore-time now).
201 + let restored_annotation = &restored.annotations[0];
202 + assert_eq!(
203 + restored_annotation.id, annotation.id,
204 + "annotation id preserved"
205 + );
206 + assert_eq!(
207 + restored_annotation.timestamp.format("%Y").to_string(),
208 + "2020",
209 + "annotation timestamp must survive restore (regression: reset to now)"
210 + );
211 +
212 + // Re-restore is a no-op: no duplicate children.
213 + let r2 = restore_all(&pool, user_id, &input).await.unwrap();
214 + assert_eq!(r2.tasks_restored, 0);
215 + assert_eq!(r2.subtasks_restored, 0);
216 + let again = tasks.get_by_id(task.id, user_id).await.unwrap().unwrap();
217 + assert_eq!(again.subtasks.len(), 2, "no duplicate subtasks");
218 + assert_eq!(again.annotations.len(), 1, "no duplicate annotations");
219 + }
220 +
221 + #[tokio::test]
114 222 async fn restore_event_preserves_external_metadata() {
115 223 let pool = common::setup_test_db().await;
116 224 let user_id = common::create_test_user(&pool).await;
@@ -135,13 +243,21 @@ async fn restore_event_preserves_external_metadata() {
135 243 let full = events.get_by_id(ev.id, user_id).await.unwrap().unwrap();
136 244 assert_eq!(full.external_source.as_deref(), Some("google"));
137 245
138 - // Lose it, then restore verbatim.
246 + // Lose it, then restore verbatim through the transactional path.
139 247 sqlx::query("DELETE FROM events WHERE id = ?")
140 248 .bind(ev.id.to_string())
141 249 .execute(&pool)
142 250 .await
143 251 .unwrap();
144 - events.restore(user_id, &full).await.unwrap();
252 + let input = RestoreInput {
253 + projects: vec![],
254 + tasks: vec![],
255 + events: vec![full.clone()],
256 + emails: vec![],
257 + contacts: vec![],
258 + };
259 + let r = restore_all(&pool, user_id, &input).await.unwrap();
260 + assert_eq!(r.events_restored, 1);
145 261
146 262 let restored = events
147 263 .get_by_id(ev.id, user_id)
M src-tauri/src/commands/export.rs +4 -9
@@ -392,15 +392,10 @@ pub async fn restore_backup(
392 392 contacts: export.contacts,
393 393 };
394 394
395 - let result = goingson_core::backup_restore::restore_from_backup(
396 - DESKTOP_USER_ID,
397 - &input,
398 - state.projects.as_ref(),
399 - state.tasks.as_ref(),
400 - state.events.as_ref(),
401 - state.emails.as_ref(),
402 - state.contacts.as_ref(),
403 - ).await?;
395 + // Single all-or-nothing transaction over the shared pool: a mid-restore failure
396 + // rolls back cleanly, and sub-collections (subtask completion/order, annotation
397 + // timestamps) are restored verbatim.
398 + let result = goingson_db_sqlite::restore_all(&state.pool, DESKTOP_USER_ID, &input).await?;
404 399
405 400 Ok(RestoreResponse {
406 401 projects_restored: result.projects_restored,
M src-tauri/src/sync_service/apply.rs +67 -53
@@ -20,61 +20,75 @@ use sqlx::SqliteConnection;
20 20
21 21 use super::EMAIL_ACCOUNT_SYNC_COLS;
22 22
23 + /// The single source of truth for which columns each table syncs.
24 + ///
25 + /// CHRONIC-C invariant: every column a sync changelog trigger emits into
26 + /// `sync_changelog.data` (via `json_object(...)`) must appear here, and every column
27 + /// here must be emitted by the trigger. `apply_upsert` and `create_initial_snapshot`
28 + /// both read their column list from this table, so the apply side has exactly one list.
29 + /// The trigger side lives in raw migration SQL (`migrations/sqlite/*.sql`); the
30 + /// `trigger_columns_match_whitelist` round-trip test reads the live trigger DDL and
31 + /// asserts the emitted `json_object` keys equal this list for every table, so the two
32 + /// can never silently diverge again. This is the bug behind UF-1: the trigger emitted
33 + /// `recurrence_rule` but it was missing here, so it arrived NULL on every pull.
34 + pub(crate) const SYNCED_COLUMNS: &[(&str, &[&str])] = &[
35 + ("projects", &[
36 + "id", "name", "description", "project_type", "status", "created_at", "user_id",
37 + ]),
38 + ("tasks", &[
39 + "id", "project_id", "description", "status", "priority", "due", "tags", "urgency",
40 + "recurrence", "recurrence_rule", "created_at", "user_id", "recurrence_parent_id",
41 + "source_email_id", "snoozed_until", "waiting_for_response", "waiting_since",
42 + "expected_response_date", "scheduled_start", "scheduled_duration", "is_focus",
43 + "focus_set_at", "contact_id", "milestone_id", "completed_at", "estimated_minutes",
44 + "actual_minutes",
45 + ]),
46 + ("events", &[
47 + "id", "project_id", "title", "description", "start_time", "end_time", "location",
48 + "user_id", "linked_task_id", "recurrence", "recurrence_parent_id", "recurrence_rule",
49 + "contact_id", "block_type", "external_source", "external_id", "is_read_only",
50 + "snoozed_until", "reminder_offsets_seconds",
51 + ]),
52 + ("contacts", &[
53 + "id", "user_id", "display_name", "nickname", "company", "title", "notes", "tags",
54 + "birthday", "timezone", "external_source", "external_id", "created_at", "updated_at",
55 + ]),
56 + ("contact_emails", &["id", "contact_id", "address", "label", "is_primary"]),
57 + ("contact_phones", &["id", "contact_id", "number", "label", "is_primary"]),
58 + ("contact_social_handles", &["id", "contact_id", "platform", "handle", "url"]),
59 + ("contact_custom_fields", &["id", "contact_id", "label", "value", "url"]),
60 + ("annotations", &["id", "task_id", "timestamp", "note"]),
61 + ("subtasks", &[
62 + "id", "task_id", "text", "is_completed", "position", "created_at", "linked_task_id",
63 + ]),
64 + ("milestones", &[
65 + "id", "user_id", "project_id", "name", "description", "position", "target_date",
66 + "status", "created_at",
67 + ]),
68 + ("time_sessions", &[
69 + "id", "task_id", "user_id", "started_at", "ended_at", "duration_minutes", "created_at",
70 + ]),
71 + ("attachments", &[
72 + "id", "user_id", "task_id", "project_id", "filename", "file_size", "mime_type",
73 + "blob_hash", "source_email_id", "created_at",
74 + ]),
75 + ("sync_accounts", &[
76 + "id", "user_id", "provider", "account_name", "email", "sync_calendars", "sync_contacts",
77 + "calendar_ids", "sync_interval_minutes", "enabled", "created_at",
78 + ]),
79 + ("email_accounts", EMAIL_ACCOUNT_SYNC_COLS),
80 + ("daily_notes", &[
81 + "id", "user_id", "note_date", "went_well", "could_improve", "is_reviewed", "reviewed_at",
82 + "created_at", "updated_at",
83 + ]),
84 + ];
85 +
23 86 /// Return the syncable column whitelist for a given table, or `None` if unknown.
24 87 pub(crate) fn table_columns(table: &str) -> Option<&'static [&'static str]> {
25 - match table {
26 - "projects" => Some(&[
27 - "id", "name", "description", "project_type", "status", "created_at", "user_id",
28 - ]),
29 - "tasks" => Some(&[
30 - "id", "project_id", "description", "status", "priority", "due", "tags", "urgency",
31 - "recurrence", "created_at", "user_id", "recurrence_parent_id", "source_email_id",
32 - "snoozed_until", "waiting_for_response", "waiting_since", "expected_response_date",
33 - "scheduled_start", "scheduled_duration", "is_focus", "focus_set_at", "contact_id",
34 - "milestone_id", "completed_at", "estimated_minutes", "actual_minutes",
35 - ]),
36 - "events" => Some(&[
37 - "id", "project_id", "title", "description", "start_time", "end_time", "location",
38 - "user_id", "linked_task_id", "recurrence", "recurrence_parent_id", "contact_id",
39 - "block_type", "external_source", "external_id", "is_read_only", "snoozed_until",
40 - "reminder_offsets_seconds",
41 - ]),
42 - "contacts" => Some(&[
43 - "id", "user_id", "display_name", "nickname", "company", "title", "notes", "tags",
44 - "birthday", "timezone", "external_source", "external_id", "created_at", "updated_at",
45 - ]),
46 - "contact_emails" => Some(&["id", "contact_id", "address", "label", "is_primary"]),
47 - "contact_phones" => Some(&["id", "contact_id", "number", "label", "is_primary"]),
48 - "contact_social_handles" => Some(&["id", "contact_id", "platform", "handle", "url"]),
49 - "contact_custom_fields" => Some(&["id", "contact_id", "label", "value", "url"]),
50 - "annotations" => Some(&["id", "task_id", "timestamp", "note"]),
51 - "subtasks" => Some(&[
52 - "id", "task_id", "text", "is_completed", "position", "created_at", "linked_task_id",
53 - ]),
54 - "milestones" => Some(&[
55 - "id", "user_id", "project_id", "name", "description", "position", "target_date",
56 - "status", "created_at",
57 - ]),
58 - "time_sessions" => Some(&[
59 - "id", "task_id", "user_id", "started_at", "ended_at", "duration_minutes",
60 - "created_at",
61 - ]),
62 - "attachments" => Some(&[
63 - "id", "user_id", "task_id", "project_id", "filename",
64 - "file_size", "mime_type", "blob_hash", "source_email_id", "created_at",
65 - ]),
66 - "sync_accounts" => Some(&[
67 - "id", "user_id", "provider", "account_name", "email",
68 - "sync_calendars", "sync_contacts", "calendar_ids",
69 - "sync_interval_minutes", "enabled", "created_at",
70 - ]),
71 - "email_accounts" => Some(EMAIL_ACCOUNT_SYNC_COLS),
72 - "daily_notes" => Some(&[
73 - "id", "user_id", "note_date", "went_well", "could_improve",
74 - "is_reviewed", "reviewed_at", "created_at", "updated_at",
75 - ]),
76 - _ => None,
77 - }
88 + SYNCED_COLUMNS
89 + .iter()
90 + .find(|(t, _)| *t == table)
91 + .map(|(_, cols)| *cols)
78 92 }
79 93
80 94 /// Apply an INSERT OR REPLACE for a remote change.
M src-tauri/src/sync_service/tests.rs +115
@@ -1606,3 +1606,118 @@ use sqlx::SqlitePool;
1606 1606 ).fetch_one(&pool).await.unwrap();
1607 1607 assert_eq!(ts_count.0, 1);
1608 1608 }
1609 +
1610 + // ── CHRONIC-C round-trip invariant (UF-1 regression guard) ──────────────────
1611 + //
1612 + // Every column a sync changelog trigger emits into `sync_changelog.data` must be
1613 + // in the apply-side whitelist (`apply::SYNCED_COLUMNS`), and every whitelisted
1614 + // column must be emitted by the trigger, for every synced table. This is read
1615 + // from the live trigger DDL in a migrated DB so the two lists cannot silently
1616 + // diverge again -- the exact failure mode of UF-1, where the trigger emitted
1617 + // `recurrence_rule` but the apply whitelist dropped it, so it arrived NULL on pull.
1618 +
1619 + /// First single-quoted literal after `VALUES (` in a trigger body -- the changelog
1620 + /// target table name.
1621 + fn extract_changelog_table(sql: &str) -> Option<String> {
1622 + let after_values = &sql[sql.find("VALUES")?..];
1623 + let after_paren = &after_values[after_values.find('(')? + 1..];
1624 + let start = after_paren.find('\'')? + 1;
1625 + let rest = &after_paren[start..];
1626 + let end = rest.find('\'')?;
1627 + Some(rest[..end].to_string())
1628 + }
1629 +
1630 + /// Keys of the `json_object(...)` payload: every single-quoted identifier that is
1631 + /// immediately followed by `, NEW.` or `, OLD.`. Scoped to the substring after
1632 + /// `json_object(`, so the table-name and op literals that precede it (which are
1633 + /// not followed by a column reference) are never captured.
1634 + fn extract_json_object_keys(sql: &str) -> std::collections::BTreeSet<String> {
1635 + let mut keys = std::collections::BTreeSet::new();
1636 + let Some(jidx) = sql.find("json_object(") else {
1637 + return keys;
1638 + };
1639 + let body = &sql[jidx..];
1640 + let bytes = body.as_bytes();
1641 + let is_ws = |b: u8| b == b' ' || b == b'\n' || b == b'\t' || b == b'\r';
1642 + let mut i = 0;
1643 + while i < bytes.len() {
1644 + if bytes[i] != b'\'' {
1645 + i += 1;
1646 + continue;
1647 + }
1648 + let key_start = i + 1;
1649 + let mut j = key_start;
1650 + while j < bytes.len() && bytes[j] != b'\'' {
1651 + j += 1;
1652 + }
1653 + if j >= bytes.len() {
1654 + break;
1655 + }
1656 + let key = &body[key_start..j];
1657 + let mut k = j + 1;
1658 + while k < bytes.len() && is_ws(bytes[k]) {
1659 + k += 1;
1660 + }
1661 + if k < bytes.len() && bytes[k] == b',' {
1662 + k += 1;
1663 + while k < bytes.len() && is_ws(bytes[k]) {
1664 + k += 1;
1665 + }
1666 + if body[k..].starts_with("NEW.") || body[k..].starts_with("OLD.") {
1667 + keys.insert(key.to_string());
1668 + }
1669 + }
1670 + i = j + 1;
1671 + }
1672 + keys
1673 + }
1674 +
1675 + #[tokio::test]
1676 + async fn trigger_columns_match_whitelist() {
1677 + use std::collections::{BTreeMap, BTreeSet};
1678 +
1679 + let pool = setup_test_db().await;
1680 +
1681 + let triggers: Vec<(String, String)> = sqlx::query_as(
1682 + "SELECT name, sql FROM sqlite_master \
1683 + WHERE type = 'trigger' AND sql LIKE '%sync_changelog%'",
1684 + )
1685 + .fetch_all(&pool)
1686 + .await
1687 + .expect("query changelog triggers");
1688 +
1689 + // table -> union of columns emitted by its INSERT/UPDATE triggers.
1690 + let mut emitted: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
1691 + for (name, sql) in &triggers {
1692 + let upper = sql.to_uppercase();
1693 + // DELETE triggers carry no row payload; only INSERT/UPDATE define columns.
1694 + if !(upper.contains("AFTER INSERT") || upper.contains("AFTER UPDATE")) {
1695 + continue;
1696 + }
1697 + let table = extract_changelog_table(sql)
1698 + .unwrap_or_else(|| panic!("trigger {name} has no VALUES ('<table>', ...) literal"));
1699 + let keys = extract_json_object_keys(sql);
1700 + assert!(!keys.is_empty(), "trigger {name} emitted no json_object keys");
1701 + emitted.entry(table).or_default().extend(keys);
1702 + }
1703 +
1704 + // Every synced table must have a trigger and match its whitelist exactly.
1705 + for (table, cols) in apply::SYNCED_COLUMNS {
1706 + let whitelist: BTreeSet<String> = cols.iter().map(|c| c.to_string()).collect();
1707 + let got = emitted.get(*table).unwrap_or_else(|| {
1708 + panic!("synced table `{table}` has no INSERT/UPDATE changelog trigger")
1709 + });
1710 + assert_eq!(
1711 + got, &whitelist,
1712 + "column drift for `{table}`: trigger emits {got:?}, apply whitelist is {whitelist:?}",
1713 + );
1714 + }
1715 +
1716 + // And no trigger may emit for a table absent from the whitelist (orphan).
1717 + for table in emitted.keys() {
1718 + assert!(
1719 + apply::table_columns(table).is_some(),
1720 + "trigger emits for `{table}` but it is not in SYNCED_COLUMNS",
1721 + );
1722 + }
1723 + }