[Sequencer] Scaffolding Leader Sequencer

crates/bifrost/Cargo.toml

@@ -28,6 +28,7 @@ futures = { workspace = true }
 metrics = { workspace = true }
 parking_lot = { workspace = true }
 pin-project = { workspace = true }
+rand = { workspace = true }
 rocksdb = { workspace = true }
 serde = { workspace = true }
 smallvec = { workspace = true }
@@ -40,6 +41,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/mod.rs

@@ -154,6 +154,20 @@ pub trait LogletReadStream: Stream<Item = Result<LogEntry<LogletOffset>, Operati
 
 pub type SendableLogletReadStream = Pin<Box<dyn LogletReadStream + Send>>;
 
+pub(crate) struct Resolver {
+    tx: oneshot::Sender<Result<LogletOffset, AppendError>>,
+}
+
+impl Resolver {
+    pub fn sealed(self) {
+        let _ = self.tx.send(Err(AppendError::Sealed));
+    }
+
+    pub fn offset(self, offset: LogletOffset) {
+        let _ = self.tx.send(Ok(offset));
+    }
+}
+
 pub struct LogletCommit {
     rx: oneshot::Receiver<Result<LogletOffset, AppendError>>,
 }
@@ -170,6 +184,11 @@ impl LogletCommit {
         let _ = tx.send(Ok(offset));
         Self { rx }
     }
+
+    pub(crate) fn later() -> (Self, Resolver) {
+        let (tx, rx) = oneshot::channel();
+        (Self { rx }, Resolver { tx })
+    }
 }
 
 impl std::future::Future for LogletCommit {

crates/bifrost/src/providers/replicated_loglet/mod.rs

@@ -10,5 +10,6 @@
 
 pub(crate) mod metric_definitions;
 mod provider;
+mod sequencer;
 
 pub use provider::Factory;