From d74269c16d81df650aba78b08f2fe0dc2b985cf1 Mon Sep 17 00:00:00 2001 From: Aly Raffauf Date: Wed, 5 Aug 2026 17:46:16 -0400 Subject: [PATCH] Add safe SQLite-backed blob garbage collection --- src/app/manifest.rs | 22 ++++++--- src/app/materialize.rs | 20 ++++++++ src/iroh.rs | 90 ++++++++++++++++++++++++++++++++-- src/iroh/tests.rs | 75 ++++++++++++++++++++++++++-- src/storage.rs | 51 +++++++++++++++++++ src/storage/live_blobs.rs | 56 +++++++++++++++++++++ src/storage/tests.rs | 100 +++++++++++++++++++++++++++++++++++++- 7 files changed, 396 insertions(+), 18 deletions(-) create mode 100644 src/storage/live_blobs.rs diff --git a/src/app/manifest.rs b/src/app/manifest.rs index e2d1a14..fb0fdf2 100644 --- a/src/app/manifest.rs +++ b/src/app/manifest.rs @@ -12,7 +12,7 @@ use crate::{ app::{MAX_CONCURRENT_BLOB_TRANSFERS, SyncContext}, domain::{DeviceId, Entry, EntryKind, Manifest}, filesystem::{IgnoreRules, collect_entries, relative_path}, - iroh::NodeHost, + iroh::{NodeHost, TempTag}, storage::FileFingerprint, }; @@ -100,12 +100,18 @@ where let imported_file_count = files_to_import.len(); let mut imported_batch_count = 0; let mut imported_files_so_far = 0; + let mut import_temp_tags: Vec = Vec::new(); while !files_to_import.is_empty() { let batch_size = files_to_import.len().min(SOURCE_PUBLICATION_BATCH_FILES); let files = files_to_import.drain(..batch_size).collect(); let imported_files = import_files(context.node, files).await?; - imported_files_so_far += imported_files.len(); - save_imported_files(context, &previous, &mut entries, &device_id, imported_files)?; + let (candidates, temp_tags): (Vec<(FileCandidate, String)>, Vec) = imported_files + .into_iter() + .map(|(candidate, hash, temp_tag)| ((candidate, hash), temp_tag)) + .unzip(); + imported_files_so_far += candidates.len(); + import_temp_tags.extend(temp_tags); + save_imported_files(context, &previous, &mut entries, &device_id, candidates)?; imported_batch_count += 1; let staged_manifest = Manifest { folder_id: context.folder.id, @@ -120,6 +126,9 @@ where ); publish_stage(staged_manifest).await?; } + // The manifest now references every imported blob, so the temp-tag guards + // are no longer needed — the GC callback will discover the hashes via SQLite. + import_temp_tags.clear(); context .state_store .remove_missing_file_fingerprints(context.folder.id, &observed_file_paths)?; @@ -215,17 +224,18 @@ struct FileCandidate { async fn import_files( node: &NodeHost, files: Vec, -) -> anyhow::Result> { +) -> anyhow::Result> { stream::iter(files) .map(|candidate| async move { - let blob_hash = node.import_file(&candidate.file_path).await?.to_string(); + let (hash, temp_tag) = node.import_file(&candidate.file_path).await?; let current_metadata = fs::metadata(&candidate.file_path)?; let current_fingerprint = FileFingerprint::from_metadata(¤t_metadata)?; if current_fingerprint != candidate.fingerprint { tracing::debug!(path = %candidate.path, "File changed during blob import; retrying on the next scan"); + drop(temp_tag); return Ok(None); } - Ok::<_, anyhow::Error>(Some((candidate, blob_hash))) + Ok::<_, anyhow::Error>(Some((candidate, hash.to_string(), temp_tag))) }) .buffer_unordered(MAX_CONCURRENT_BLOB_TRANSFERS) .try_filter_map(|candidate| async move { Ok(candidate) }) diff --git a/src/app/materialize.rs b/src/app/materialize.rs index 6306338..ed6aa23 100644 --- a/src/app/materialize.rs +++ b/src/app/materialize.rs @@ -406,6 +406,9 @@ async fn receive_and_materialize_files( plan: &mut MaterializationPlan, downloads: Vec, ) -> anyhow::Result<()> { + // Protect every blob we are about to download from garbage collection until + // the manifest references it (which happens inside `materialize_entries`). + let _download_guards = protect_download_hashes(context.node, &downloads).await; let mut downloads = stream::iter(plan_download_batches(downloads)) .map(|files| download_files_with_isolation(context, peer, files)) .buffer_unordered(MAX_CONCURRENT_BLOB_TRANSFERS); @@ -425,6 +428,23 @@ async fn receive_and_materialize_files( Ok(()) } +async fn protect_download_hashes( + node: &NodeHost, + downloads: &[RemoteFile], +) -> Vec { + let hashes = downloads + .iter() + .filter_map(|file| iroh_blobs::Hash::from_str(&file.blob_hash).ok()) + .collect::>(); + match node.protect_blobs(&hashes).await { + Ok(guards) => guards, + Err(error) => { + tracing::warn!(%error, "Could not protect download blobs from GC; relying on GC timing"); + Vec::new() + } + } +} + fn plan_download_batches(downloads: Vec) -> Vec> { let mut batches = Vec::new(); let mut small_files = Vec::new(); diff --git a/src/iroh.rs b/src/iroh.rs index 424971f..720dfa2 100644 --- a/src/iroh.rs +++ b/src/iroh.rs @@ -1,8 +1,9 @@ //! Shared Iroh endpoint, authenticated folder control streams, and blob transfers. use std::{ - collections::{BTreeMap, BTreeSet}, + collections::{BTreeMap, BTreeSet, HashSet}, path::Path, + str::FromStr, sync::{Arc, Mutex, MutexGuard}, }; @@ -10,11 +11,12 @@ use ::iroh as iroh_network; use anyhow::Context; use futures_util::{StreamExt, stream}; use iroh_blobs::{ - BlobsProtocol, Hash, + BlobsProtocol, Hash, HashAndFormat, api::remote::{GetProgress, GetProgressItem}, protocol::GetManyRequest, provider::events::{EventMask, EventSender, ProviderMessage, RequestMode}, store::fs::FsStore, + store::{GcConfig, ProtectCb, ProtectOutcome}, }; use iroh_mdns_address_lookup::{DiscoveryEvent, MdnsAddressLookup}; use iroh_network::{ @@ -36,6 +38,9 @@ mod handler; use handler::AppaProtocol; /// ALPN for Appa's Iroh control streams. A new value is wire-incompatible. pub const APPA_ALPN: &[u8] = b"appa/sync/4"; + +/// Re-exported so callers can hold temp-tag guards across blob imports. +pub use iroh_blobs::api::TempTag; const APPA_MDNS_SERVICE_NAME: &str = "appa"; const MAX_CONTROL_MESSAGE_BYTES: usize = 16 * 1024 * 1024; pub(crate) const MAX_AUDIT_EVENTS_PER_RESPONSE: usize = 256; @@ -122,7 +127,7 @@ impl NodeHost { .secret_key(identity) .bind() .await?; - let store = FsStore::load(&blob_directory).await?; + let store = Self::load_store_with_gc(&blob_directory, data_directory).await?; let folders = Arc::new(RwLock::new(BTreeMap::new())); let discovered_peers = Arc::new(Mutex::new(BTreeMap::new())); let announced_folders = Arc::new(Mutex::new(BTreeMap::new())); @@ -204,8 +209,29 @@ impl NodeHost { Ok(()) } - pub async fn import_file(&self, file_path: &Path) -> anyhow::Result { - Ok(self.store.blobs().add_path(file_path).await?.hash) + pub async fn import_file(&self, file_path: &Path) -> anyhow::Result<(Hash, TempTag)> { + let temp_tag = self.store.blobs().add_path(file_path).temp_tag().await?; + Ok((temp_tag.hash(), temp_tag)) + } + + /// Creates temp tags that protect the given blobs from garbage collection until + /// the returned tags are dropped. Used to keep downloaded data alive between + /// receiving it and persisting the manifest that references it. + pub async fn protect_blobs(&self, hashes: &[Hash]) -> anyhow::Result> { + let tags = self.store.tags(); + let mut temp_tags = Vec::with_capacity(hashes.len()); + for &hash in hashes { + temp_tags.push(tags.temp_tag(HashAndFormat::raw(hash)).await?); + } + Ok(temp_tags) + } + + /// Removes every named tag from the store. Appa's source of truth for live + /// blobs is the SQLite manifest database, so legacy persistent tags created by + /// older Appa versions can be dropped safely before the next garbage-collection + /// cycle reclaims the orphaned bytes. + pub async fn sweep_legacy_tags(&self) -> anyhow::Result { + Ok(self.store.tags().delete_all().await?) } pub async fn download_blob( @@ -515,6 +541,32 @@ impl NodeHost { .collect() } + /// Loads the filesystem blob store with automatic garbage collection. + /// + /// GC runs on a fixed interval (configurable via `APPA_GC_INTERVAL_SECS`). + /// Before each sweep the [`ProtectCb`] callback opens a fresh read-only + /// connection to the SQLite state and feeds every live blob hash — those + /// referenced by current manifests, retained history revisions, and pending + /// materializations — into the GC's live set so they survive the sweep. + async fn load_store_with_gc( + blob_directory: &Path, + data_directory: &Path, + ) -> anyhow::Result { + let database_path = data_directory.join("state.sqlite3"); + let db_path = blob_directory.join("blobs.db"); + let mut options = iroh_blobs::store::fs::options::Options::new(blob_directory); + let interval = crate::storage::gc_interval()?; + let add_protected: ProtectCb = Arc::new(move |live: &mut HashSet| { + let path = database_path.clone(); + Box::pin(async move { protect_live_blobs(&path, live) }) + }); + options.gc = Some(GcConfig { + interval, + add_protected: Some(add_protected), + }); + Ok(FsStore::load_with_opts(db_path, options).await?) + } + pub async fn shutdown(self) -> anyhow::Result<()> { if let Some(task) = self.lan_discovery_task { task.abort(); @@ -526,6 +578,34 @@ impl NodeHost { } } +fn add_live_blobs_from_state(database_path: &Path, live: &mut HashSet) -> anyhow::Result<()> { + let live_hashes = crate::storage::StateStore::live_blob_hashes_for_gc(database_path)?; + if live_hashes.is_empty() { + tracing::trace!("GC live set is empty; protecting nothing extra"); + } + for hash_str in live_hashes { + match Hash::from_str(&hash_str) { + Ok(hash) => { + live.insert(hash); + } + Err(error) => { + tracing::warn!(%hash_str, %error, "Could not parse a blob hash from Appa state; skipping"); + } + } + } + Ok(()) +} + +fn protect_live_blobs(database_path: &Path, live: &mut HashSet) -> ProtectOutcome { + match add_live_blobs_from_state(database_path, live) { + Ok(()) => ProtectOutcome::Continue, + Err(error) => { + tracing::warn!(%error, "Could not compute live blob set for GC; aborting sweep"); + ProtectOutcome::Abort + } + } +} + pub(super) fn recover_mutex(mutex: &Mutex) -> MutexGuard<'_, T> { mutex .lock() diff --git a/src/iroh/tests.rs b/src/iroh/tests.rs index 09e6b29..6c3d593 100644 --- a/src/iroh/tests.rs +++ b/src/iroh/tests.rs @@ -1,10 +1,13 @@ -use std::{collections::BTreeSet, fs}; +use std::{ + collections::{BTreeSet, HashSet}, + fs, +}; use iroh::SecretKey; use tempfile::TempDir; use uuid::Uuid; -use super::{FolderSession, MAX_AUDIT_EVENTS_PER_RESPONSE, NodeHost}; +use super::{FolderSession, MAX_AUDIT_EVENTS_PER_RESPONSE, NodeHost, protect_live_blobs}; fn test_roster( folder_id: Uuid, @@ -44,7 +47,7 @@ async fn transfers_a_file_between_shared_blob_stores() -> anyhow::Result<()> { let target = NodeHost::load_with_lan_discovery(target_directory.path(), SecretKey::generate(), false) .await?; - let blob_hash = source.import_file(&source_file).await?; + let (blob_hash, _temp_tag) = source.import_file(&source_file).await?; source.wait_until_online().await?; target.wait_until_online().await?; target @@ -71,8 +74,8 @@ async fn transfers_tiny_files_in_one_blob_request() -> anyhow::Result<()> { let second_source = source_directory.path().join("second.txt"); fs::write(&first_source, "first")?; fs::write(&second_source, "second")?; - let first_hash = source.import_file(&first_source).await?; - let second_hash = source.import_file(&second_source).await?; + let (first_hash, _first_tag) = source.import_file(&first_source).await?; + let (second_hash, _second_tag) = source.import_file(&second_source).await?; source.wait_until_online().await?; target.wait_until_online().await?; @@ -99,6 +102,68 @@ async fn transfers_tiny_files_in_one_blob_request() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +async fn import_file_uses_temp_tags_not_persistent_named_tags() -> anyhow::Result<()> { + use futures_util::StreamExt; + + let directory = TempDir::new()?; + let node = + NodeHost::load_with_lan_discovery(directory.path(), SecretKey::generate(), false).await?; + let file = directory.path().join("hello.txt"); + fs::write(&file, "protected content")?; + + let (hash, _temp_tag) = node.import_file(&file).await?; + assert!(node.store.blobs().has(hash).await?); + + // import_file must not leave a persistent named tag behind; only temp tags + // should protect the blob so GC can reclaim it once the manifest references it. + let named_tags: Vec<_> = node.store.tags().list().await?.collect().await; + assert!( + named_tags.is_empty(), + "import_file should not create persistent named tags" + ); + + node.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn sweep_legacy_tags_removes_persistent_named_tags() -> anyhow::Result<()> { + use futures_util::StreamExt; + + let directory = TempDir::new()?; + let node = + NodeHost::load_with_lan_discovery(directory.path(), SecretKey::generate(), false).await?; + let file = directory.path().join("legacy.txt"); + fs::write(&file, "old import style")?; + + // Simulate the old behavior: add_path().await creates a persistent named tag. + let tag_info = node.store.blobs().add_path(&file).await?; + let named_before: Vec<_> = node.store.tags().list().await?.collect().await; + assert_eq!(named_before.len(), 1); + + let swept = node.sweep_legacy_tags().await?; + assert_eq!(swept, 1); + let named_after: Vec<_> = node.store.tags().list().await?.collect().await; + assert!(named_after.is_empty()); + + drop(tag_info); + node.shutdown().await?; + Ok(()) +} + +#[test] +fn gc_protection_aborts_when_the_state_database_cannot_be_read() { + let directory = TempDir::new().expect("temporary directory"); + let mut live = HashSet::new(); + + assert!(matches!( + protect_live_blobs(directory.path(), &mut live), + iroh_blobs::store::ProtectOutcome::Abort + )); + assert!(live.is_empty()); +} + #[tokio::test] async fn announces_folder_changes_to_an_authorized_peer() -> anyhow::Result<()> { let source_directory = TempDir::new()?; diff --git a/src/storage.rs b/src/storage.rs index 10fe13f..9374a62 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -18,6 +18,7 @@ use crate::{ mod audit; mod fingerprints; +mod live_blobs; mod manifests; mod materializations; mod paths; @@ -94,6 +95,33 @@ impl StateStore { Ok(store) } + /// Opens a short-lived connection for the garbage collector's `add_protected` + /// callback. This runs on the blob store's own runtime and needs independent + /// access to the SQLite database to compute the set of live blob hashes. + pub(crate) fn open_for_gc(database_path: &Path) -> anyhow::Result { + const DATABASE_BUSY_TIMEOUT_MILLIS: u64 = 5_000; + let store = Self { + connection: Connection::open(database_path).with_context(|| { + format!( + "could not open Appa state for GC at {}", + database_path.display() + ) + })?, + // The history limit is irrelevant for a read-only live-set query; use + // the default so validation never rejects the connection. + history_limit: DEFAULT_HISTORY_REVISIONS, + }; + store + .connection + .busy_timeout(std::time::Duration::from_millis( + DATABASE_BUSY_TIMEOUT_MILLIS, + ))?; + store + .connection + .execute_batch("PRAGMA journal_mode = WAL; PRAGMA query_only = ON;")?; + Ok(store) + } + pub fn register_folder(&self, path: &Path) -> anyhow::Result { if !path.is_dir() { anyhow::bail!("{} is not a directory", path.display()); @@ -318,6 +346,29 @@ fn validate_history_revision_limit(limit: usize) -> anyhow::Result { Ok(limit) } +const DEFAULT_GC_INTERVAL_SECS: u64 = 10 * 60; +const MIN_GC_INTERVAL_SECS: u64 = 60; + +pub fn gc_interval() -> anyhow::Result { + match env::var("APPA_GC_INTERVAL_SECS") { + Ok(value) => value + .parse() + .map_err(|_| anyhow::anyhow!("APPA_GC_INTERVAL_SECS must be a positive integer")) + .and_then(validate_gc_interval), + Err(env::VarError::NotPresent) => { + Ok(std::time::Duration::from_secs(DEFAULT_GC_INTERVAL_SECS)) + } + Err(error) => Err(error.into()), + } +} + +fn validate_gc_interval(secs: u64) -> anyhow::Result { + if secs < MIN_GC_INTERVAL_SECS { + anyhow::bail!("APPA_GC_INTERVAL_SECS must be at least {MIN_GC_INTERVAL_SECS} seconds"); + } + Ok(std::time::Duration::from_secs(secs)) +} + fn prune_manifest_history( transaction: &rusqlite::Transaction<'_>, folder_id: FolderId, diff --git a/src/storage/live_blobs.rs b/src/storage/live_blobs.rs new file mode 100644 index 0000000..c3bef90 --- /dev/null +++ b/src/storage/live_blobs.rs @@ -0,0 +1,56 @@ +use std::collections::HashSet; + +use crate::{ + domain::{Entry, EntryKind, Manifest}, + storage::StateStore, +}; + +impl StateStore { + /// Returns every blob hash that must be protected from garbage collection. + /// + /// A blob is live when it is referenced by a current manifest, a retained + /// manifest-history revision (so `appa folder restore` still works), or a + /// pending materialization (an interrupted write that will be replayed). + pub fn live_blob_hashes(&self) -> anyhow::Result> { + let mut live = HashSet::new(); + for folder in self.folders()? { + let manifest = self.load_manifest(folder.id)?; + collect_manifest_blob_hashes(&manifest, &mut live); + + for revision in self.manifest_history(folder.id)? { + collect_manifest_blob_hashes(&revision.manifest, &mut live); + } + if let Some(pending) = self.pending_materialization(folder.id)? { + for entry in &pending.entries { + collect_entry_blob_hashes(entry, &mut live); + } + collect_manifest_blob_hashes(&pending.resulting_manifest, &mut live); + } + } + Ok(live) + } + + /// Like [`live_blob_hashes`](Self::live_blob_hashes) but driven from a fresh + /// connection opened against `database_path`. Used by the garbage collector + /// callback, which runs on the blob store's own runtime and cannot borrow the + /// daemon's `StateStore`. + pub(crate) fn live_blob_hashes_for_gc( + database_path: &std::path::Path, + ) -> anyhow::Result> { + StateStore::open_for_gc(database_path)?.live_blob_hashes() + } +} + +fn collect_manifest_blob_hashes(manifest: &Manifest, live: &mut HashSet) { + for entry in manifest.entries.values() { + collect_entry_blob_hashes(entry, live); + } +} + +fn collect_entry_blob_hashes(entry: &Entry, live: &mut HashSet) { + if entry.kind == EntryKind::File + && let Some(blob_hash) = &entry.blob_hash + { + live.insert(blob_hash.clone()); + } +} diff --git a/src/storage/tests.rs b/src/storage/tests.rs index 90b85d4..03e14d5 100644 --- a/src/storage/tests.rs +++ b/src/storage/tests.rs @@ -4,8 +4,8 @@ use tempfile::TempDir; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use super::{ - AppPaths, DEFAULT_HISTORY_REVISIONS, FileFingerprint, StateStore, prune_manifest_history, - validate_history_revision_limit, + AppPaths, DEFAULT_HISTORY_REVISIONS, FileFingerprint, StateStore, gc_interval, + prune_manifest_history, validate_history_revision_limit, }; #[test] @@ -371,3 +371,99 @@ fn rejects_a_zero_history_revision_limit() { fn keeps_ten_revisions_by_default() { assert_eq!(DEFAULT_HISTORY_REVISIONS, 10); } + +#[test] +fn gc_live_blob_query_fails_when_state_database_is_unavailable() { + let directory = TempDir::new().expect("temporary directory"); + assert!(StateStore::live_blob_hashes_for_gc(directory.path()).is_err()); +} + +#[test] +fn live_blob_query_propagates_manifest_read_failures() -> anyhow::Result<()> { + let directory = TempDir::new()?; + let folder_path = directory.path().join("folder"); + fs::create_dir(&folder_path)?; + let paths = AppPaths::from_data_directory(directory.path().join("state"))?; + let store = StateStore::open(&paths)?; + store.register_folder(&folder_path)?; + store.connection.execute_batch("DROP TABLE manifests")?; + + assert!(store.live_blob_hashes().is_err()); + Ok(()) +} + +#[test] +fn live_blob_hashes_include_current_manifest_history_and_pending() -> anyhow::Result<()> { + let directory = TempDir::new()?; + let folder_path = directory.path().join("folder"); + fs::create_dir(&folder_path)?; + let paths = AppPaths::from_data_directory(directory.path().join("state"))?; + let store = StateStore::open(&paths)?; + let folder = store.register_folder(&folder_path)?; + + let mut manifest = store.load_manifest(folder.id)?; + manifest.entries.insert( + "one.txt".to_owned(), + test_file_entry_with_hash("one.txt", "blob-1"), + ); + manifest.entries.insert( + "two.txt".to_owned(), + test_file_entry_with_hash("two.txt", "blob-2"), + ); + store.save_manifest(&manifest)?; + + let mut historical = manifest.clone(); + historical.entries.insert( + "old.txt".to_owned(), + test_file_entry_with_hash("old.txt", "blob-old"), + ); + store.save_manifest(&historical)?; + let mut current = historical.clone(); + current.entries.remove("old.txt"); + store.save_manifest(¤t)?; + + let pending_entry = test_file_entry_with_hash("pending.txt", "blob-pending"); + store.save_pending_materialization(folder.id, &[pending_entry], ¤t)?; + + let live = store.live_blob_hashes()?; + assert!(live.contains("blob-1")); + assert!(live.contains("blob-2")); + assert!(live.contains("blob-old")); + assert!(live.contains("blob-pending")); + + store.clear_pending_materialization(folder.id)?; + let live_after_clear = store.live_blob_hashes()?; + assert!(!live_after_clear.contains("blob-pending")); + Ok(()) +} + +#[test] +fn live_blob_hashes_are_empty_for_a_folder_with_no_files() -> anyhow::Result<()> { + let directory = TempDir::new()?; + let folder_path = directory.path().join("folder"); + fs::create_dir(&folder_path)?; + let paths = AppPaths::from_data_directory(directory.path().join("state"))?; + let store = StateStore::open(&paths)?; + store.register_folder(&folder_path)?; + assert!(store.live_blob_hashes()?.is_empty()); + Ok(()) +} + +#[test] +fn gc_interval_has_a_default_and_rejects_values_below_the_minimum() -> anyhow::Result<()> { + let interval = gc_interval()?; + assert!(interval.as_secs() >= 60); + Ok(()) +} + +fn test_file_entry_with_hash(path: &str, blob_hash: &str) -> crate::domain::Entry { + crate::domain::Entry { + path: path.to_owned(), + kind: crate::domain::EntryKind::File, + blob_hash: Some(blob_hash.to_owned()), + size_bytes: Some(1), + modified_at: OffsetDateTime::now_utc(), + clock: Default::default(), + author_device_id: "device".to_owned(), + } +} -- 2.51.2