scaffolding appender

crates/bifrost/Cargo.toml

@@ -41,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/providers/replicated_loglet/sequencer/append.rs

@@ -0,0 +1,276 @@
+use std::sync::Arc;
+
+use restate_core::{
+    network::{rpc_router::ResponseTracker, Incoming, NetworkError, Outgoing},
+    Metadata,
+};
+use restate_types::{
+    logs::{LogletOffset, Record, SequenceNumber},
+    net::log_server::{Status, Store, StoreFlags, Stored},
+    replicated_loglet::NodeSet,
+};
+use tokio::task::JoinSet;
+
+use crate::providers::replicated_loglet::replication::spread_selector::SpreadSelector;
+
+use super::{
+    node::{LogServer, LogServerManager},
+    SequencerGlobalState,
+};
+
+enum AppenderState {
+    Wave {
+        replicated: NodeSet,
+        gray_list: NodeSet,
+    },
+    Done,
+}
+
+/// Appender makes sure a batch of records will run to completion
+pub(crate) struct Appender {
+    global: Arc<SequencerGlobalState>,
+    log_server_manager: LogServerManager,
+    response_tracker: ResponseTracker<Stored>,
+    selector: SpreadSelector,
+    metadata: Metadata,
+    first_offset: LogletOffset,
+    records: Arc<[Record]>,
+}
+
+impl Appender {
+    pub fn new(
+        global: Arc<SequencerGlobalState>,
+        log_server_manager: LogServerManager,
+        response_tracker: ResponseTracker<Stored>,
+        selector: SpreadSelector,
+        metadata: Metadata,
+        first_offset: LogletOffset,
+        records: Arc<[Record]>,
+    ) -> Self {
+        Self {
+            global,
+            log_server_manager,
+            response_tracker,
+            selector,
+            metadata,
+            first_offset,
+            records,
+        }
+    }
+
+    pub async fn run(mut self) {
+        let mut state = AppenderState::Wave {
+            replicated: NodeSet::empty(),
+            gray_list: NodeSet::empty(),
+        };
+
+        loop {
+            state = match state {
+                AppenderState::Wave {
+                    replicated,
+                    gray_list,
+                } => self.wave(replicated, gray_list).await,
+                AppenderState::Done => break,
+            }
+        }
+    }
+
+    async fn wave(&mut self, replicated: NodeSet, mut gray_list: NodeSet) -> AppenderState {
+        let spread = match self.selector.select(
+            &mut rand::thread_rng(),
+            &self.metadata.nodes_config_ref(),
+            &gray_list,
+        ) {
+            Ok(spread) => spread,
+            Err(_) => {
+                //todo(azmy): retry without a gray-list !
+                return AppenderState::Wave {
+                    replicated,
+                    gray_list: NodeSet::empty(),
+                };
+            }
+        };
+
+        let mut gray = false;
+        let mut servers = Vec::with_capacity(spread.len());
+        for id in spread {
+            // at this stage, if we fail to get connection to this server it must be
+            // a first time use. We can safely assume this has to be gray listed
+            let server = match self.log_server_manager.get(id).await {
+                Ok(server) => server,
+                Err(err) => {
+                    tracing::error!("Failed to connect to {}: {}", id, err);
+                    gray = true;
+                    gray_list.insert(id);
+                    continue;
+                }
+            };
+
+            servers.push(server);
+        }
+
+        if gray {
+            // so not all nodes in the new spread can be used
+            // todo(azmy): is it possible that we still can get a write quorum on the remaining
+            //    set of nodes. The selector probably should provide an interface to validate that!
+
+            // we basically try again with a new set of nodes
+            return AppenderState::Wave {
+                replicated,
+                gray_list,
+            };
+        }
+
+        return self.send_wave(servers).await;
+    }
+
+    async fn send_wave(&mut self, spread: Vec<LogServer>) -> AppenderState {
+        let mut waiters = JoinSet::new();
+
+        for server in spread {
+            // it is possible that we have visited this server
+            // in a previous wave. So we can short circuit here
+            // and just skip
+            if server.state().local_tail() > self.first_offset {
+                continue;
+            }
+
+            let sender = Sender {
+                global: Arc::clone(&self.global),
+                server_manager: self.log_server_manager.clone(),
+                server,
+                first_offset: self.first_offset,
+                records: Arc::clone(&self.records),
+                tracker: self.response_tracker.clone(),
+            };
+
+            waiters.spawn(sender.send());
+        }
+
+        let mut gray_list = NodeSet::empty();
+        let mut replicated = NodeSet::empty();
+        // todo(azmy): join_next should timeout if nodes are taking too long to respond!
+        while let Ok((server, result)) = waiters.join_next().await.expect("task run to completion")
+        {
+            let status = match result {
+                Ok(status) => status,
+                Err(_err) => {
+                    // todo(azmy): handle errors differently
+                    gray_list.insert(*server.node());
+                    continue;
+                }
+            };
+
+            // we had a response from this node and there is still a lot we can do
+            match status {
+                Status::Ok => {
+                    replicated.insert(*server.node());
+                }
+                Status::Sealed | Status::Sealing => {
+                    server.state().seal();
+                }
+                _ => {
+                    //todo(azmy): handle other status
+                    gray_list.insert(*server.node());
+                }
+            }
+        }
+
+        if gray_list.is_empty() {
+            AppenderState::Done
+        } else {
+            AppenderState::Wave {
+                replicated,
+                gray_list,
+            }
+        }
+    }
+}
+
+struct Sender {
+    global: Arc<SequencerGlobalState>,
+    server_manager: LogServerManager,
+    server: LogServer,
+    first_offset: LogletOffset,
+    records: Arc<[Record]>,
+    tracker: ResponseTracker<Stored>,
+}
+
+impl Sender {
+    async fn send(mut self) -> (LogServer, Result<Status, NetworkError>) {
+        tokio::select! {
+            _ = self.global.wait_for_committed_tail(self.first_offset) => {
+                // we reached global tail and we can actually commit
+            },
+            Some(offset) = self.server.state().wait_for_tail(self.first_offset) => {
+                if offset > self.first_offset {
+                    return (self.server, Ok(Status::Ok));
+                }
+            }
+        };
+
+        let incoming = match self.try_send().await {
+            Ok(incoming) => incoming,
+            Err(err) => {
+                return (self.server, Err(err));
+            }
+        };
+
+        // quick actions
+        match incoming.status {
+            Status::Ok => {
+                self.server
+                    .state()
+                    .maybe_update_local_tail(incoming.local_tail);
+            }
+            _ => {}
+        }
+
+        (self.server, Ok(incoming.status))
+    }
+
+    async fn try_send(&mut self) -> Result<Incoming<Stored>, NetworkError> {
+        // if we are here so either we at global committed tail or node tail
+        // is at first_offset. In either cases, we can try to send our Store message!
+        let store = Store {
+            first_offset: self.first_offset,
+            flags: StoreFlags::empty(),
+            known_archived: LogletOffset::INVALID,
+            known_global_tail: self.global.committed_tail(),
+            loglet_id: self.server.loglet_id(),
+            payloads: Vec::from_iter(self.records.iter().cloned()),
+            sequencer: self.global.node_id,
+            timeout_at: None,
+        };
+
+        let mut msg = Outgoing::new(*self.server.node(), store);
+        let token = self
+            .tracker
+            .new_token(msg.msg_id())
+            .expect("unique message id");
+
+        loop {
+            match self.server.sender().send(msg).await {
+                Ok(_) => break,
+                Err(send) => {
+                    msg = send.message;
+
+                    match send.source {
+                        NetworkError::ConnectionClosed
+                        | NetworkError::ConnectError(_)
+                        | NetworkError::Timeout(_) => {
+                            self.server_manager.renew(&mut self.server).await?
+                        }
+                        _ => return Err(send.source.into()),
+                    }
+                }
+            }
+        }
+
+        // message has been sent! there is noway
+        // we are sure it has been received by the other peer
+        // so we wait for response. indefinitely
+        // it's up to the appender time outs to try again
+        token.recv().await.map_err(NetworkError::from)
+    }
+}

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

