fix(storage): reclaim transfer cache and track received files

crates/vnidrop/CORE_FLOW.md

@@ -82,10 +82,14 @@ bytes through Kotlin memory.
 ## Blob Retention Policy
 
 Stopping a share immediately removes its provider mapping and approval state,
-so outstanding VniDrop tickets can no longer download content. Physical blob chunks are
-not force-deleted at stop time because content-addressed chunks may be shared by
-another active collection. They remain eligible for the blob store's garbage
-collection. Restart reconciliation never restores a stopped share.
+so outstanding VniDrop tickets can no longer download content. Physical blob
+chunks are not force-deleted at stop time because content-addressed chunks may be
+shared by another active collection. Every active outgoing share has a persistent
+`vnidrop/share/<local-id>` tag; stopping or deleting it removes that tag, and the
+configured garbage collector later reclaims content with no remaining persistent
+or temporary tag. Receive downloads keep a temporary tag through export and become
+reclaimable after publication. Restart reconciliation repairs active-share tags,
+removes orphan share tags, and never restores a stopped share.
 
 ## Resource Limits
 

crates/vnidrop/src/api.rs

@@ -127,6 +127,60 @@ pub trait ReceiveOutputSink: Send + Sync {
     ) -> Result<(), crate::error::VnidropError>;
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, uniffi::Enum)]
+pub enum ReceivedLocatorKind {
+    FilesystemPath,
+    AndroidMediaStore,
+    AndroidDocument,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, uniffi::Record)]
+pub struct PublishedOutput {
+    pub locator_kind: ReceivedLocatorKind,
+    pub locator: String,
+}
+
+/// Versioned receive sink that reports the durable locator created at publish time.
+#[uniffi::export(with_foreign)]
+pub trait ReceiveOutputSinkV2: Send + Sync {
+    fn start_file(&self, relative_path: String) -> Result<(), crate::error::VnidropError>;
+    fn write_chunk(
+        &self,
+        relative_path: String,
+        bytes: Vec<u8>,
+    ) -> Result<(), crate::error::VnidropError>;
+    fn finish_file(
+        &self,
+        relative_path: String,
+    ) -> Result<PublishedOutput, crate::error::VnidropError>;
+    fn abort_file(
+        &self,
+        relative_path: String,
+        reason: String,
+    ) -> Result<(), crate::error::VnidropError>;
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, uniffi::Record)]
+pub struct ReceivedArtifact {
+    pub id: String,
+    pub transfer_local_id: String,
+    pub protocol_transfer_id: u64,
+    pub relative_path: String,
+    pub locator_kind: ReceivedLocatorKind,
+    pub locator: String,
+    pub logical_size: u64,
+    pub published_at: i64,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, uniffi::Record)]
+pub struct CoreStorageUsage {
+    pub blob_store_bytes: u64,
+    pub database_bytes: u64,
+    pub logs_bytes: u64,
+    pub previews_bytes: u64,
+    pub other_core_bytes: u64,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
 pub struct RuntimeStatus {
     pub endpoint_id: String,

crates/vnidrop/src/filesystem.rs

@@ -109,6 +109,10 @@ impl AtomicOutputFile {
         self.committed = true;
         Ok(())
     }
+
+    pub(crate) fn target(&self) -> &Path {
+        &self.target
+    }
 }
 
 /// Publish a fully written temporary file as the final destination without

crates/vnidrop/src/lib.rs

@@ -14,7 +14,8 @@ mod transfer_state;
 mod util;
 
 pub use api::{
-    default_core_limits, CoreEvent, CoreEventSink, CoreLimits, ReceiveOutputSink, ReceiverRequest,
+    default_core_limits, CoreEvent, CoreEventSink, CoreLimits, CoreStorageUsage, PublishedOutput,
+    ReceiveOutputSink, ReceiveOutputSinkV2, ReceivedArtifact, ReceivedLocatorKind, ReceiverRequest,
     RuntimeStatus, ShareMetadataInput, ShareResult, ShareSource, SourceKind, StoredTransfer,
     TicketInspection, TransferAccessMode, TransferMetadata,
 };

crates/vnidrop/src/repository.rs

@@ -15,12 +15,12 @@ use uuid::Uuid;
 
 use crate::{
     access_policy::mode_from_storage,
-    api::{CoreEvent, ReceiverRequest, StoredTransfer},
+    api::{CoreEvent, ReceivedArtifact, ReceivedLocatorKind, ReceiverRequest, StoredTransfer},
     transfer_state::{ReceiverRequestStatus, TransferDirection, TransferStatus},
     util::now_ms,
 };
 
-const SCHEMA_VERSION: i64 = 4;
+const SCHEMA_VERSION: i64 = 5;
 
 #[derive(Debug, Clone)]
 pub(crate) struct Repository {
@@ -47,6 +47,7 @@ pub(crate) struct TransferUpsert<'a> {
 #[derive(Debug, Clone)]
 pub(crate) struct PersistedShare {
     pub(crate) transfer_id: u64,
+    pub(crate) local_id: String,
     pub(crate) content_hash: String,
     pub(crate) access_mode: String,
 }
@@ -58,6 +59,15 @@ pub(crate) struct RecoveredTransfer {
     pub(crate) previous_status: TransferStatus,
 }
 
+pub(crate) struct ReceivedArtifactInsert<'a> {
+    pub(crate) transfer_local_id: &'a str,
+    pub(crate) protocol_transfer_id: u64,
+    pub(crate) relative_path: &'a str,
+    pub(crate) locator_kind: ReceivedLocatorKind,
+    pub(crate) locator: &'a str,
+    pub(crate) logical_size: u64,
+}
+
 pub(crate) struct ReceiverRequestInsert<'a> {
     pub(crate) id: &'a str,
     pub(crate) transfer_id: u64,
@@ -169,6 +179,24 @@ impl Repository {
         .execute(&self.pool)
         .await?;
 
+        sqlx::query(
+            r#"
+            CREATE TABLE IF NOT EXISTS received_artifacts (
+                id TEXT PRIMARY KEY,
+                transfer_local_id TEXT NOT NULL,
+                protocol_transfer_id INTEGER NOT NULL,
+                relative_path TEXT NOT NULL,
+                locator_kind TEXT NOT NULL,
+                locator TEXT NOT NULL,
+                logical_size INTEGER NOT NULL,
+                published_at INTEGER NOT NULL,
+                UNIQUE(transfer_local_id, relative_path)
+            );
+            "#,
+        )
+        .execute(&self.pool)
+        .await?;
+
         sqlx::query(
             r#"
             CREATE TABLE IF NOT EXISTS transfer_events (
@@ -330,6 +358,71 @@ impl Repository {
         Ok(())
     }
 
+    pub(crate) async fn transfer_local_id(&self, transfer_id: u64) -> Result<String> {
+        let row = sqlx::query("SELECT local_id FROM transfers WHERE transfer_id = ?1")
+            .bind(to_db_id(transfer_id)?)
+            .fetch_one(&self.pool)
+            .await?;
+        Ok(row.get(0))
+    }
+
+    pub(crate) async fn record_received_artifact(
+        &self,
+        artifact: ReceivedArtifactInsert<'_>,
+    ) -> Result<()> {
+        sqlx::query(
+            r#"
+            INSERT INTO received_artifacts (
+                id, transfer_local_id, protocol_transfer_id, relative_path,
+                locator_kind, locator, logical_size, published_at
+            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
+            ON CONFLICT(transfer_local_id, relative_path) DO UPDATE SET
+                locator_kind = excluded.locator_kind,
+                locator = excluded.locator,
+                logical_size = excluded.logical_size,
+                published_at = excluded.published_at
+            "#,
+        )
+        .bind(Uuid::new_v4().to_string())
+        .bind(artifact.transfer_local_id)
+        .bind(to_db_id(artifact.protocol_transfer_id)?)
+        .bind(artifact.relative_path)
+        .bind(locator_kind_to_storage(&artifact.locator_kind))
+        .bind(artifact.locator)
+        .bind(to_db_id(artifact.logical_size)?)
+        .bind(now_ms())
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
+    pub(crate) async fn list_received_artifacts(&self) -> Result<Vec<ReceivedArtifact>> {
+        let rows = sqlx::query(
+            r#"
+            SELECT id, transfer_local_id, protocol_transfer_id, relative_path,
+                   locator_kind, locator, logical_size, published_at
+            FROM received_artifacts
+            ORDER BY published_at DESC
+            "#,
+        )
+        .fetch_all(&self.pool)
+        .await?;
+        rows.into_iter()
+            .map(|row| {
+                Ok(ReceivedArtifact {
+                    id: row.get("id"),
+                    transfer_local_id: row.get("transfer_local_id"),
+                    protocol_transfer_id: row.get::<i64, _>("protocol_transfer_id") as u64,
+                    relative_path: row.get("relative_path"),
+                    locator_kind: locator_kind_from_storage(&row.get::<String, _>("locator_kind"))?,
+                    locator: row.get("locator"),
+                    logical_size: row.get::<i64, _>("logical_size") as u64,
+                    published_at: row.get("published_at"),
+                })
+            })
+            .collect()
+    }
+
     pub(crate) async fn complete_share_import(&self, transfer: TransferUpsert<'_>) -> Result<()> {
         self.maybe_fail_write()?;
         if transfer.direction != TransferDirection::Send
@@ -508,7 +601,7 @@ impl Repository {
     pub(crate) async fn list_active_shares(&self) -> Result<Vec<PersistedShare>> {
         let rows = sqlx::query(
             r#"
-            SELECT transfer_id, content_hash, access_mode
+            SELECT transfer_id, local_id, content_hash, access_mode
             FROM transfers
             WHERE direction = 'send'
               AND status = 'sharing'
@@ -521,8 +614,9 @@ impl Repository {
             .into_iter()
             .map(|row| PersistedShare {
                 transfer_id: row.get::<i64, _>(0) as u64,
-                content_hash: row.get::<String, _>(1),
-                access_mode: row.get::<String, _>(2),
+                local_id: row.get::<String, _>(1),
+                content_hash: row.get::<String, _>(2),
+                access_mode: row.get::<String, _>(3),
             })
             .collect())
     }
@@ -870,6 +964,23 @@ fn to_db_id(value: u64) -> Result<i64> {
     i64::try_from(value).context("transfer id exceeds SQLite signed integer range")
 }
 
+fn locator_kind_to_storage(kind: &ReceivedLocatorKind) -> &'static str {
+    match kind {
+        ReceivedLocatorKind::FilesystemPath => "filesystem_path",
+        ReceivedLocatorKind::AndroidMediaStore => "android_media_store",
+        ReceivedLocatorKind::AndroidDocument => "android_document",
+    }
+}
+
+fn locator_kind_from_storage(value: &str) -> Result<ReceivedLocatorKind> {
+    match value {
+        "filesystem_path" => Ok(ReceivedLocatorKind::FilesystemPath),
+        "android_media_store" => Ok(ReceivedLocatorKind::AndroidMediaStore),
+        "android_document" => Ok(ReceivedLocatorKind::AndroidDocument),
+        _ => anyhow::bail!("unknown received artifact locator kind: {value}"),
+    }
+}
+
 fn row_to_transfer(row: sqlx::sqlite::SqliteRow) -> Result<StoredTransfer> {
     let direction = row.get::<String, _>("direction");
     let status = row.get::<String, _>("status");

crates/vnidrop/src/runtime/facade.rs

@@ -6,9 +6,9 @@ use serde_json::json;
 use super::CoreInner;
 use crate::{
     api::{
-        CoreEvent, CoreEventSink, CoreLimits, ReceiveOutputSink, ReceiverRequest, RuntimeStatus,
-        ShareMetadataInput, ShareResult, ShareSource, StoredTransfer, TicketInspection,
-        TransferAccessMode,
+        CoreEvent, CoreEventSink, CoreLimits, CoreStorageUsage, ReceiveOutputSink,
+        ReceiveOutputSinkV2, ReceivedArtifact, ReceiverRequest, RuntimeStatus, ShareMetadataInput,
+        ShareResult, ShareSource, StoredTransfer, TicketInspection, TransferAccessMode,
     },
     error::VnidropError,
     filesystem::platform_path,
@@ -126,6 +126,24 @@ impl VnidropCore {
         .map_err(VnidropError::transfer)
     }
 
+    pub fn receive_with_output_sink_v2(
+        &self,
+        ticket: String,
+        output_sink: Arc<dyn ReceiveOutputSinkV2>,
+        receiver_name: Option<String>,
+    ) -> Result<(), VnidropError> {
+        if let Err(error) = parse_transfer_ticket_with_limits(&ticket, &self.inner.limits)
+            .context("failed to parse transfer ticket")
+        {
+            return Err(VnidropError::ticket(error));
+        }
+        self.block_on(
+            self.inner
+                .receive_with_output_sink_v2(ticket, output_sink, receiver_name),
+        )
+        .map_err(VnidropError::transfer)
+    }
+
     pub fn cancel_transfer(&self, transfer_id: u64) -> Result<(), VnidropError> {
         // Fire the oneshot on this thread before entering the runtime so a
         // blocked export write cannot prevent the cancel signal from being
@@ -214,6 +232,16 @@ impl VnidropCore {
             .map_err(VnidropError::repository)
     }
 
+    pub fn list_received_artifacts(&self) -> Result<Vec<ReceivedArtifact>, VnidropError> {
+        self.block_on(self.inner.repository.list_received_artifacts())
+            .map_err(VnidropError::repository)
+    }
+
+    pub fn storage_usage(&self) -> Result<CoreStorageUsage, VnidropError> {
+        self.block_on(self.inner.storage_usage())
+            .map_err(VnidropError::filesystem)
+    }
+
     pub fn list_events(&self, transfer_id: Option<u64>) -> Result<Vec<CoreEvent>, VnidropError> {
         self.block_on(self.inner.list_events(transfer_id))
             .map_err(VnidropError::repository)

crates/vnidrop/src/runtime/lifecycle.rs

@@ -3,7 +3,7 @@ use std::sync::atomic::Ordering;
 use anyhow::Result;
 use serde_json::json;
 
-use super::CoreInner;
+use super::{share_tag_name, CoreInner};
 use crate::{
     access_policy::mode_to_storage,
     api::{RuntimeStatus, TransferAccessMode},
@@ -42,6 +42,7 @@ impl CoreInner {
     pub(super) async fn cancel_idle_or_share(&self, transfer_id: u64) -> Result<()> {
         let mut active_shares = self.active_shares.lock().await;
         if active_shares.contains_key(&transfer_id) {
+            let local_id = self.repository.transfer_local_id(transfer_id).await?;
             self.repository
                 .transition_transfer_status(
                     transfer_id,
@@ -53,6 +54,7 @@ impl CoreInner {
             drop(active_shares);
             self.unregister_transfer_hashes(transfer_id).await;
             self.access_policy.remove_transfer(transfer_id).await;
+            self.store.tags().delete(share_tag_name(&local_id)).await?;
             self.emit_transfer(transfer_id, "send", "lifecycle", "share-stopped", json!({}));
             return Ok(());
         }
@@ -105,6 +107,12 @@ impl CoreInner {
         self.active_shares.lock().await.remove(&transfer_id);
         self.unregister_transfer_hashes(transfer_id).await;
         self.access_policy.remove_transfer(transfer_id).await;
+        if transfer.direction == TransferDirection::Send.as_str() {
+            self.store
+                .tags()
+                .delete(share_tag_name(&transfer.local_id))
+                .await?;
+        }
         // Events are persisted asynchronously. Drain events emitted before this
         // request so none can be written back after the transfer is deleted.
         self.event_hub.flush().await;

crates/vnidrop/src/runtime/mod.rs

@@ -12,6 +12,7 @@ mod lifecycle;
 mod provider;
 mod receive;
 mod share;
+mod storage;
 
 pub use facade::VnidropCore;
 
@@ -20,15 +21,19 @@ use std::{
     path::PathBuf,
     str::FromStr,
     sync::{atomic::AtomicBool, Arc},
+    time::Duration,
 };
 
 use anyhow::Result;
+use futures_lite::StreamExt as _;
 use iroh::{endpoint::presets, protocol::Router, Endpoint};
 use iroh_blobs::{
-    api::TempTag,
     format::collection::Collection,
     provider::events::{EventMask, EventSender},
-    store::fs::FsStore,
+    store::{
+        fs::{options::Options as FsStoreOptions, FsStore},
+        GcConfig,
+    },
     BlobsProtocol, Hash,
 };
 use serde_json::json;
@@ -52,6 +57,7 @@ use crate::{
 /// Owns the Iroh endpoint, blob store, transfer history, and byte streaming.
 /// Kotlin owns app lifecycle and platform file picking.
 pub(super) struct CoreInner {
+    pub(super) app_data_dir: PathBuf,
     pub(super) endpoint: Endpoint,
     pub(super) router: Router,
     pub(super) store: FsStore,
@@ -64,10 +70,8 @@ pub(super) struct CoreInner {
     /// Sync mutex so cancel can remove + signal without awaiting (and without
     /// holding a Tokio lock across repository I/O).
     pub(super) active_transfers: std::sync::Mutex<HashMap<u64, ActiveTransfer>>,
-    // Newly imported shares retain a TempTag for the lifetime of this process.
-    // Restored shares have no in-memory tag, but remain tracked so they can be
-    // counted and explicitly revoked after a restart.
-    pub(super) active_shares: TokioMutex<HashMap<u64, Option<TempTag>>>,
+    // Active shares are protected by persistent Iroh tags.
+    pub(super) active_shares: TokioMutex<HashMap<u64, ()>>,
     /// Content hash → active share transfer ids (root and collection members).
     /// Multiple transfers can share the same content-addressed hash.
     pub(super) hash_to_transfer: TokioMutex<HashMap<String, HashSet<u64>>>,
@@ -92,7 +96,12 @@ impl CoreInner {
         let secret_key = load_or_create_secret(&app_data_dir).await?;
         let repository = Repository::open(&app_data_dir).await?;
         let store_root = app_data_dir.join("blobs");
-        let store = FsStore::load(&store_root).await?;
+        let mut store_options = FsStoreOptions::new(&store_root);
+        store_options.gc = Some(GcConfig {
+            interval: Duration::from_secs(30 * 60),
+            add_protected: None,
+        });
+        let store = FsStore::load_with_opts(store_root.join("blobs.db"), store_options).await?;
         let endpoint = Endpoint::builder(presets::N0)
             .secret_key(secret_key)
             .bind()
@@ -135,6 +144,7 @@ impl CoreInner {
         // Register root + every collection member so child gets stay under ACL.
         let mut restored_hashes: HashMap<String, HashSet<u64>> = HashMap::new();
         let mut restored_active_shares = HashMap::new();
+        let mut active_tag_names = HashSet::new();
         for share in repository.list_active_shares().await? {
             let transfer_id = share.transfer_id;
             let Ok(root_hash) = Hash::from_str(&share.content_hash) else {
@@ -176,6 +186,12 @@ impl CoreInner {
                 );
                 continue;
             };
+            let tag_name = share_tag_name(&share.local_id);
+            store
+                .tags()
+                .set(&tag_name, (root_hash, iroh_blobs::BlobFormat::HashSeq))
+                .await?;
+            active_tag_names.insert(tag_name);
             restored_hashes
                 .entry(root_hash.to_string())
                 .or_default()
@@ -186,11 +202,19 @@ impl CoreInner {
                     .or_default()
                     .insert(transfer_id);
             }
-            restored_active_shares.insert(transfer_id, None);
+            restored_active_shares.insert(transfer_id, ());
             access_policy
                 .set_mode(transfer_id, mode_from_storage(&share.access_mode))
                 .await;
         }
+        let mut share_tags = store.tags().list_prefix("vnidrop/share/").await?;
+        while let Some(tag) = share_tags.next().await {
+            let tag = tag?;
+            let name = String::from_utf8_lossy(tag.name.as_ref()).to_string();
+            if !active_tag_names.contains(&name) {
+                store.tags().delete(name).await?;
+            }
+        }
         let approval = ApprovalService::new(
             repository.clone(),
             event_hub.clone(),
@@ -205,6 +229,7 @@ impl CoreInner {
             .spawn();
 
         let inner = Arc::new(Self {
+            app_data_dir,
             endpoint,
             router,
             store,
@@ -277,3 +302,7 @@ impl CoreInner {
             .await
     }
 }
+
+pub(super) fn share_tag_name(local_id: &str) -> String {
+    format!("vnidrop/share/{local_id}")
+}

crates/vnidrop/src/runtime/receive.rs

@@ -17,13 +17,16 @@ use tokio::sync::oneshot;
 use super::{ActiveTransfer, CoreInner};
 use crate::{
     access_policy::mode_to_storage,
-    api::{ReceiveOutputSink, TransferAccessMode, TransferMetadata},
+    api::{
+        ReceiveOutputSink, ReceiveOutputSinkV2, ReceivedLocatorKind, TransferAccessMode,
+        TransferMetadata,
+    },
     filesystem::{
         validated_relative_string, wait_for_writer, write_stream_to_blocking_writer,
         AtomicOutputFile,
     },
     handshake::{DeliveryReceipt, DeliveryReceiptResponse, HandshakeResponse, HandshakeService},
-    repository::TransferUpsert,
+    repository::{ReceivedArtifactInsert, TransferUpsert},
     ticket::{parse_transfer_ticket_with_limits, ParsedTransferTicket},
     transfer_state::{TransferDirection, TransferStatus},
 };
@@ -31,6 +34,13 @@ use crate::{
 pub(super) enum ReceiveTarget {
     Directory(PathBuf),
     OutputSink(Arc<dyn ReceiveOutputSink>),
+    OutputSinkV2(Arc<dyn ReceiveOutputSinkV2>),
+}
+
+#[derive(Clone, Copy)]
+struct ReceivedTransfer<'a> {
+    protocol_id: u64,
+    local_id: &'a str,
 }
 
 pub(super) struct OutputSinkFile<'a> {
@@ -85,6 +95,51 @@ impl Drop for OutputSinkFile<'_> {
     }
 }
 
+pub(super) struct OutputSinkFileV2<'a> {
+    sink: &'a dyn ReceiveOutputSinkV2,
+    relative_path: String,
+    terminal: bool,
+}
+
+impl<'a> OutputSinkFileV2<'a> {
+    fn start(sink: &'a dyn ReceiveOutputSinkV2, relative_path: String) -> Result<Self> {
+        sink.start_file(relative_path.clone())
+            .map_err(|error| anyhow::anyhow!(error.to_string()))?;
+        Ok(Self {
+            sink,
+            relative_path,
+            terminal: false,
+        })
+    }
+
+    fn write(&self, bytes: Vec<u8>) -> Result<()> {
+        self.sink
+            .write_chunk(self.relative_path.clone(), bytes)
+            .map_err(|error| anyhow::anyhow!(error.to_string()))
+    }
+
+    fn finish(mut self) -> Result<crate::api::PublishedOutput> {
+        self.terminal = true;
+        self.sink
+            .finish_file(self.relative_path.clone())
+            .map_err(|error| anyhow::anyhow!(error.to_string()))
+    }
+}
+
+impl Drop for OutputSinkFileV2<'_> {
+    fn drop(&mut self) {
+        if !self.terminal {
+            self.terminal = true;
+            if let Err(error) = self.sink.abort_file(
+                self.relative_path.clone(),
+                "transfer interrupted before file completion".to_string(),
+            ) {
+                tracing::warn!(%error, relative_path = %self.relative_path, "failed to abort receive output sink file");
+            }
+        }
+    }
+}
+
 impl CoreInner {
     pub(super) async fn receive(
         self: &Arc<Self>,
@@ -110,6 +165,20 @@ impl CoreInner {
         .await
     }
 
+    pub(super) async fn receive_with_output_sink_v2(
+        self: &Arc<Self>,
+        ticket: String,
+        output_sink: Arc<dyn ReceiveOutputSinkV2>,
+        receiver_name: Option<String>,
+    ) -> Result<()> {
+        self.receive_to_target(
+            ticket,
+            ReceiveTarget::OutputSinkV2(output_sink),
+            receiver_name,
+        )
+        .await
+    }
+
     pub(super) async fn receive_to_target(
         self: &Arc<Self>,
         ticket: String,
@@ -242,9 +311,15 @@ impl CoreInner {
             json!({ "total_files": total_files, "total_size": total_size }),
         );
 
+        // Protect both partial download state and the completed collection until
+        // every output has been published and recorded.
+        let download_tag = self.store.tags().temp_tag(hash_and_format).await?;
         let get = self.store.remote().fetch(connection, hash_and_format);
         let mut stream = get.stream();
-        while let Some(item) = stream.next().await {
+        loop {
+            let Some(item) = stream.next().await else {
+                anyhow::bail!("download ended without completion");
+            };
             match item {
                 GetProgressItem::Progress(downloaded) => {
                     self.emit_transfer(
@@ -270,6 +345,7 @@ impl CoreInner {
                 TransferStatus::Done,
             )
             .await?;
+        drop(download_tag);
         self.emit_transfer(transfer_id, "receive", "lifecycle", "done", json!({}));
         let sender_transfer_id = delivery_receipt.transfer_id;
         let client = HandshakeService::client(self.endpoint.clone(), sender_addr);
@@ -393,11 +469,16 @@ impl CoreInner {
         target: ReceiveTarget,
         collection: Collection,
     ) -> Result<()> {
+        let transfer_local_id = self.repository.transfer_local_id(transfer_id).await?;
+        let received_transfer = ReceivedTransfer {
+            protocol_id: transfer_id,
+            local_id: &transfer_local_id,
+        };
         for (i, (name, hash)) in collection.iter().enumerate() {
             match &target {
                 ReceiveTarget::Directory(output_dir) => {
                     self.export_blob_to_directory(
-                        transfer_id,
+                        received_transfer,
                         total_files,
                         i as u64,
                         output_dir,
@@ -417,14 +498,25 @@ impl CoreInner {
                     )
                     .await?;
                 }
+                ReceiveTarget::OutputSinkV2(output_sink) => {
+                    self.export_blob_to_sink_v2(
+                        received_transfer,
+                        total_files,
+                        i as u64,
+                        output_sink.as_ref(),
+                        name.as_ref(),
+                        *hash,
+                    )
+                    .await?;
+                }
             }
         }
         Ok(())
     }
 
-    pub(super) async fn export_blob_to_directory(
+    async fn export_blob_to_directory(
         &self,
-        transfer_id: u64,
+        transfer: ReceivedTransfer<'_>,
         total_files: u64,
         current_file_index: u64,
         output_dir: &Path,
@@ -458,7 +550,7 @@ impl CoreInner {
                         .await
                         .map_err(|_| anyhow::anyhow!("export writer closed for {relative_path}"))?;
                     self.emit_transfer(
-                        transfer_id,
+                        transfer.protocol_id,
                         "receive",
                         "export",
                         "progress",
@@ -482,7 +574,18 @@ impl CoreInner {
             .map_err(|_| anyhow::anyhow!("export writer closed for {relative_path}"))?;
         let writer = wait_for_writer(writer_task).await??;
         tokio::task::spawn_blocking(move || writer.sync_all()).await??;
+        let locator = pending_file.target().to_string_lossy().to_string();
         pending_file.commit()?;
+        self.repository
+            .record_received_artifact(ReceivedArtifactInsert {
+                transfer_local_id: transfer.local_id,
+                protocol_transfer_id: transfer.protocol_id,
+                relative_path,
+                locator_kind: ReceivedLocatorKind::FilesystemPath,
+                locator: &locator,
+                logical_size: exported,
+            })
+            .await?;
         Ok(())
     }
 
@@ -545,4 +648,71 @@ impl CoreInner {
         output_file.finish()?;
         Ok(())
     }
+
+    async fn export_blob_to_sink_v2(
+        &self,
+        transfer: ReceivedTransfer<'_>,
+        total_files: u64,
+        current_file_index: u64,
+        output_sink: &dyn ReceiveOutputSinkV2,
+        relative_path: &str,
+        hash: Hash,
+    ) -> Result<()> {
+        if relative_path.len() as u64 > self.limits.max_path_bytes {
+            anyhow::bail!(
+                "output path exceeds {} bytes: {relative_path}",
+                self.limits.max_path_bytes
+            );
+        }
+        let relative_path = validated_relative_string(relative_path)?;
+        let output_file = OutputSinkFileV2::start(output_sink, relative_path.clone())?;
+        let mut stream = self.store.export_ranges(hash, 0..u64::MAX).stream();
+        let mut file_size = 0;
+        let mut exported = 0;
+
+        while let Some(item) = stream.next().await {
+            match item {
+                ExportRangesItem::Size(size) => file_size = size,
+                ExportRangesItem::Data(leaf) => {
+                    if leaf.offset != exported {
+                        anyhow::bail!(
+                            "export stream for {relative_path} yielded out-of-order data"
+                        );
+                    }
+                    exported += leaf.data.len() as u64;
+                    output_file.write(leaf.data.to_vec())?;
+                    self.emit_transfer(
+                        transfer.protocol_id,
+                        "receive",
+                        "export",
+                        "progress",
+                        json!({
+                            "total_files": total_files,
+                            "current_file_index": current_file_index,
+                            "file_name": relative_path,
+                            "file_size": file_size,
+                            "exported": exported,
+                        }),
+                    );
+                    tokio::task::yield_now().await;
+                }
+                ExportRangesItem::Error(error) => {
+                    anyhow::bail!("export failed for {relative_path}: {error}");
+                }
+            }
+        }
+
+        let published = output_file.finish()?;
+        self.repository
+            .record_received_artifact(ReceivedArtifactInsert {
+                transfer_local_id: transfer.local_id,
+                protocol_transfer_id: transfer.protocol_id,
+                relative_path: &relative_path,
+                locator_kind: published.locator_kind,
+                locator: &published.locator,
+                logical_size: exported,
+            })
+            .await?;
+        Ok(())
+    }
 }

crates/vnidrop/src/runtime/share.rs

@@ -12,7 +12,7 @@ use n0_future::BufferedStreamExt;
 use serde_json::json;
 use tokio::sync::oneshot;
 
-use super::{ActiveTransfer, CoreInner};
+use super::{share_tag_name, ActiveTransfer, CoreInner};
 use crate::{
     access_policy::mode_to_storage,
     api::TransferMetadata,
@@ -142,11 +142,24 @@ impl CoreInner {
             .encode()
             .context("failed to encode VniDrop transfer ticket")?;
         let content_hash = import.root_hash.to_string();
+        let local_id = self
+            .repository
+            .transfer_local_id(metadata.transfer_id)
+            .await?;
+        let tag_name = share_tag_name(&local_id);
+        self.store
+            .tags()
+            .set(
+                &tag_name,
+                (import.root_hash, iroh_blobs::BlobFormat::HashSeq),
+            )
+            .await?;
 
         // Persist the completed share before exposing it through the provider.
         // The remaining in-memory registrations are infallible and can be
         // reconstructed from SQLite if the process exits immediately after.
-        self.repository
+        if let Err(error) = self
+            .repository
             .complete_share_import(TransferUpsert {
                 transfer_id: metadata.transfer_id,
                 peer_id: None,
@@ -159,7 +172,11 @@ impl CoreInner {
                 total_size: import.total_size,
                 access_mode: mode_to_storage(&access_mode),
             })
-            .await?;
+            .await
+        {
+            let _ = self.store.tags().delete(&tag_name).await;
+            return Err(error);
+        }
         // Map root + every collection member so provider ACL cannot fail-open
         // on child blob hashes that are not the collection root.
         self.register_share_hashes(
@@ -173,7 +190,8 @@ impl CoreInner {
         self.active_shares
             .lock()
             .await
-            .insert(metadata.transfer_id, Some(import.tag));
+            .insert(metadata.transfer_id, ());
+        drop(import.tag);
 
         // Tickets are capabilities: never persist the full string in events.
         self.emit_transfer(

crates/vnidrop/src/runtime/storage.rs

@@ -0,0 +1,69 @@
+use std::path::{Path, PathBuf};
+
+use anyhow::Result;
+
+use super::CoreInner;
+use crate::api::CoreStorageUsage;
+
+impl CoreInner {
+    pub(super) async fn storage_usage(&self) -> Result<CoreStorageUsage> {
+        let app_data_dir = self.app_data_dir.clone();
+        tokio::task::spawn_blocking(move || scan_storage(&app_data_dir)).await?
+    }
+}
+
+fn scan_storage(app_data_dir: &Path) -> Result<CoreStorageUsage> {
+    let blob_store_bytes = directory_size(&app_data_dir.join("blobs"))?;
+    let logs_bytes = directory_size(&app_data_dir.join("logs"))?;
+    let previews_bytes = directory_size(&app_data_dir.join("ui").join("previews"))?;
+    let database_bytes = [
+        "vnidrop.sqlite3",
+        "vnidrop.sqlite3-wal",
+        "vnidrop.sqlite3-shm",
+    ]
+    .into_iter()
+    .try_fold(0u64, |total, name| {
+        Ok::<_, std::io::Error>(total.saturating_add(file_size(&app_data_dir.join(name))?))
+    })?;
+    let total = directory_size(app_data_dir)?;
+    let classified = blob_store_bytes
+        .saturating_add(logs_bytes)
+        .saturating_add(previews_bytes)
+        .saturating_add(database_bytes);
+    Ok(CoreStorageUsage {
+        blob_store_bytes,
+        database_bytes,
+        logs_bytes,
+        previews_bytes,
+        other_core_bytes: total.saturating_sub(classified),
+    })
+}
+
+fn directory_size(path: &Path) -> Result<u64> {
+    if !path.exists() {
+        return Ok(0);
+    }
+    let mut total = 0u64;
+    let mut pending = vec![PathBuf::from(path)];
+    while let Some(directory) = pending.pop() {
+        for entry in std::fs::read_dir(directory)? {
+            let entry = entry?;
+            let metadata = entry.metadata()?;
+            if metadata.is_dir() {
+                pending.push(entry.path());
+            } else if metadata.is_file() {
+                total = total.saturating_add(metadata.len());
+            }
+        }
+    }
+    Ok(total)
+}
+
+fn file_size(path: &Path) -> std::io::Result<u64> {
+    match std::fs::metadata(path) {
+        Ok(metadata) if metadata.is_file() => Ok(metadata.len()),
+        Ok(_) => Ok(0),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(0),
+        Err(error) => Err(error),
+    }
+}

crates/vnidrop/src/tests/repository.rs

@@ -1,6 +1,6 @@
 use crate::{
-    api::CoreEvent,
-    repository::{ReceiverRequestInsert, Repository, TransferUpsert},
+    api::{CoreEvent, ReceivedLocatorKind},
+    repository::{ReceivedArtifactInsert, ReceiverRequestInsert, Repository, TransferUpsert},
     transfer_state::{ReceiverRequestStatus, TransferDirection, TransferStatus},
 };
 
@@ -23,11 +23,47 @@ fn transfer(
     }
 }
 
+#[tokio::test]
+async fn received_artifacts_survive_history_deletion() {
+    let temp = tempfile::tempdir().unwrap();
+    let repository = Repository::open(temp.path()).await.unwrap();
+    repository
+        .start_receive(transfer(
+            91,
+            TransferDirection::Receive,
+            TransferStatus::Receiving,
+        ))
+        .await
+        .unwrap();
+    let local_id = repository.transfer_local_id(91).await.unwrap();
+    repository
+        .record_received_artifact(ReceivedArtifactInsert {
+            transfer_local_id: &local_id,
+            protocol_transfer_id: 91,
+            relative_path: "folder/file.txt",
+            locator_kind: ReceivedLocatorKind::FilesystemPath,
+            locator: "/tmp/folder/file.txt",
+            logical_size: 12,
+        })
+        .await
+        .unwrap();
+    repository
+        .transition_transfer_status(91, TransferStatus::Receiving, TransferStatus::Done)
+        .await
+        .unwrap();
+
+    assert_eq!(repository.delete_receive_history().await.unwrap(), 1);
+    let artifacts = repository.list_received_artifacts().await.unwrap();
+    assert_eq!(artifacts.len(), 1);
+    assert_eq!(artifacts[0].transfer_local_id, local_id);
+    assert_eq!(artifacts[0].logical_size, 12);
+}
+
 #[tokio::test]
 async fn persists_transfers_and_events_across_reopen() {
     let temp = tempfile::tempdir().unwrap();
     let repository = Repository::open(temp.path()).await.unwrap();
-    assert_eq!(repository.schema_version().await.unwrap(), 4);
+    assert_eq!(repository.schema_version().await.unwrap(), 5);
     repository
         .insert_transfer(transfer(
             7,
@@ -491,7 +527,7 @@ async fn migrates_schema_v2_identity_without_losing_transfer() {
     pool.close().await;
 
     let repository = Repository::open(temp.path()).await.unwrap();
-    assert_eq!(repository.schema_version().await.unwrap(), 4);
+    assert_eq!(repository.schema_version().await.unwrap(), 5);
     let stored = repository.list_transfers().await.unwrap().remove(0);
     assert_eq!(stored.transfer_id, 7);
     assert_eq!(stored.local_id, "legacy-7-send");

crates/vnidrop/tests/lifecycle.rs

@@ -3,6 +3,8 @@ mod support;
 use std::sync::{Arc, Condvar, Mutex};
 use std::time::Duration;
 
+use futures_lite::StreamExt as _;
+use iroh_blobs::store::fs::FsStore;
 use support::{share_path, CoreGuard, RecordingSink, TestNode};
 use vnidrop::{
     CoreEvent, CoreEventSink, CoreLimits, ShareMetadataInput, ShareSource, SourceKind,
@@ -77,6 +79,39 @@ fn deleting_share_revokes_it_and_removes_persisted_history() {
     assert_eq!(restarted.status().active_shares, 0);
 }
 
+#[test]
+fn active_share_tag_is_persistent_and_removed_with_transfer() {
+    let source_dir = tempfile::tempdir().unwrap();
+    let core_dir = tempfile::tempdir().unwrap();
+    let source_path = source_dir.path().join("tagged.txt");
+    std::fs::write(&source_path, b"tagged content").unwrap();
+
+    let sender = CoreGuard::start(core_dir.path(), Arc::new(RecordingSink::default()));
+    let share = share_path(&sender, &source_path, 102, "tagged.txt", false);
+    drop(sender);
+    assert_eq!(share_tag_count(core_dir.path()), 1);
+
+    let restarted = CoreGuard::start(core_dir.path(), Arc::new(RecordingSink::default()));
+    restarted.delete_transfer(share.transfer_id).unwrap();
+    drop(restarted);
+    assert_eq!(share_tag_count(core_dir.path()), 0);
+}
+
+fn share_tag_count(core_dir: &std::path::Path) -> usize {
+    let runtime = tokio::runtime::Runtime::new().unwrap();
+    runtime.block_on(async {
+        let store = FsStore::load(core_dir.join("blobs")).await.unwrap();
+        let mut tags = store.tags().list_prefix("vnidrop/share/").await.unwrap();
+        let mut count = 0;
+        while let Some(tag) = tags.next().await {
+            tag.unwrap();
+            count += 1;
+        }
+        store.shutdown().await.unwrap();
+        count
+    })
+}
+
 #[test]
 fn persisted_share_is_recovered_and_can_be_stopped_after_restart() {
     let source_dir = tempfile::tempdir().unwrap();

crates/vnidrop/tests/output_sink.rs

@@ -5,9 +5,36 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use support::{
-    receive_with_sink_response, share_path, wait_for_receiver_request, MemoryOutputSink, TestNode,
+    receive_with_sink_response, receive_with_sink_v2_response, share_path,
+    wait_for_receiver_request, MemoryOutputSink, TestNode,
 };
 
+#[test]
+fn versioned_sink_records_published_locator() {
+    let source_dir = tempfile::tempdir().unwrap();
+    let source_path = source_dir.path().join("tracked.txt");
+    std::fs::write(&source_path, b"tracked").unwrap();
+    let sender = TestNode::new();
+    let receiver = TestNode::new();
+    let share = share_path(&sender.core, &source_path, 38, "tracked.txt", false);
+    let output_sink = Arc::new(MemoryOutputSink::default());
+
+    receive_with_sink_v2_response(
+        &sender.core,
+        share.transfer_id,
+        receiver.core.arc(),
+        share.ticket,
+        output_sink,
+        true,
+    )
+    .unwrap();
+
+    let artifacts = receiver.core.list_received_artifacts().unwrap();
+    assert_eq!(artifacts.len(), 1);
+    assert_eq!(artifacts[0].locator, "content://test/tracked.txt");
+    assert_eq!(artifacts[0].logical_size, 7);
+}
+
 #[test]
 fn exports_nested_files_to_output_sink() {
     let source_dir = tempfile::tempdir().unwrap();

crates/vnidrop/tests/support/mod.rs

@@ -14,8 +14,9 @@ use std::{
 };
 
 use vnidrop::{
-    CoreEvent, CoreEventSink, CoreLimits, ReceiveOutputSink, ReceiverRequest, ShareMetadataInput,
-    ShareResult, ShareSource, SourceKind, TransferAccessMode, VnidropCore, VnidropError,
+    CoreEvent, CoreEventSink, CoreLimits, PublishedOutput, ReceiveOutputSink, ReceiveOutputSinkV2,
+    ReceivedLocatorKind, ReceiverRequest, ShareMetadataInput, ShareResult, ShareSource, SourceKind,
+    TransferAccessMode, VnidropCore, VnidropError,
 };
 
 #[derive(Default)]
@@ -219,6 +220,28 @@ impl ReceiveOutputSink for MemoryOutputSink {
     }
 }
 
+impl ReceiveOutputSinkV2 for MemoryOutputSink {
+    fn start_file(&self, relative_path: String) -> Result<(), VnidropError> {
+        ReceiveOutputSink::start_file(self, relative_path)
+    }
+
+    fn write_chunk(&self, relative_path: String, bytes: Vec<u8>) -> Result<(), VnidropError> {
+        ReceiveOutputSink::write_chunk(self, relative_path, bytes)
+    }
+
+    fn finish_file(&self, relative_path: String) -> Result<PublishedOutput, VnidropError> {
+        ReceiveOutputSink::finish_file(self, relative_path.clone())?;
+        Ok(PublishedOutput {
+            locator_kind: ReceivedLocatorKind::AndroidDocument,
+            locator: format!("content://test/{relative_path}"),
+        })
+    }
+
+    fn abort_file(&self, relative_path: String, reason: String) -> Result<(), VnidropError> {
+        ReceiveOutputSink::abort_file(self, relative_path, reason)
+    }
+}
+
 pub fn share_path(
     sender: &VnidropCore,
     source: &Path,
@@ -297,6 +320,23 @@ pub fn receive_with_sink_response(
     handle.join().unwrap()
 }
 
+pub fn receive_with_sink_v2_response(
+    sender: &VnidropCore,
+    transfer_id: u64,
+    receiver: Arc<VnidropCore>,
+    ticket: String,
+    output_sink: Arc<dyn ReceiveOutputSinkV2>,
+    accepted: bool,
+) -> Result<(), String> {
+    let handle = std::thread::spawn(move || {
+        receiver
+            .receive_with_output_sink_v2(ticket, output_sink, Some("receiver".to_string()))
+            .map_err(|error| error.to_string())
+    });
+    respond_to_pending_request(sender, transfer_id, accepted);
+    handle.join().unwrap()
+}
+
 fn respond_to_pending_request(sender: &VnidropCore, transfer_id: u64, accepted: bool) {
     let request = wait_for_receiver_request(sender, transfer_id);
     sender

crates/vnidrop/tests/transfer.rs

@@ -38,6 +38,20 @@ fn transfers_file_between_two_cores() {
         received.peer_id.as_deref(),
         Some(sender.core.status().endpoint_id.as_str())
     );
+    let artifacts = receiver.core.list_received_artifacts().unwrap();
+    assert_eq!(artifacts.len(), 1);
+    assert_eq!(artifacts[0].relative_path, "hello.txt");
+    assert_eq!(
+        artifacts[0].logical_size,
+        b"hello from vnidrop".len() as u64
+    );
+    assert_eq!(
+        artifacts[0].locator,
+        output_dir.path().join("hello.txt").to_string_lossy()
+    );
+
+    receiver.core.delete_receive_history().unwrap();
+    assert_eq!(receiver.core.list_received_artifacts().unwrap(), artifacts);
 }
 
 #[test]