Skip to main content

max / audiofiles

Resilience: drain sync push fully, roll back partial chop import, fail-open export validate F1: push_changes now loops over PUSH_BATCH_LIMIT batches until the backlog is drained (mirroring pull_changes), instead of one 500-row batch per cycle. This closed a silent cross-device divergence: on a large first-sync import, changelog retention could cull still-unpushed entries before the push side reached them. enforce_changelog_retention now returns RetentionOutcome {deleted, unpushed_dropped} and the scheduler raises a hard last_error when unpushed rows are dropped, rather than only logging. F6: import_chop is now two-stage — copy all slice blobs (content-addressed, idempotent) then create the {stem}_slices directory and links inside one transaction_core, so a mid-batch failure rolls the whole directory back. A RunDirGuard owns the temp dir and sweeps it on every exit, replacing the per-slice removal that leaked temps k+1..n on early return. F7: device-profile validate_sample distinguishes Ok(false) (deliberate reject) from Err (script fault); a faulting script now keeps the sample and logs loudly instead of the silent unwrap_or(false) that dropped files while reporting export success. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-02 21:39 UTC
Commit: f8e9072bc57f7521dfca5f292b13bfeb1138f03b
Parent: 398b48c
6 files changed, +180 insertions, -28 deletions
M crates/audiofiles-browser/src/backend/direct.rs +16 -2
@@ -313,8 +313,22 @@ impl DirectBackend {
313 313 let engine = self.plugin_registry.engine();
314 314 let mut keep = vec![true; items.len()];
315 315 for (i, info) in infos.into_iter().enumerate() {
316 - keep[i] = audiofiles_rhai::hooks::run_validate_sample(engine, ast, info)
317 - .unwrap_or(false);
316 + keep[i] = match audiofiles_rhai::hooks::run_validate_sample(engine, ast, info) {
317 + // A deliberate `false` from the script rejects the sample.
318 + Ok(decision) => decision,
319 + // A script *fault* (not a rejection) must not silently shrink
320 + // the export — previously `.unwrap_or(false)` dropped the file
321 + // and reported success. Fail open (keep the sample) and surface
322 + // the error loudly instead.
323 + Err(e) => {
324 + tracing::error!(
325 + profile = %profile_name,
326 + error = %e,
327 + "device-profile validate_sample script errored; keeping sample in export rather than dropping it silently"
328 + );
329 + true
330 + }
331 + };
318 332 }
319 333 let mut ki = 0;
320 334 items.retain(|_| { let k = keep[ki]; ki += 1; k });
M crates/audiofiles-core/src/forge/runner.rs +34 -9
@@ -167,18 +167,43 @@ pub fn import_chop(
167 167 staging: ChopStaging,
168 168 ) -> Result<ChopResult, CoreError> {
169 169 let dir_name = format!("{}_slices", staging.stem);
170 - let dir_node = vfs::create_directory(db, vfs_id, parent_id, &dir_name)?;
171 170
172 - let run_dir = staging.slices.first().and_then(|(_, p)| p.parent().map(Path::to_path_buf));
173 - let mut written = 0usize;
171 + let run_dir = staging
172 + .slices
173 + .first()
174 + .and_then(|(_, p)| p.parent().map(Path::to_path_buf));
175 +
176 + // Own the temp dir for the whole import: swept on every exit (success, error,
177 + // or panic) once its blobs have been copied into the content store. Replaces
178 + // the old per-slice remove + trailing `cleanup_run_dir`, which leaked temps
179 + // `k+1..n` whenever the loop bailed partway.
180 + let _run_dir_guard = run_dir.as_ref().map(|dir| RunDirGuard {
181 + dir: dir.clone(),
182 + armed: true,
183 + });
184 +
185 + // Stage 1: copy every slice blob into the content store. Content-addressed
186 + // and idempotent, so a failure here touches no VFS state — there is nothing
187 + // to roll back, and a retry reuses any blobs already written.
188 + let mut imported: Vec<(&str, String)> = Vec::with_capacity(staging.slices.len());
174 189 for (slice_name, temp_path) in &staging.slices {
175 - let import = store.import(temp_path, db);
176 - let _ = std::fs::remove_file(temp_path);
177 - let hash = import?;
178 - vfs::create_sample_link(db, vfs_id, Some(dir_node), slice_name, &hash)?;
179 - written += 1;
190 + let hash = store.import(temp_path, db)?;
191 + imported.push((slice_name.as_str(), hash));
180 192 }
181 - cleanup_run_dir(run_dir.as_deref());
193 +
194 + // Stage 2: create the "{stem}_slices" directory and link every slice in a
195 + // single transaction. A mid-batch failure (e.g. a name conflict on slice k)
196 + // rolls the whole directory back instead of leaving a half-populated
197 + // "{stem}_slices" tree the user has to clean up by hand.
198 + let (dir_node, written) = db.transaction_core(|_tx| {
199 + let dir_node = vfs::create_directory(db, vfs_id, parent_id, &dir_name)?;
200 + let mut written = 0usize;
201 + for (slice_name, hash) in &imported {
202 + vfs::create_sample_link(db, vfs_id, Some(dir_node), slice_name, hash)?;
203 + written += 1;
204 + }
205 + Ok((dir_node, written))
206 + })?;
182 207
183 208 Ok(ChopResult { dir_node, slice_count: written })
184 209 }
M crates/audiofiles-sync/src/scheduler.rs +13 -1
@@ -275,7 +275,7 @@ async fn run_sync_cycle(
275 275
276 276 // Cleanup old entries + enforce retention cap
277 277 let p = db_path.to_path_buf();
278 - tokio::task::spawn_blocking(move || {
278 + let retention = tokio::task::spawn_blocking(move || {
279 279 let conn = rusqlite::Connection::open(&p)?;
280 280 service::cleanup_changelog(&conn)?;
281 281 service::enforce_changelog_retention(&conn)
@@ -283,6 +283,18 @@ async fn run_sync_cycle(
283 283 .await
284 284 .map_err(|e| SyncError::Other(e.to_string()))??;
285 285
286 + // Dropping unpushed entries is silent data loss from the user's other
287 + // devices — surface it as a hard error, not just a log line. `push_changes`
288 + // now fully drains the backlog each cycle, so this only fires when sync has
289 + // genuinely been unable to reach the server for a long time.
290 + if retention.unpushed_dropped > 0 {
291 + status.lock().last_error = Some(format!(
292 + "Sync could not keep up: {} local change(s) were dropped and will not \
293 + reach your other devices. Check your connection and re-sync.",
294 + retention.unpushed_dropped
295 + ));
296 + }
297 +
286 298 // Update status
287 299 {
288 300 let p = db_path.to_path_buf();
M crates/audiofiles-sync/src/service/mod.rs +1 -1
@@ -195,5 +195,5 @@ pub fn count_pending_changes(conn: &Connection) -> Result<i64> {
195 195
196 196 // Re-exports from submodules
197 197 pub use download::{download_missing_blobs, download_one_blob};
198 - pub use state::{cleanup_changelog, create_initial_snapshot, enforce_changelog_retention, mark_cloud_only_samples};
198 + pub use state::{cleanup_changelog, create_initial_snapshot, enforce_changelog_retention, mark_cloud_only_samples, RetentionOutcome};
199 199 pub use upload::{perform_sync, upload_pending_blobs};
M crates/audiofiles-sync/src/service/state.rs +81 -13
@@ -71,15 +71,33 @@ pub fn cleanup_changelog(conn: &Connection) -> Result<i64> {
71 71 Ok(deleted as i64)
72 72 }
73 73
74 + /// Result of a changelog-retention sweep.
75 + #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
76 + pub struct RetentionOutcome {
77 + /// Total entries deleted (pushed + unpushed).
78 + pub deleted: i64,
79 + /// Of those, how many were **unpushed** — changes that had not yet reached
80 + /// the server. A non-zero value means local edits were permanently dropped
81 + /// from propagation; the scheduler surfaces this as a hard `last_error`.
82 + pub unpushed_dropped: i64,
83 + }
84 +
74 85 /// Enforce a hard cap on total changelog entries to prevent unbounded growth
75 86 /// when sync is disconnected or failing. Deletes the oldest entries (by rowid)
76 87 /// to bring the count back under `MAX_CHANGELOG_ENTRIES`.
88 + ///
89 + /// Pushed entries are culled first; unpushed entries are dropped only if the cap
90 + /// still isn't met (a genuinely-offline-too-long situation, since `push_changes`
91 + /// now fully drains the backlog every cycle before retention runs). Dropping an
92 + /// unpushed entry is silent data loss from the user's other devices, so the
93 + /// count is reported back and the scheduler raises a visible error rather than
94 + /// only logging.
77 95 #[instrument(skip_all)]
78 - pub fn enforce_changelog_retention(conn: &Connection) -> Result<i64> {
96 + pub fn enforce_changelog_retention(conn: &Connection) -> Result<RetentionOutcome> {
79 97 let count: i64 = conn.query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))?;
80 98
81 99 if count <= MAX_CHANGELOG_ENTRIES {
82 - return Ok(0);
100 + return Ok(RetentionOutcome::default());
83 101 }
84 102
85 103 let excess = count - MAX_CHANGELOG_ENTRIES;
@@ -92,6 +110,7 @@ pub fn enforce_changelog_retention(conn: &Connection) -> Result<i64> {
92 110 )?;
93 111
94 112 let mut total_deleted = deleted_pushed as i64;
113 + let mut unpushed_dropped = 0i64;
95 114
96 115 // If still over cap, reluctantly delete unpushed entries (oldest first).
97 116 if total_deleted < excess {
@@ -102,12 +121,14 @@ pub fn enforce_changelog_retention(conn: &Connection) -> Result<i64> {
102 121 [remaining],
103 122 )?;
104 123 if deleted_unpushed > 0 {
105 - tracing::warn!(
124 + tracing::error!(
106 125 deleted_unpushed,
107 - "Changelog retention: dropped unpushed entries — sync was offline too long"
126 + "Changelog retention: dropped unpushed entries — these local changes will \
127 + not reach your other devices. Sync has been unable to push for too long."
108 128 );
109 129 }
110 - total_deleted += deleted_unpushed as i64;
130 + unpushed_dropped = deleted_unpushed as i64;
131 + total_deleted += unpushed_dropped;
111 132 }
112 133
113 134 tracing::warn!(
@@ -116,7 +137,10 @@ pub fn enforce_changelog_retention(conn: &Connection) -> Result<i64> {
116 137 limit = MAX_CHANGELOG_ENTRIES,
117 138 "Changelog retention cap enforced"
118 139 );
119 - Ok(total_deleted)
140 + Ok(RetentionOutcome {
141 + deleted: total_deleted,
142 + unpushed_dropped,
143 + })
120 144 }
121 145
122 146 /// Mark samples as cloud_only when their blob doesn't exist on disk.
@@ -790,8 +814,11 @@ mod tests {
790 814 conn.query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0)).unwrap();
791 815 assert_eq!(before, total);
792 816
793 - let deleted = enforce_changelog_retention(conn).unwrap();
794 - assert_eq!(deleted, 500);
817 + let outcome = enforce_changelog_retention(conn).unwrap();
818 + assert_eq!(outcome.deleted, 500);
819 + // Every seeded entry was unpushed, so retention had to drop unpushed rows —
820 + // the scheduler surfaces this as a hard error.
821 + assert_eq!(outcome.unpushed_dropped, 500);
795 822
796 823 let after: i64 =
797 824 conn.query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0)).unwrap();
@@ -823,8 +850,49 @@ mod tests {
823 850 .unwrap();
824 851 }
825 852
826 - let deleted = enforce_changelog_retention(conn).unwrap();
827 - assert_eq!(deleted, 0);
853 + let outcome = enforce_changelog_retention(conn).unwrap();
854 + assert_eq!(outcome.deleted, 0);
855 + assert_eq!(outcome.unpushed_dropped, 0);
856 + }
857 +
858 + #[test]
859 + fn enforce_retention_prefers_pushed_over_unpushed() {
860 + let db = setup_test_db();
861 + let conn = db.conn();
862 + clear_changelog(conn);
863 +
864 + // Seed exactly the cap in already-pushed rows, then 300 unpushed on top.
865 + // Retention must remove the 300 excess entirely from the pushed pool and
866 + // leave every unpushed row intact — no silent divergence.
867 + for i in 0..MAX_CHANGELOG_ENTRIES {
868 + conn.execute(
869 + "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) \
870 + VALUES ('samples', 'UPDATE', ?1, '{}', 1)",
871 + [format!("pushed-{}", i)],
872 + )
873 + .unwrap();
874 + }
875 + for i in 0..300 {
876 + conn.execute(
877 + "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) \
878 + VALUES ('samples', 'INSERT', ?1, '{}', 0)",
879 + [format!("unpushed-{}", i)],
880 + )
881 + .unwrap();
882 + }
883 +
884 + let outcome = enforce_changelog_retention(conn).unwrap();
885 + assert_eq!(outcome.deleted, 300);
886 + assert_eq!(outcome.unpushed_dropped, 0);
887 +
888 + let unpushed_left: i64 = conn
889 + .query_row(
890 + "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
891 + [],
892 + |r| r.get(0),
893 + )
894 + .unwrap();
895 + assert_eq!(unpushed_left, 300);
828 896 }
829 897
830 898 #[test]
@@ -1202,8 +1270,8 @@ mod tests {
1202 1270 ).unwrap();
1203 1271 }
1204 1272
1205 - let deleted = enforce_changelog_retention(conn).unwrap();
1206 - assert_eq!(deleted, 0);
1273 + let outcome = enforce_changelog_retention(conn).unwrap();
1274 + assert_eq!(outcome.deleted, 0);
1207 1275
1208 1276 let count: i64 = conn.query_row(
1209 1277 "SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0),
@@ -1241,7 +1309,7 @@ mod tests {
1241 1309
1242 1310 // retention should be a noop now (exactly MAX_CHANGELOG_ENTRIES remain)
1243 1311 let retained = enforce_changelog_retention(conn).unwrap();
1244 - assert_eq!(retained, 0);
1312 + assert_eq!(retained.deleted, 0);
1245 1313 }
1246 1314
1247 1315 #[test]
M crates/audiofiles-sync/src/service/upload.rs +35 -2
@@ -160,12 +160,44 @@ async fn ensure_device_registered(
160 160 }
161 161
162 162 /// Push unpushed changelog entries to the server.
163 + ///
164 + /// Drains the full backlog: each cycle repeatedly pushes `PUSH_BATCH_LIMIT`-sized
165 + /// batches until no unpushed entries remain (mirroring `pull_changes`' `has_more`
166 + /// loop). Draining a single batch per cycle used to let changelog retention
167 + /// (`enforce_changelog_retention`, capped at `MAX_CHANGELOG_ENTRIES`) cull
168 + /// still-unpushed rows on a large bulk import before the push side could reach
169 + /// them — silently dropping those changes from every other device.
163 170 #[instrument(skip_all)]
164 171 async fn push_changes(
165 172 db_path: &std::path::Path,
166 173 client: &SyncKitClient,
167 174 device_id: DeviceId,
168 175 ) -> Result<i64> {
176 + let mut total_pushed = 0i64;
177 +
178 + loop {
179 + let (pushed, rows_read) = push_one_batch(db_path, client, device_id).await?;
180 + total_pushed += pushed;
181 + // An empty batch means the backlog is drained. A short (< limit) batch
182 + // also means drained, but reading one more empty batch is cheap and keeps
183 + // the termination condition trivially correct.
184 + if rows_read == 0 {
185 + break;
186 + }
187 + }
188 +
189 + Ok(total_pushed)
190 + }
191 +
192 + /// Push a single `PUSH_BATCH_LIMIT`-sized batch of unpushed changelog entries.
193 + ///
194 + /// Returns `(pushed_count, rows_read)`. `rows_read == 0` signals the backlog is
195 + /// drained; the caller loops until then.
196 + async fn push_one_batch(
197 + db_path: &std::path::Path,
198 + client: &SyncKitClient,
199 + device_id: DeviceId,
200 + ) -> Result<(i64, usize)> {
169 201 let node = device_id.as_uuid();
170 202 let p = db_path.to_path_buf();
171 203 let rows = tokio::task::spawn_blocking(move || -> Result<_> {
@@ -203,8 +235,9 @@ async fn push_changes(
203 235 .await
204 236 .map_err(|e| SyncError::Other(e.to_string()))??;
205 237
238 + let rows_read = rows.len();
206 239 if rows.is_empty() {
207 - return Ok(0);
240 + return Ok((0, 0));
208 241 }
209 242
210 243 let mut pushed_ids: Vec<i64> = Vec::with_capacity(rows.len());
@@ -303,5 +336,5 @@ async fn push_changes(
303 336 .map_err(|e| SyncError::Other(e.to_string()))??;
304 337 }
305 338
306 - Ok(pushed_count)
339 + Ok((pushed_count, rows_read))
307 340 }