@@ -1,8 +1,68 @@
-use restate_core::network::NetworkError;
-use restate_types::nodes_config::NodesConfigError;
+// Copyright (c) 2024 -  Restate Software, Inc., Restate GmbH.
+// All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
 
+mod append;
 mod node;
 
+use restate_core::{network::NetworkError, ShutdownError};
+use restate_types::{
+    logs::LogletOffset, nodes_config::NodesConfigError, replicated_loglet::ReplicatedLogletId,
+    GenerationalNodeId,
+};
+use tokio::sync::watch;
+
+/// 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,
+    committed_tail: watch::Sender<LogletOffset>,
+}
+
+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 {
+        *self.committed_tail.borrow()
+    }
+
+    /// update committed tail if and only if new_tail is greater than
+    /// current global committed tail
+    pub fn maybe_update_committed_tail(&self, new_tail: LogletOffset) {
+        self.committed_tail.send_if_modified(|t| {
+            if new_tail > *t {
+                *t = new_tail;
+                true
+            } else {
+                false
+            }
+        });
+    }
+
+    /// wait for global tail to be greater than or equal new_tail
+    pub async fn wait_for_committed_tail(&self, new_tail: LogletOffset) -> Option<LogletOffset> {
+        let mut subscriber = self.committed_tail.subscribe();
+        subscriber
+            .wait_for(|f| *f >= new_tail)
+            .await
+            .map(|t| *t)
+            .ok()
+    }
+}
+
 #[derive(thiserror::Error, Debug)]
 enum Error {
     #[error("invalid node-set configuration")]

crates/types/src/time.rs

@@ -15,7 +15,17 @@ use std::time::{Duration, SystemTime};
 
 /// Milliseconds since the unix epoch
 #[derive(
-    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
+    Debug,
+    Clone,
+    Copy,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    Hash,
+    serde::Serialize,
+    serde::Deserialize,
+    derive_more::Into,
 )]
 #[serde(transparent)]
 #[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]