scaffolding leader Sequencer (#1982)

* scaffolding appender

* scaffolding leader Sequencer

crates/bifrost/Cargo.toml

@@ -43,6 +43,7 @@ tracing = { workspace = true }
 
 [dev-dependencies]
 restate-core = { workspace = true, features = ["test-util"] }
+restate-log-server = { workspace = true }
 restate-metadata-store = { workspace = true }
 restate-test-util = { workspace = true }
 restate-types = { workspace = true, features = ["test-util"] }

crates/bifrost/src/loglet/util.rs

@@ -63,6 +63,23 @@ impl TailOffsetWatch {
         Ok(())
     }
 
+    pub async fn wait_for_offset_or_seal(
+        &self,
+        offset: LogletOffset,
+    ) -> Result<TailState<LogletOffset>, ShutdownError> {
+        let mut receiver = self.sender.subscribe();
+        receiver.mark_changed();
+        receiver
+            .wait_for(|tail_state| match tail_state {
+                TailState::Sealed(_) => true,
+                TailState::Open(tail) if *tail >= offset => true,
+                _ => false,
+            })
+            .await
+            .map(|m| *m)
+            .map_err(|_| ShutdownError)
+    }
+
     /// The first yielded value is the latest known tail
     pub fn to_stream(&self) -> WatchStream<TailState<LogletOffset>> {
         let mut receiver = self.sender.subscribe();

crates/bifrost/src/providers/replicated_loglet/replication/spread_selector.rs

@@ -26,7 +26,7 @@ pub enum SpreadSelectorError {
     InsufficientWriteableNodes,
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum SelectorStrategy {
     /// Selects all writeable nodes in the nodeset, this might lead to over-replication,
     /// and it's up to the appender state machine to continue replicating beyond the
@@ -38,6 +38,7 @@ pub enum SelectorStrategy {
 }
 
 /// Spread selector is thread-safe and can be used concurrently.
+#[derive(Clone)]
 pub struct SpreadSelector {
     nodeset: NodeSet,
     strategy: SelectorStrategy,
@@ -57,6 +58,14 @@ impl SpreadSelector {
         }
     }
 
+    pub fn nodeset(&self) -> &NodeSet {
+        &self.nodeset
+    }
+
+    pub fn replication_property(&self) -> &ReplicationProperty {
+        &self.replication_property
+    }
+
     /// Generates a spread or fails if it's not possible to generate a spread out of
     /// the nodeset modulo the non-writeable nodes in the nodes configuration and after excluding
     /// the set of nodes passed in `exclude_nodes`.