Skip to main content

max / audiofiles

Perf: require a transaction token for bulk row writes Chronic finding (ultra-fuzz runs 1-4): per-row autocommits in bulk loops kept recurring as new writers were added (analysis save, harvest, rules reconcile, cluster/exemplar/ML tagging) — batching was enforced per call site and drifted. Structural fix: a zero-sized Tx proof token, constructible only by Database::transaction / transaction_core, is now required by the row-write primitives (analysis::save_analysis, rules::apply_tag_sourced, rules::reconcile_sample). They cannot be called outside a transaction, so a bare per-row autocommit loop no longer compiles. Add batched entry points (save_analysis_batch, apply_rules_to_samples) and wrap every bulk caller in one transaction. The import flow now buffers analysis results and flushes them in batches of 64 (and on batch-complete), applying rules per flushed batch so they still see the persisted analysis — replacing ~100k single-row fsyncs on a full library import. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-22 00:22 UTC
Commit: b3f7a24e38c1986422844718bfd5c1944c782984
Parent: f9817b7
16 files changed, +228 insertions, -70 deletions
M crates/audiofiles-browser/src/backend/direct.rs +16 -6
@@ -628,12 +628,15 @@ impl Backend for DirectBackend {
628 628 Ok(audiofiles_core::analysis::load_analysis(&db, hash))
629 629 }
630 630
631 - fn save_analysis(&self, result: &AnalysisResult) -> BackendResult<()> {
631 + fn save_analysis_batch(&self, results: &[AnalysisResult]) -> BackendResult<()> {
632 + if results.is_empty() {
633 + return Ok(());
634 + }
632 635 let db = self.db.lock();
633 - audiofiles_core::analysis::save_analysis(&db, result)?;
634 - // Invalidate the search worker's cached VP-trees — new analysis data
635 - // changes normalization ranges and may add a new fingerprint, so both
636 - // indexes must rebuild on the next query.
636 + audiofiles_core::analysis::save_analysis_batch(&db, results)?;
637 + // Invalidate the search worker's cached VP-trees once for the batch — new
638 + // analysis data changes normalization ranges and may add fingerprints, so
639 + // both indexes must rebuild on the next query.
637 640 if let Some(w) = self.search_worker.lock().as_ref() {
638 641 w.send(SearchCommand::Invalidate);
639 642 }
@@ -645,6 +648,11 @@ impl Backend for DirectBackend {
645 648 Ok(audiofiles_core::rules::apply_rules_to_sample(&db, hash)?)
646 649 }
647 650
651 + fn apply_rules_to_samples(&self, hashes: &[String]) -> BackendResult<usize> {
652 + let db = self.db.lock();
653 + Ok(audiofiles_core::rules::apply_rules_to_samples(&db, hashes)?)
654 + }
655 +
648 656 fn samples_needing_features(&self) -> BackendResult<Vec<(String, String)>> {
649 657 let db = self.db.lock();
650 658 Ok(audiofiles_core::analysis::samples_needing_features(&db)?)
@@ -801,7 +809,9 @@ impl Backend for DirectBackend {
801 809
802 810 fn accept_ml_tag(&self, hash: &str, tag: &str) -> BackendResult<()> {
803 811 let db = self.db.lock();
804 - audiofiles_core::rules::apply_tag_sourced(&db, hash, tag, "ml")?;
812 + db.transaction_core(|tx| {
813 + audiofiles_core::rules::apply_tag_sourced(&db, tx, hash, tag, "ml").map(|_| ())
814 + })?;
805 815 Ok(())
806 816 }
807 817
M crates/audiofiles-browser/src/backend/mod.rs +9 -2
@@ -420,13 +420,20 @@ pub trait Backend: Send + Sync {
420 420 /// Get the full analysis result for a sample, if it exists.
421 421 fn get_analysis(&self, hash: &str) -> BackendResult<Option<AnalysisResult>>;
422 422
423 - /// Save an analysis result to the database.
424 - fn save_analysis(&self, result: &AnalysisResult) -> BackendResult<()>;
423 + /// Save a batch of analysis results in one transaction. The analysis worker
424 + /// emits results in bulk, so callers buffer and flush rather than saving one
425 + /// result per call (each save is several fsynced writes).
426 + fn save_analysis_batch(&self, results: &[AnalysisResult]) -> BackendResult<()>;
425 427
426 428 /// Apply the deterministic tag rules to a sample, reconciling rule-sourced tags.
427 429 /// Returns whether any tag changed. No-op while the rule set is empty.
428 430 fn apply_rules_to_sample(&self, hash: &str) -> BackendResult<bool>;
429 431
432 + /// Batched [`apply_rules_to_sample`](Self::apply_rules_to_sample): reconcile a
433 + /// set of freshly analyzed samples in one transaction. Returns the number
434 + /// changed.
435 + fn apply_rules_to_samples(&self, hashes: &[String]) -> BackendResult<usize>;
436 +
430 437 /// Hashes (+ extension) of samples lacking a current-version feature vector
431 438 /// (the feature-backfill work-list).
432 439 fn samples_needing_features(&self) -> BackendResult<Vec<(String, String)>>;
M crates/audiofiles-browser/src/state/import_workflow.rs +28 -4
@@ -4,6 +4,10 @@ use tracing::{error, warn};
4 4
5 5 use super::*;
6 6
7 + /// How many analysis results to accumulate before flushing them to the DB in one
8 + /// transaction. Bounds buffered memory while keeping per-result fsyncs rare.
9 + const ANALYSIS_SAVE_BATCH: usize = 64;
10 +
7 11 /// Count audio files in a directory (recursive). Used for the import dry-run preview.
8 12 fn count_audio_files(dir: &Path) -> usize {
9 13 walk_folder_stats(dir).0
@@ -53,6 +57,19 @@ pub struct ImportPreflight {
53 57 }
54 58
55 59 impl BrowserState {
60 + /// Persist buffered analysis results in one transaction, then apply tag rules
61 + /// to the flushed hashes (also one transaction). Rules run after the save so
62 + /// they evaluate against the persisted analysis. No-op when the buffer empty.
63 + fn flush_analysis_saves(&mut self) {
64 + if self.analysis_save_buffer.is_empty() {
65 + return;
66 + }
67 + let results = std::mem::take(&mut self.analysis_save_buffer);
68 + let hashes: Vec<String> = results.iter().map(|r| r.hash.clone()).collect();
69 + let _ = self.backend.save_analysis_batch(&results);
70 + let _ = self.backend.apply_rules_to_samples(&hashes);
71 + }
72 +
56 73 /// Quick-import a single file or directory via drag-and-drop into the current folder.
57 74 /// After importing, starts the analysis flow so columns (BPM, Key, etc.) get populated.
58 75 pub fn import_path(&mut self, path: &Path) {
@@ -566,10 +583,14 @@ impl BrowserState {
566 583 suggestions,
567 584 } => {
568 585 let result = *result;
569 - let _ = self.backend.save_analysis(&result);
570 - // Pipeline wiring: apply deterministic tag rules to the freshly
571 - // analyzed sample (no-op while the rule set is empty).
572 - let _ = self.backend.apply_rules_to_sample(&result.hash);
586 + // Buffer the DB write; flushed in one transaction every
587 + // ANALYSIS_SAVE_BATCH results and on AnalysisBatchComplete.
588 + // (Rules are applied per flushed batch, after the save, so they
589 + // still see the persisted analysis.)
590 + self.analysis_save_buffer.push(result.clone());
591 + if self.analysis_save_buffer.len() >= ANALYSIS_SAVE_BATCH {
592 + self.flush_analysis_saves();
593 + }
573 594
574 595 // Backfill: persist vectors + apply rules only — no review queue.
575 596 if !self.backfill_in_progress {
@@ -597,6 +618,9 @@ impl BrowserState {
597 618 self.analysis_errors.push(super::AnalysisFileError { hash, name, error });
598 619 }
599 620 BackendEvent::AnalysisBatchComplete => {
621 + // Persist any analysis results still buffered, then apply rules,
622 + // before the review/backfill bookkeeping below.
623 + self.flush_analysis_saves();
600 624 if self.backfill_in_progress {
601 625 self.backfill_in_progress = false;
602 626 let errors = self.analysis_errors.len();
M crates/audiofiles-browser/src/state/mod.rs +6
@@ -225,6 +225,11 @@ pub struct BrowserState {
225 225 pub bulk_move_filter: String,
226 226 pub pending_review_items: Vec<ReviewItem>,
227 227
228 + /// Analysis results awaiting a batched DB write. The analysis worker emits
229 + /// results far faster than per-row autocommits can fsync, so they accumulate
230 + /// here and flush in one transaction (see `flush_analysis_saves`).
231 + pub analysis_save_buffer: Vec<audiofiles_core::analysis::AnalysisResult>,
232 +
228 233 /// True while a silent feature-backfill batch is running (reuses the analysis
229 234 /// worker but suppresses the review/progress screens). See `start_backfill`.
230 235 pub backfill_in_progress: bool,
@@ -539,6 +544,7 @@ impl BrowserState {
539 544 // M-6.
540 545 bulk_move_filter: String::new(),
541 546 pending_review_items: Vec::new(),
547 + analysis_save_buffer: Vec::new(),
542 548 backfill_in_progress: false,
543 549 import_file_errors: Vec::new(),
544 550 analysis_errors: Vec::new(),
M crates/audiofiles-core/src/analysis/cluster.rs +9 -6
@@ -212,13 +212,16 @@ pub fn cluster_library(db: &Database, k: usize, max_iters: usize) -> Result<Clus
212 212 /// Returns the number of samples that gained the tag.
213 213 #[instrument(skip_all)]
214 214 pub fn apply_cluster_tag(db: &Database, member_hashes: &[String], tag: &str) -> Result<usize> {
215 - let mut added = 0;
216 - for hash in member_hashes {
217 - if crate::rules::apply_tag_sourced(db, hash, tag, "cluster")? {
218 - added += 1;
215 + // One transaction for the whole cluster, not an autocommit per member.
216 + db.transaction_core(|tx| {
217 + let mut added = 0;
218 + for hash in member_hashes {
219 + if crate::rules::apply_tag_sourced(db, tx, hash, tag, "cluster")? {
220 + added += 1;
221 + }
219 222 }
220 - }
221 - Ok(added)
223 + Ok(added)
224 + })
222 225 }
223 226
224 227 #[cfg(test)]
M crates/audiofiles-core/src/analysis/exemplar.rs +8 -4
@@ -409,9 +409,12 @@ pub fn apply_ml_suggestions(
409 409 k: usize,
410 410 ) -> Result<MlOutcome> {
411 411 let outcome = preview_sample(db, hash, index, k)?;
412 - for s in &outcome.auto_applied {
413 - crate::rules::apply_tag_sourced(db, hash, &s.tag, "ml")?;
414 - }
412 + db.transaction_core(|tx| {
413 + for s in &outcome.auto_applied {
414 + crate::rules::apply_tag_sourced(db, tx, hash, &s.tag, "ml")?;
415 + }
416 + Ok(())
417 + })?;
415 418 Ok(outcome)
416 419 }
417 420
@@ -517,7 +520,8 @@ mod tests {
517 520 // One real exemplar + one whose only tag is ML-sourced (must not count as label).
518 521 insert(&db, "k1", vec_at(0.0), &["instrument.drum.kick"]);
519 522 insert(&db, "m1", vec_at(0.0), &[]);
520 - crate::rules::apply_tag_sourced(&db, "m1", "bogus.ml", "ml").unwrap();
523 + db.transaction_core(|tx| crate::rules::apply_tag_sourced(&db, tx, "m1", "bogus.ml", "ml").map(|_| ()))
524 + .unwrap();
521 525
522 526 let index = build_index(&db).unwrap();
523 527 // Only k1 is a labeled exemplar (m1's lone tag is ML).
M crates/audiofiles-core/src/analysis/mod.rs +27 -4
@@ -27,7 +27,7 @@ use std::path::Path;
27 27 use classify::SampleClass;
28 28 use config::AnalysisConfig;
29 29
30 - use crate::db::Database;
30 + use crate::db::{Database, Tx};
31 31 use crate::error::{unix_now, CoreError};
32 32 use crate::fingerprint;
33 33 use tracing::instrument;
@@ -286,9 +286,32 @@ pub fn analyze_sample(
286 286 Ok(result)
287 287 }
288 288
289 + /// Save many analysis results in a single transaction.
290 + ///
291 + /// This is the batched entry point for high-volume callers (the analysis worker
292 + /// produces results far faster than per-row autocommits can be fsynced). Use it
293 + /// instead of calling [`save_analysis`] in a loop.
294 + #[instrument(skip_all)]
295 + pub fn save_analysis_batch(db: &Database, results: &[AnalysisResult]) -> Result<(), CoreError> {
296 + if results.is_empty() {
297 + return Ok(());
298 + }
299 + db.transaction_core(|tx| {
300 + for result in results {
301 + save_analysis(db, tx, result)?;
302 + }
303 + Ok(())
304 + })
305 + }
306 +
289 307 /// Save analysis results to the database, overwriting any previous results for this hash.
308 + ///
309 + /// Requires a [`Tx`]: each call issues several writes (analysis row, feature
310 + /// vector, fingerprint), and the analysis worker emits results in bulk, so these
311 + /// must be batched in one transaction rather than autocommitting per result. For
312 + /// a single sample use [`save_analysis_batch`] with a one-element slice.
290 313 #[instrument(skip_all)]
291 - pub fn save_analysis(db: &Database, result: &AnalysisResult) -> Result<(), CoreError> {
314 + pub fn save_analysis(db: &Database, _tx: &Tx, result: &AnalysisResult) -> Result<(), CoreError> {
292 315 let now = unix_now();
293 316 db.conn().execute(
294 317 "INSERT OR REPLACE INTO audio_analysis (
@@ -550,7 +573,7 @@ mod tests {
550 573 let mut result = minimal_result("feat1");
551 574 result.feature_vector = Some((0..35).map(|i| i as f64).collect());
552 575 result.feature_version = Some(classify::FEATURE_VERSION);
553 - save_analysis(&db, &result).unwrap();
576 + save_analysis_batch(&db, std::slice::from_ref(&result)).unwrap();
554 577
555 578 let (ver, vector_json): (u32, String) = db
556 579 .conn()
@@ -577,7 +600,7 @@ mod tests {
577 600 )
578 601 .unwrap();
579 602 // No feature_vector set — sample_features must stay empty.
580 - save_analysis(&db, &minimal_result("feat2")).unwrap();
603 + save_analysis_batch(&db, std::slice::from_ref(&minimal_result("feat2"))).unwrap();
581 604 let count: i64 = db
582 605 .conn()
583 606 .query_row("SELECT COUNT(*) FROM sample_features WHERE hash = 'feat2'", [], |r| r.get(0))
M crates/audiofiles-core/src/analysis/trained_head.rs +6 -3
@@ -275,9 +275,12 @@ pub fn preview_sample(db: &Database, hash: &str, head: &TrainedHead) -> Result<M
275 275 #[instrument(skip_all)]
276 276 pub fn apply_ml_suggestions(db: &Database, hash: &str, head: &TrainedHead) -> Result<MlOutcome> {
277 277 let outcome = preview_sample(db, hash, head)?;
278 - for s in &outcome.auto_applied {
279 - crate::rules::apply_tag_sourced(db, hash, &s.tag, "ml")?;
280 - }
278 + db.transaction_core(|tx| {
279 + for s in &outcome.auto_applied {
280 + crate::rules::apply_tag_sourced(db, tx, hash, &s.tag, "ml")?;
281 + }
282 + Ok(())
283 + })?;
281 284 Ok(outcome)
282 285 }
283 286
M crates/audiofiles-core/src/db.rs +49 -7
@@ -1505,6 +1505,20 @@ fn register_hash_row_id(conn: &Connection) -> Result<(), DbError> {
1505 1505 Ok(())
1506 1506 }
1507 1507
1508 + /// Compile-time proof that a write transaction is open on the connection.
1509 + ///
1510 + /// Constructed only by [`Database::transaction`], and required by the row-write
1511 + /// functions that must run inside a batched transaction rather than as
1512 + /// standalone autocommits in a loop (e.g. [`crate::analysis::save_analysis`],
1513 + /// [`crate::rules::apply_tag_sourced`]). Holding a `&Tx` is the only way to call
1514 + /// those functions, so "bulk loop of per-row autocommits" — the chronic
1515 + /// per-row-commit pattern — does not compile: there is no `Tx` to pass except
1516 + /// inside a `transaction` closure that already batches the whole loop.
1517 + ///
1518 + /// The token is a zero-sized marker; the SQL still runs on `db.conn()`, which
1519 + /// participates in the ambient `BEGIN IMMEDIATE` opened by `transaction`.
1520 + pub struct Tx(());
1521 +
1508 1522 impl Database {
1509 1523 /// Open (or create) the database at the given path and run migrations.
1510 1524 #[instrument(skip_all)]
@@ -1669,16 +1683,44 @@ impl Database {
1669 1683 /// Run a closure inside a SQLite transaction.
1670 1684 ///
1671 1685 /// Uses `BEGIN IMMEDIATE` to acquire a write lock upfront, preventing
1672 - /// deadlocks when the closure issues writes. The closure receives no
1673 - /// arguments — it accesses the same `Database` through the shared
1674 - /// `Mutex<Database>`, which is safe because the caller already holds the lock.
1686 + /// deadlocks when the closure issues writes. The closure receives a [`Tx`]
1687 + /// token proving a transaction is open; pass it to row-write functions that
1688 + /// require batching (the token cannot be constructed any other way, so those
1689 + /// functions cannot be called in an un-batched per-row loop). The closure
1690 + /// accesses the same `Database` through the shared `Mutex<Database>`, which is
1691 + /// safe because the caller already holds the lock.
1675 1692 #[instrument(skip_all)]
1676 1693 pub fn transaction<T, F>(&self, f: F) -> Result<T, DbError>
1677 1694 where
1678 - F: FnOnce() -> Result<T, DbError>,
1695 + F: FnOnce(&Tx) -> Result<T, DbError>,
1696 + {
1697 + self.conn.execute_batch("BEGIN IMMEDIATE")?;
1698 + match f(&Tx(())) {
1699 + Ok(val) => {
1700 + self.conn.execute_batch("COMMIT")?;
1701 + Ok(val)
1702 + }
1703 + Err(e) => {
1704 + if let Err(rb_err) = self.conn.execute_batch("ROLLBACK") {
1705 + tracing::warn!("ROLLBACK failed after transaction error: {rb_err}");
1706 + }
1707 + Err(e)
1708 + }
1709 + }
1710 + }
1711 +
1712 + /// [`transaction`](Self::transaction) for closures that produce a
1713 + /// [`CoreError`](crate::error::CoreError) — i.e. the row-write batchers in
1714 + /// `analysis`, `rules`, and `harvest`, which call functions returning
1715 + /// `CoreError`. Same `BEGIN IMMEDIATE` / commit / rollback semantics and the
1716 + /// same [`Tx`] proof token; only the closure's error type differs.
1717 + #[instrument(skip_all)]
1718 + pub fn transaction_core<T, F>(&self, f: F) -> Result<T, crate::error::CoreError>
1719 + where
1720 + F: FnOnce(&Tx) -> Result<T, crate::error::CoreError>,
1679 1721 {
1680 1722 self.conn.execute_batch("BEGIN IMMEDIATE")?;
1681 - match f() {
1723 + match f(&Tx(())) {
1682 1724 Ok(val) => {
1683 1725 self.conn.execute_batch("COMMIT")?;
1684 1726 Ok(val)
@@ -2182,7 +2224,7 @@ mod tests {
2182 2224 #[test]
2183 2225 fn transaction_commits_on_success() {
2184 2226 let db = Database::open_in_memory().unwrap();
2185 - db.transaction(|| {
2227 + db.transaction(|_tx| {
2186 2228 db.conn().execute(
2187 2229 "INSERT INTO user_config (key, value) VALUES ('test_key', 'test_value')",
2188 2230 [],
@@ -2205,7 +2247,7 @@ mod tests {
2205 2247 #[test]
2206 2248 fn transaction_rolls_back_on_error() {
2207 2249 let db = Database::open_in_memory().unwrap();
2208 - let result: Result<(), DbError> = db.transaction(|| {
2250 + let result: Result<(), DbError> = db.transaction(|_tx| {
2209 2251 db.conn().execute(
2210 2252 "INSERT INTO user_config (key, value) VALUES ('rollback_key', 'val')",
2211 2253 [],
M crates/audiofiles-core/src/harvest.rs +9 -6
@@ -91,13 +91,16 @@ pub fn harvest_folder_labels(db: &Database) -> Result<Vec<FolderLabel>> {
91 91 /// Returns the number of samples that gained the tag.
92 92 #[instrument(skip_all)]
93 93 pub fn apply_harvested_tag(db: &Database, hashes: &[String], tag: &str) -> Result<usize> {
94 - let mut added = 0;
95 - for hash in hashes {
96 - if crate::rules::apply_tag_sourced(db, hash, tag, "harvest")? {
97 - added += 1;
94 + // One transaction for the whole harvest label, not an autocommit per sample.
95 + db.transaction_core(|tx| {
96 + let mut added = 0;
97 + for hash in hashes {
98 + if crate::rules::apply_tag_sourced(db, tx, hash, tag, "harvest")? {
99 + added += 1;
100 + }
98 101 }
99 - }
100 - Ok(added)
102 + Ok(added)
103 + })
101 104 }
102 105
103 106 #[cfg(test)]
M crates/audiofiles-core/src/rules.rs +48 -15
@@ -15,7 +15,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
15 15 use serde::{Deserialize, Serialize};
16 16 use tracing::instrument;
17 17
18 - use crate::db::Database;
18 + use crate::db::{Database, Tx};
19 19 use crate::error::{CoreError, Result};
20 20 use crate::tags::validate_tag;
21 21
@@ -586,11 +586,18 @@ pub fn set_rule_enabled(db: &Database, id: &str, enabled: bool) -> Result<()> {
586 586 #[instrument(skip_all)]
587 587 pub fn delete_rule(db: &Database, id: &str) -> Result<()> {
588 588 let affected = samples_tagged_by_rule(db, id)?;
589 - db.conn().execute("DELETE FROM tag_rules WHERE id = ?1", [id])?;
590 - let rules = list_rules(db)?;
591 - for hash in affected {
592 - reconcile_sample(db, &hash, &rules)?;
593 - }
589 + // One transaction for the rule delete + reconciling every affected sample,
590 + // instead of an autocommit per sample. The rule must be deleted *before*
591 + // re-listing, so reconcile evaluates against the rule set without it and
592 + // removes its now-orphaned tags.
593 + db.transaction_core(|tx| {
594 + db.conn().execute("DELETE FROM tag_rules WHERE id = ?1", [id])?;
595 + let rules = list_rules(db)?;
596 + for hash in &affected {
597 + reconcile_sample(db, tx, hash, &rules)?;
598 + }
599 + Ok(())
600 + })?;
594 601 Ok(())
595 602 }
596 603
@@ -608,7 +615,7 @@ fn samples_tagged_by_rule(db: &Database, id: &str) -> Result<Vec<String>> {
608 615 ///
609 616 /// Only rule-sourced tags are added/removed; manual tags (no provenance row) are
610 617 /// never touched. Returns `true` if any tag changed.
611 - fn reconcile_sample(db: &Database, hash: &str, rules: &[Rule]) -> Result<bool> {
618 + fn reconcile_sample(db: &Database, _tx: &Tx, hash: &str, rules: &[Rule]) -> Result<bool> {
612 619 let Some(ctx) = load_context(db, hash)? else {
613 620 return Ok(false);
614 621 };
@@ -681,7 +688,28 @@ fn reconcile_sample(db: &Database, hash: &str, rules: &[Rule]) -> Result<bool> {
681 688 #[instrument(skip_all)]
682 689 pub fn apply_rules_to_sample(db: &Database, hash: &str) -> Result<bool> {
683 690 let rules = list_rules(db)?;
684 - reconcile_sample(db, hash, &rules)
691 + db.transaction_core(|tx| reconcile_sample(db, tx, hash, &rules))
692 + }
693 +
694 + /// Apply all enabled rules to a set of samples in a single transaction. The
695 + /// batched form of [`apply_rules_to_sample`]: callers processing many freshly
696 + /// analyzed samples reconcile them in one commit instead of one per sample.
697 + /// Returns the number of samples whose tags changed.
698 + #[instrument(skip_all)]
699 + pub fn apply_rules_to_samples(db: &Database, hashes: &[String]) -> Result<usize> {
700 + if hashes.is_empty() {
701 + return Ok(0);
702 + }
703 + let rules = list_rules(db)?;
704 + db.transaction_core(|tx| {
705 + let mut changed = 0;
706 + for hash in hashes {
707 + if reconcile_sample(db, tx, hash, &rules)? {
708 + changed += 1;
709 + }
710 + }
711 + Ok(changed)
712 + })
685 713 }
686 714
687 715 /// Re-apply all rules across the whole library. Returns the number of samples changed.
@@ -695,13 +723,15 @@ pub fn apply_all_rules(db: &Database) -> Result<usize> {
695 723 stmt.query_map([], |row| row.get(0))?
696 724 .collect::<std::result::Result<Vec<_>, _>>()?
697 725 };
698 - let mut changed = 0;
699 - for hash in hashes {
700 - if reconcile_sample(db, &hash, &rules)? {
701 - changed += 1;
726 + db.transaction_core(|tx| {
727 + let mut changed = 0;
728 + for hash in hashes {
729 + if reconcile_sample(db, tx, &hash, &rules)? {
730 + changed += 1;
731 + }
702 732 }
703 - }
704 - Ok(changed)
733 + Ok(changed)
734 + })
705 735 }
706 736
707 737 /// Dry-run: how many (non-deleted) samples a rule's conditions would match.
@@ -731,8 +761,11 @@ pub fn preview_rule_matches(db: &Database, rule: &Rule) -> Result<usize> {
731 761 /// are regenerable). Existing provenance is preserved (`INSERT OR IGNORE`), so this
732 762 /// won't downgrade a tag that is already rule-sourced or manual. Returns `true` if the
733 763 /// tag was newly added to the sample.
764 + /// Requires a [`Tx`]: machine-source tagging is always a bulk operation (a whole
765 + /// cluster / harvest / ML pass), so callers must batch the loop in one
766 + /// transaction rather than autocommitting per sample.
734 767 #[instrument(skip_all)]
735 - pub fn apply_tag_sourced(db: &Database, hash: &str, tag: &str, source: &str) -> Result<bool> {
768 + pub fn apply_tag_sourced(db: &Database, _tx: &Tx, hash: &str, tag: &str, source: &str) -> Result<bool> {
736 769 validate_tag(tag)?;
737 770 let conn = db.conn();
738 771 let added = conn.execute(
M crates/audiofiles-core/src/similarity.rs +1 -1
@@ -512,7 +512,7 @@ mod tests {
512 512 feature_vector: None,
513 513 feature_version: None,
514 514 };
515 - analysis::save_analysis(db, &result).unwrap();
515 + analysis::save_analysis_batch(db, std::slice::from_ref(&result)).unwrap();
516 516 }
517 517
518 518 #[test]
M crates/audiofiles-core/src/store.rs +3 -3
@@ -329,7 +329,7 @@ impl SampleStore {
329 329 // re-writes the blob instead. A failed unlink only leaves an orphan
330 330 // blob (disk waste, reclaimed by a later re-import or verify sweep) and
331 331 // must not roll back the committed row delete, so its error is ignored.
332 - let deleted = db.transaction(|| {
332 + let deleted = db.transaction(|_tx| {
333 333 db.conn().execute(
334 334 "UPDATE sync_state SET value = '1' WHERE key = 'applying_remote'",
335 335 [],
@@ -787,7 +787,7 @@ pub fn relocate_missing_loose_files(
787 787 // 4. Atomic update of all relocated source_paths.
788 788 let relocated = relocated_pairs.len();
789 789 if relocated > 0 {
790 - db.transaction(|| {
790 + db.transaction(|_tx| {
791 791 for (hash, new_path) in &relocated_pairs {
792 792 db.conn().execute(
793 793 "UPDATE samples SET source_path = ?1 WHERE hash = ?2",
@@ -822,7 +822,7 @@ pub fn purge_missing_loose_files(db: &Database) -> Result<usize> {
822 822 let purged = to_purge.len();
823 823
824 824 if purged > 0 {
825 - db.transaction(|| {
825 + db.transaction(|_tx| {
826 826 for hash in &to_purge {
827 827 db.conn()
828 828 .execute("DELETE FROM samples WHERE hash = ?1", [hash])?;
M crates/audiofiles-core/src/tags.rs +4 -4
@@ -100,7 +100,7 @@ pub fn list_all_tags(db: &Database) -> Result<Vec<String>> {
100 100 #[instrument(skip_all)]
101 101 pub fn bulk_add_tag(db: &Database, hashes: &[&str], tag: &str) -> Result<usize> {
102 102 validate_tag(tag)?;
103 - Ok(db.transaction(|| {
103 + Ok(db.transaction(|_tx| {
104 104 let mut count = 0;
105 105 for hash in hashes {
106 106 count += db.conn().execute(
@@ -116,7 +116,7 @@ pub fn bulk_add_tag(db: &Database, hashes: &[&str], tag: &str) -> Result<usize>
116 116 /// Returns count of rows deleted.
117 117 #[instrument(skip_all)]
118 118 pub fn bulk_remove_tag(db: &Database, hashes: &[&str], tag: &str) -> Result<usize> {
119 - Ok(db.transaction(|| {
119 + Ok(db.transaction(|_tx| {
120 120 let mut count = 0;
121 121 for hash in hashes {
122 122 count += db.conn().execute(
@@ -137,7 +137,7 @@ pub fn rename_tag_globally(db: &Database, old_tag: &str, new_tag: &str) -> Resul
137 137 if old_tag == new_tag {
138 138 return Ok(0);
139 139 }
140 - Ok(db.transaction(|| {
140 + Ok(db.transaction(|_tx| {
141 141 // Add new_tag to every sample currently carrying old_tag (ignore if
142 142 // the sample already has new_tag), then drop all rows of old_tag.
143 143 db.conn().execute(
@@ -157,7 +157,7 @@ pub fn rename_tag_globally(db: &Database, old_tag: &str, new_tag: &str) -> Resul
157 157 /// deleted (i.e. the number of samples that had the tag).
158 158 #[instrument(skip_all)]
159 159 pub fn remove_tag_globally(db: &Database, tag: &str) -> Result<usize> {
160 - Ok(db.transaction(|| {
160 + Ok(db.transaction(|_tx| {
161 161 let count = db.conn().execute(
162 162 "DELETE FROM tags WHERE tag = ?1",
163 163 rusqlite::params![tag],
M crates/audiofiles-core/src/test_helpers.rs +1 -1
@@ -59,7 +59,7 @@ pub fn insert_sample_with_analysis(
59 59 feature_vector: None,
60 60 feature_version: None,
61 61 };
62 - crate::analysis::save_analysis(db, &result).unwrap();
62 + crate::analysis::save_analysis_batch(db, std::slice::from_ref(&result)).unwrap();
63 63
64 64 crate::vfs::create_sample_link(db, vfs_id, None, name, hash).unwrap()
65 65 }
M crates/audiofiles-core/tests/e2e_pipeline.rs +4 -4
@@ -160,7 +160,7 @@ fn e2e_import_analyze_search_tag_export() {
160 160 // These are best-effort checks; the exact values depend on the algorithms.
161 161
162 162 // Save analysis to DB.
163 - analysis::save_analysis(&env.db, &result).unwrap();
163 + analysis::save_analysis_batch(&env.db, std::slice::from_ref(&result)).unwrap();
164 164
165 165 // Verify analysis is retrievable.
166 166 let loaded = analysis::load_analysis(&env.db, &hash);
@@ -372,7 +372,7 @@ fn e2e_analysis_roundtrip_verify_values() {
372 372
373 373 let config = AnalysisConfig::default();
374 374 let result = analysis::analyze_sample(&hash, &store_path, &config).unwrap();
375 - analysis::save_analysis(&env.db, &result).unwrap();
375 + analysis::save_analysis_batch(&env.db, std::slice::from_ref(&result)).unwrap();
376 376
377 377 // Load back and verify all fields survive the DB roundtrip.
378 378 let loaded = analysis::load_analysis(&env.db, &hash).unwrap();
@@ -420,11 +420,11 @@ fn e2e_multi_sample_search() {
420 420 let config = AnalysisConfig::default();
421 421 let store_path1 = env.store.sample_path(&hash1, "wav").unwrap();
422 422 let result1 = analysis::analyze_sample(&hash1, &store_path1, &config).unwrap();
423 - analysis::save_analysis(&env.db, &result1).unwrap();
423 + analysis::save_analysis_batch(&env.db, std::slice::from_ref(&result1)).unwrap();
424 424
425 425 let store_path2 = env.store.sample_path(&hash2, "wav").unwrap();
426 426 let result2 = analysis::analyze_sample(&hash2, &store_path2, &config).unwrap();
427 - analysis::save_analysis(&env.db, &result2).unwrap();
427 + analysis::save_analysis_batch(&env.db, std::slice::from_ref(&result2)).unwrap();
428 428
429 429 // Create VFS with both samples.
430 430 let vfs_id = vfs::create_vfs(&env.db, "MultiTest").unwrap();