Move the handling of the long running list blobs cmd into the metadata db

This makes the metadata db more self-contained - the level above does not need to know about snapshots.

Author: rklaehn

src/store/fs.rs

@@ -94,7 +94,6 @@ use entry_state::{DataLocation, OutboardLocation};
 use gc::run_gc;
 use import::{ImportEntry, ImportSource};
 use irpc::channel::mpsc;
-use meta::list_blobs;
 use n0_future::{future::yield_now, io};
 use nested_enum_utils::enum_conversions;
 use range_collections::range_set::RangeSetRange;
@@ -507,9 +506,7 @@ impl Actor {
             }
             Command::ListBlobs(cmd) => {
                 trace!("{cmd:?}");
-                if let Ok(snapshot) = self.db().snapshot(cmd.span.clone()).await {
-                    self.spawn(list_blobs(snapshot, cmd));
-                }
+                self.db().send(cmd.into()).await.ok();
             }
             Command::Batch(cmd) => {
                 trace!("{cmd:?}");

src/store/fs/meta.rs

@@ -15,7 +15,7 @@ use n0_snafu::SpanTrace;
 use nested_enum_utils::common_fields;
 use redb::{Database, DatabaseError, ReadableTable};
 use snafu::{Backtrace, ResultExt, Snafu};
-use tokio::pin;
+use tokio::{pin, task::JoinSet};
 
 use crate::{
     api::{
@@ -96,15 +96,6 @@ impl Db {
         Self { sender }
     }
 
-    pub async fn snapshot(&self, span: tracing::Span) -> io::Result<ReadOnlyTables> {
-        let (tx, rx) = tokio::sync::oneshot::channel();
-        self.sender
-            .send(Snapshot { tx, span }.into())
-            .await
-            .map_err(|_| io::Error::other("send snapshot"))?;
-        rx.await.map_err(|_| io::Error::other("receive snapshot"))
-    }
-
     pub async fn update_await(&self, hash: Hash, state: EntryState<Bytes>) -> io::Result<()> {
         let (tx, rx) = oneshot::channel();
         self.sender
@@ -463,6 +454,7 @@ pub struct Actor {
     ds: DeleteHandle,
     options: BatchOptions,
     protected: HashSet<Hash>,
+    tasks: JoinSet<()>,
 }
 
 impl Actor {
@@ -492,6 +484,7 @@ impl Actor {
             ds,
             options,
             protected: Default::default(),
+            tasks: JoinSet::new(),
         })
     }
 
@@ -707,6 +700,7 @@ impl Actor {
 
     async fn handle_toplevel(
         db: &mut Database,
+        tasks: &mut JoinSet<()>,
         cmd: TopLevelCommand,
         op: TxnNum,
     ) -> ActorResult<Option<ShutdownMsg>> {
@@ -726,11 +720,11 @@ impl Actor {
                 // nothing to do here, since the database will be dropped
                 Some(cmd)
             }
-            TopLevelCommand::Snapshot(cmd) => {
+            TopLevelCommand::ListBlobs(cmd) => {
                 trace!("{cmd:?}");
                 let txn = db.begin_read().context(TransactionSnafu)?;
                 let snapshot = ReadOnlyTables::new(&txn).context(TableSnafu)?;
-                cmd.tx.send(snapshot).ok();
+                tasks.spawn(list_blobs(snapshot, cmd));
                 None
             }
         })
@@ -741,14 +735,20 @@ impl Actor {
         let options = &self.options;
         let mut op = 0u64;
         let shutdown = loop {
+            let cmd = tokio::select! {
+                cmd = self.cmds.recv() => cmd,
+                _ = self.tasks.join_next(), if !self.tasks.is_empty() => continue,
+            };
             op += 1;
-            let Some(cmd) = self.cmds.recv().await else {
+            let Some(cmd) = cmd else {
                 break None;
             };
             match cmd {
                 Command::TopLevel(cmd) => {
                     let op = TxnNum::TopLevel(op);
-                    if let Some(shutdown) = Self::handle_toplevel(&mut db, cmd, op).await? {
+                    if let Some(shutdown) =
+                        Self::handle_toplevel(&mut db, &mut self.tasks, cmd, op).await?
+                    {
                         break Some(shutdown);
                     }
                 }

src/store/fs/meta/proto.rs

@@ -5,11 +5,11 @@ use bytes::Bytes;
 use nested_enum_utils::enum_conversions;
 use tracing::Span;
 
-use super::{ActorResult, ReadOnlyTables};
+use super::ActorResult;
 use crate::{
     api::proto::{
-        BlobStatusMsg, ClearProtectedMsg, DeleteBlobsMsg, ProcessExitRequest, ShutdownMsg,
-        SyncDbMsg,
+        BlobStatusMsg, ClearProtectedMsg, DeleteBlobsMsg, ListBlobsMsg, ProcessExitRequest,
+        ShutdownMsg, SyncDbMsg,
     },
     store::{fs::entry_state::EntryState, util::DD},
     util::channel::oneshot,
@@ -49,12 +49,6 @@ pub struct Dump {
     pub span: Span,
 }
 
-#[derive(Debug)]
-pub struct Snapshot {
-    pub(crate) tx: tokio::sync::oneshot::Sender<ReadOnlyTables>,
-    pub span: Span,
-}
-
 pub struct Update {
     pub hash: Hash,
     pub state: EntryState<Bytes>,
@@ -167,7 +161,7 @@ impl ReadWriteCommand {
 pub enum TopLevelCommand {
     SyncDb(SyncDbMsg),
     Shutdown(ShutdownMsg),
-    Snapshot(Snapshot),
+    ListBlobs(ListBlobsMsg),
 }
 
 impl TopLevelCommand {
@@ -181,7 +175,7 @@ impl TopLevelCommand {
         match self {
             Self::SyncDb(x) => x.parent_span_opt(),
             Self::Shutdown(x) => x.parent_span_opt(),
-            Self::Snapshot(x) => Some(&x.span),
+            Self::ListBlobs(x) => Some(&x.span),
         }
     }
 }