[bifrost] Introduce a new LogChainWatcher for efficient subscriptions to chain mutations

crates/bifrost/src/appender.rs

@@ -16,7 +16,7 @@ use std::time::{Duration, Instant};
 use bytes::BytesMut;
 use tracing::{debug, info, instrument};
 
-use restate_core::{Metadata, MetadataKind, TaskCenter};
+use restate_core::{Metadata, TaskCenter};
 use restate_futures_util::overdue::OverdueLoggingExt;
 use restate_types::Versioned;
 use restate_types::config::Configuration;
@@ -448,12 +448,17 @@ impl<T: StorageEncode> Appender<T> {
                 }
             }
 
+            let mut wait_for_writable_segment = bifrost_inner.log_chain_watcher().subscribe(
+                log_id,
+                crate::log_chain_watcher::ChainCondition::WritableSegmentAfter(sealed_segment),
+            );
+
             tokio::select! {
                 biased;
                 // if error it means that metadata manager has stopped. We are shutting down.
                 // the check for shutdown in the loop above will catch if this happened and bubble
                 // up the shutdown error.
-                _ = metadata.wait_for_version(MetadataKind::Logs, log_metadata_version.next()) => {
+                _ = wait_for_writable_segment.wait() => {
                     // do not advance the sleep duration. Successive metadata updates that are
                     // irrelavant to this loglet should not increase the sleep duration.
                 }

crates/bifrost/src/bifrost.rs

@@ -34,6 +34,7 @@ use restate_types::storage::StorageEncode;
 
 use crate::appender::Appender;
 use crate::background_appender::BackgroundAppender;
+use crate::log_chain_watcher::LogChainWatcherHandle;
 use crate::log_chain_writer::LogChainCommand;
 use crate::loglet::{FindTailOptions, LogletProvider, OperationError};
 use crate::loglet_wrapper::LogletWrapper;
@@ -294,15 +295,17 @@ pub struct BifrostInner {
     pub(crate) providers: OnceLock<EnumMap<ProviderKind, Option<Arc<dyn LogletProvider>>>>,
     shutting_down: AtomicBool,
     pub(crate) read_stream_registry: crate::read_stream_registry::ActiveReadStreamRegistry,
+    log_chain_watcher: LogChainWatcherHandle,
 }
 
 impl BifrostInner {
-    pub fn new(watchdog: WatchdogSender) -> Self {
+    pub fn new(watchdog: WatchdogSender, log_chain_watcher: LogChainWatcherHandle) -> Self {
         Self {
             watchdog,
             providers: Default::default(),
             shutting_down: AtomicBool::new(false),
             read_stream_registry: Default::default(),
+            log_chain_watcher,
         }
     }
 
@@ -660,6 +663,10 @@ impl BifrostInner {
             loglet,
         ))
     }
+
+    pub fn log_chain_watcher(&self) -> &LogChainWatcherHandle {
+        &self.log_chain_watcher
+    }
 }
 
 /// Result of a lookup by lsn in the chain

crates/bifrost/src/lib.rs

@@ -13,6 +13,7 @@ mod background_appender;
 mod bifrost;
 mod bifrost_admin;
 mod error;
+mod log_chain_watcher;
 mod log_chain_writer;
 pub mod loglet;
 mod loglet_wrapper;