scaffolding nodeset and spread-selector

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,22 @@ pub trait LogletReadStream: Stream<Item = Result<LogEntry<LogletOffset>, Operati
 
 pub type SendableLogletReadStream = Pin<Box<dyn LogletReadStream + Send>>;
 
+#[allow(dead_code)]
+pub(crate) struct LogletCommitResolver {
+ tx: oneshot::Sender<Result<LogletOffset, AppendError>>,
+}
+
+#[allow(dead_code)]
+impl LogletCommitResolver {
+ 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 +186,12 @@ impl LogletCommit {
  let _ = tx.send(Ok(offset));
  Self { rx }
  }
+
+ #[allow(dead_code)]
+ pub(crate) fn deferred() -> (Self, LogletCommitResolver) {
+ let (tx, rx) = oneshot::channel();
+ (Self { rx }, LogletCommitResolver { tx })
+ }
 }
 
 impl std::future::Future for LogletCommit {

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

@@ -10,5 +10,7 @@
 
 pub(crate) mod metric_definitions;
 mod provider;
+#[allow(dead_code)]
+mod sequencer;
 
 pub use provider::Factory;

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

@@ -0,0 +1,130 @@
+use std::{
+ sync::{
+ atomic::{self, Ordering},
+ Arc,
+ },
+ time::Duration,
+};
+
+use restate_core::{network::NetworkError, ShutdownError};
+
+use restate_types::{
+ logs::LogletOffset, net::log_server::Store, replicated_loglet::ReplicatedLogletId,
+ time::MillisSinceEpoch, GenerationalNodeId, PlainNodeId,
+};
+
+mod node;
+mod worker;
+
+#[derive(Debug, Default)]
+pub struct NodeStatus {
+ // todo: this should be monotonic
+ last_response_time: atomic::AtomicU64,
+}
+
+impl NodeStatus {
+ pub(crate) fn touch(&self) {
+ // update value with latest timestamp
+ self.last_response_time
+ .store(MillisSinceEpoch::now().into(), Ordering::Relaxed);
+ }
+
+ pub fn last_response_time(&self) -> MillisSinceEpoch {
+ self.last_response_time.load(Ordering::Relaxed).into()
+ }
+
+ pub fn duration_since_last_response(&self) -> Duration {
+ // last_response_time should be monotonic
+ self.last_response_time().elapsed()
+ }
+}
+
+/// NodeClient trait abstracts the log-server node interface. One of possible implementations
+/// is a grpc client.
+#[async_trait::async_trait]
+pub trait NodeClient {
+ async fn enqueue_store(&self, msg: Store) -> Result<(), NetworkError>;
+ async fn enqueue_get_loglet_info(&self) -> Result<(), NetworkError>;
+}
+
+struct NodeInner<C> {
+ client: C,
+ state: NodeStatus,
+}
+
+/// Clonable node object, provides accessor to the underlying node client and
+/// its state
+pub struct Node<C> {
+ inner: Arc<NodeInner<C>>,
+}
+
+impl<C> Clone for Node<C> {
+ fn clone(&self) -> Self {
+ Self {
+ inner: Arc::clone(&self.inner),
+ }
+ }
+}
+
+impl<C> Node<C> {
+ fn new(client: C) -> Self {
+ Self {
+ inner: Arc::new(NodeInner {
+ client,
+ state: NodeStatus::default(),
+ }),
+ }
+ }
+
+ /// gets node client
+ pub fn client(&self) -> &C {
+ &self.inner.client
+ }
+
+ // gets node (general) status
+ pub fn status(&self) -> &NodeStatus {
+ &self.inner.state
+ }
+}
+
+/// A sharable part of the sequencer state. This is shared with node workers
+#[derive(Debug)]
+pub(crate) struct SequencerGlobalState {
+ node_id: GenerationalNodeId,
+ loglet_id: ReplicatedLogletId,
+ global_committed_tail: atomic::AtomicU32,
+}
+
+impl SequencerGlobalState {
+ pub fn node_id(&self) -> &GenerationalNodeId {
+ &self.node_id
+ }
+
+ pub fn loglet_id(&self) -> &ReplicatedLogletId {
+ &self.loglet_id
+ }
+
+ pub fn committed_tail(&self) -> LogletOffset {
+ LogletOffset::new(self.global_committed_tail.load(Ordering::Acquire))
+ }
+
+ pub(crate) fn set_committed_tail(&self, tail: LogletOffset) {
+ self.global_committed_tail
+ .fetch_max(tail.into(), Ordering::Release);
+ }
+}
+
+//todo: improve error names and description
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+ #[error("cannot satisfy spread")]
+ CannotSatisfySpread,
+ #[error("malformed batch")]
+ MalformedBatch,
+ #[error("invalid node set")]
+ InvalidNodeSet,
+ #[error("node {0} queue is full")]
+ TemporaryUnavailable(PlainNodeId),
+ #[error(transparent)]
+ Shutdown(#[from] ShutdownError),
+}