[ReplicatedLoglet] Remote append

Summary:
Implements a remote loglet append calls to leader sequencer

Author: muhamadazmy

crates/bifrost/src/loglet/error.rs

@@ -11,7 +11,7 @@
 use std::fmt::Debug;
 use std::sync::Arc;
 
-use restate_core::ShutdownError;
+use restate_core::{network::NetworkError, ShutdownError};
 use restate_types::errors::{IntoMaybeRetryable, MaybeRetryableError};
 
 #[derive(Debug, Clone, thiserror::Error)]
@@ -68,3 +68,13 @@ impl From<OperationError> for AppendError {
         }
     }
 }
+
+impl From<NetworkError> for OperationError {
+    fn from(value: NetworkError) -> Self {
+        match value {
+            NetworkError::Shutdown(err) => OperationError::Shutdown(err),
+            // todo(azmy): are all network errors retryable?
+            _ => OperationError::retryable(value),
+        }
+    }
+}

crates/bifrost/src/loglet/mod.rs

@@ -168,6 +168,10 @@ impl LogletCommitResolver {
     pub fn offset(self, offset: LogletOffset) {
         let _ = self.tx.send(Ok(offset));
     }
+
+    pub fn error(self, err: AppendError) {
+        let _ = self.tx.send(Err(err));
+    }
 }
 
 pub struct LogletCommit {

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

@@ -110,6 +110,22 @@ impl<T: TransportConnect> ReplicatedLoglet<T> {
             log_server_manager,
         })
     }
+
+    pub fn networking(&self) -> &Networking<T> {
+        &self.networking
+    }
+
+    pub fn params(&self) -> &ReplicatedLogletParams {
+        &self.my_params
+    }
+
+    pub fn log_id(&self) -> LogId {
+        self.log_id
+    }
+
+    pub fn segment_index(&self) -> SegmentIndex {
+        self.segment_index
+    }
 }
 
 #[derive(derive_more::Debug, derive_more::IsVariant)]
@@ -143,8 +159,8 @@ impl<T: TransportConnect> Loglet for ReplicatedLoglet<T> {
     async fn enqueue_batch(&self, payloads: Arc<[Record]>) -> Result<LogletCommit, OperationError> {
         match self.sequencer {
             SequencerAccess::Local { ref handle } => handle.enqueue_batch(payloads).await,
-            SequencerAccess::Remote { .. } => {
-                todo!("Access to remote sequencers is not implemented yet")
+            SequencerAccess::Remote { ref sequencers_rpc } => {
+                sequencers_rpc.append(self, payloads).await
             }
         }
     }