another channel

Author: joshua-spacetime

crates/core/src/db/relational_db.rs

@@ -1625,13 +1625,21 @@ pub async fn local_durability(commitlog_dir: CommitLogDir) -> io::Result<(LocalD
 /// [DurabilityProvider]: crate::host::host_controller::DurabilityProvider
 pub async fn snapshot_watching_commitlog_compressor(
     mut snapshot_rx: watch::Receiver<u64>,
-    mut snapshot_tx: Option<tokio::sync::mpsc::Sender<u64>>,
+    mut clog_tx: Option<tokio::sync::mpsc::Sender<u64>>,
+    mut snap_tx: Option<tokio::sync::mpsc::Sender<u64>>,
     durability: LocalDurability,
 ) {
     let mut prev_snapshot_offset = *snapshot_rx.borrow_and_update();
     while snapshot_rx.changed().await.is_ok() {
         let snapshot_offset = *snapshot_rx.borrow_and_update();
         let durability = durability.clone();
+
+        if let Some(snap_tx) = &mut snap_tx {
+            if let Err(err) = snap_tx.try_send(snapshot_offset) {
+                tracing::warn!("failed to send offset {snapshot_offset} after creation event: {err}");
+            }
+        }
+
         let res: io::Result<_> = asyncify(move || {
             let segment_offsets = durability.existing_segment_offsets()?;
             let start_idx = segment_offsets
@@ -1662,7 +1670,7 @@ pub async fn snapshot_watching_commitlog_compressor(
         };
         prev_snapshot_offset = snapshot_offset;
 
-        if let Some(snapshot_tx) = &mut snapshot_tx {
+        if let Some(snapshot_tx) = &mut clog_tx {
             if let Err(err) = snapshot_tx.try_send(last_compressed_segment) {
                 tracing::warn!("failed to send offset {last_compressed_segment} after compression: {err}");
             }

crates/standalone/src/lib.rs

@@ -128,6 +128,7 @@ impl DurabilityProvider for StandaloneDurabilityProvider {
                 tokio::spawn(relational_db::snapshot_watching_commitlog_compressor(
                     snapshot_rx,
                     None,
+                    None,
                     durability,
                 ));
             }