[Mechanical] Rename ReplicatedLogletId to LogletId

crates/admin/src/cluster_controller/logs_controller.rs

@@ -37,11 +37,11 @@ use restate_types::logs::metadata::{
     Chain, DefaultProvider, LogletConfig, LogletParams, Logs, LogsConfiguration,
     NodeSetSelectionStrategy, ProviderKind, ReplicatedLogletConfig, SegmentIndex,
 };
-use restate_types::logs::{LogId, Lsn, TailState};
+use restate_types::logs::{LogId, LogletId, Lsn, TailState};
 use restate_types::metadata_store::keys::BIFROST_CONFIG_KEY;
 use restate_types::nodes_config::NodesConfiguration;
 use restate_types::partition_table::PartitionTable;
-use restate_types::replicated_loglet::{ReplicatedLogletId, ReplicatedLogletParams};
+use restate_types::replicated_loglet::ReplicatedLogletParams;
 use restate_types::retries::{RetryIter, RetryPolicy};
 use restate_types::{logs, GenerationalNodeId, NodeId, PlainNodeId, Version, Versioned};
 
@@ -55,9 +55,6 @@ type Result<T, E = LogsControllerError> = std::result::Result<T, E>;
 
 const FALLBACK_MAX_RETRY_DELAY: Duration = Duration::from_secs(5);
 
-/// A single unified id type enables easier migration between loglet types.
-type LogletId = ReplicatedLogletId;
-
 #[derive(Debug, thiserror::Error)]
 pub enum LogsControllerError {
     #[error("failed writing to the metadata store: {0}")]
@@ -335,7 +332,7 @@ fn try_provisioning(
         #[cfg(feature = "replicated-loglet")]
         DefaultProvider::Replicated(ref config) => build_new_replicated_loglet_configuration(
             config,
-            ReplicatedLogletId::new(log_id, SegmentIndex::OLDEST),
+            LogletId::new(log_id, SegmentIndex::OLDEST),
             &Metadata::with_current(|m| m.nodes_config_ref()),
             observed_cluster_state,
             None,
@@ -350,7 +347,7 @@ fn try_provisioning(
 #[cfg(feature = "replicated-loglet")]
 pub fn build_new_replicated_loglet_configuration(
     replicated_loglet_config: &ReplicatedLogletConfig,
-    loglet_id: ReplicatedLogletId,
+    loglet_id: LogletId,
     nodes_config: &NodesConfiguration,
     observed_cluster_state: &ObservedClusterState,
     previous_params: Option<&ReplicatedLogletParams>,
@@ -1284,12 +1281,11 @@ pub mod tests {
     use restate_types::logs::metadata::{
         DefaultProvider, LogsConfiguration, NodeSetSelectionStrategy, ReplicatedLogletConfig,
     };
+    use restate_types::logs::LogletId;
     use restate_types::nodes_config::{
         LogServerConfig, NodeConfig, NodesConfiguration, Role, StorageState,
     };
-    use restate_types::replicated_loglet::{
-        NodeSet, ReplicatedLogletId, ReplicatedLogletParams, ReplicationProperty,
-    };
+    use restate_types::replicated_loglet::{NodeSet, ReplicatedLogletParams, ReplicationProperty};
     use restate_types::{GenerationalNodeId, NodeId, PlainNodeId};
 
     use crate::cluster_controller::logs_controller::{
@@ -1492,7 +1488,7 @@ pub mod tests {
             .build();
 
         let seq_n0 = ReplicatedLogletParams {
-            loglet_id: ReplicatedLogletId::from(1),
+            loglet_id: LogletId::from(1),
             sequencer: GenerationalNodeId::new(0, 1),
             replication: ReplicationProperty::new(NonZeroU8::new(2).unwrap()),
             nodeset: NodeSet::from([0, 1, 2]),
@@ -1583,7 +1579,7 @@ pub mod tests {
 
         let initial = build_new_replicated_loglet_configuration(
             replicated_loglet_config,
-            ReplicatedLogletId::from(1),
+            LogletId::from(1),
             &nodes.nodes_config,
             &nodes.observed_state,
             None,

crates/admin/src/cluster_controller/service.rs

@@ -17,6 +17,12 @@ use std::time::Duration;
 
 use anyhow::{anyhow, Context};
 use codederror::CodedError;
+use tokio::sync::{mpsc, oneshot};
+use tokio::time;
+use tokio::time::{Instant, Interval, MissedTickBehavior};
+use tonic::codec::CompressionEncoding;
+use tracing::{debug, info};
+
 use restate_metadata_store::ReadModifyWriteError;
 use restate_types::cluster_controller::SchedulingPlan;
 use restate_types::logs::metadata::{
@@ -28,12 +34,7 @@ use restate_types::metadata_store::keys::{
 use restate_types::partition_table::{
     self, PartitionTable, PartitionTableBuilder, ReplicationStrategy,
 };
-use restate_types::replicated_loglet::{ReplicatedLogletId, ReplicatedLogletParams};
-use tokio::sync::{mpsc, oneshot};
-use tokio::time;
-use tokio::time::{Instant, Interval, MissedTickBehavior};
-use tonic::codec::CompressionEncoding;
-use tracing::{debug, info};
+use restate_types::replicated_loglet::ReplicatedLogletParams;
 
 use restate_bifrost::{Bifrost, BifrostAdmin, SealedSegment};
 use restate_core::metadata_store::{retry_on_network_error, MetadataStoreClient};
@@ -51,19 +52,19 @@ use restate_types::config::{AdminOptions, Configuration};
 use restate_types::health::HealthStatus;
 use restate_types::identifiers::{PartitionId, SnapshotId};
 use restate_types::live::Live;
-use restate_types::logs::{LogId, Lsn};
+use restate_types::logs::{LogId, LogletId, Lsn};
 use restate_types::net::metadata::MetadataKind;
 use restate_types::net::partition_processor_manager::CreateSnapshotRequest;
 use restate_types::protobuf::common::AdminStatus;
 use restate_types::{GenerationalNodeId, Version, Versioned};
 
+use self::state::ClusterControllerState;
 use super::cluster_state_refresher::ClusterStateRefresher;
 use super::grpc_svc_handler::ClusterCtrlSvcHandler;
 use super::protobuf::cluster_ctrl_svc_server::ClusterCtrlSvcServer;
 use crate::cluster_controller::logs_controller::{self, NodeSetSelectorHints};
 use crate::cluster_controller::observed_cluster_state::ObservedClusterState;
 use crate::cluster_controller::scheduler::SchedulingPlanNodeSetSelectorHints;
-use state::ClusterControllerState;
 
 #[derive(Debug, thiserror::Error, CodedError)]
 pub enum Error {
@@ -761,13 +762,13 @@ impl SealAndExtendTask {
         let (loglet_id, previous_params) = match segment.config.kind {
             #[cfg(any(test, feature = "memory-loglet"))]
             ProviderKind::InMemory => {
-                let loglet_id = ReplicatedLogletId::from_str(&segment.config.params)
-                    .context("Invalid loglet id")?;
+                let loglet_id =
+                    LogletId::from_str(&segment.config.params).context("Invalid loglet id")?;
                 (loglet_id, None)
             }
             ProviderKind::Local => {
-                let loglet_id = ReplicatedLogletId::from_str(&segment.config.params)
-                    .context("Invalid loglet id")?;
+                let loglet_id =
+                    LogletId::from_str(&segment.config.params).context("Invalid loglet id")?;
                 (loglet_id, None)
             }
             #[cfg(feature = "replicated-loglet")]
@@ -835,7 +836,7 @@ mod tests {
     use test_log::test;
 
     use restate_bifrost::providers::memory_loglet;
-    use restate_bifrost::{Bifrost, BifrostService};
+    use restate_bifrost::{Bifrost, BifrostService, ErrorRecoveryStrategy};
     use restate_core::network::{
         FailingConnector, Incoming, MessageHandler, MockPeerConnection, NetworkServerBuilder,
     };
@@ -875,7 +876,7 @@ mod tests {
         let _ = builder.build().await;
         bifrost_svc.start().await?;
 
-        let mut appender = bifrost.create_appender(LOG_ID)?;
+        let mut appender = bifrost.create_appender(LOG_ID, ErrorRecoveryStrategy::default())?;
 
         TaskCenter::spawn(TaskKind::SystemService, "cluster-controller", svc.run())?;
 
@@ -972,7 +973,7 @@ mod tests {
         let (_node_2, _node2_reactor) =
             node_2.process_with_message_handler(get_node_state_handler)?;
 
-        let mut appender = bifrost.create_appender(LOG_ID)?;
+        let mut appender = bifrost.create_appender(LOG_ID, ErrorRecoveryStrategy::default())?;
         for i in 1..=20 {
             let lsn = appender.append("").await?;
             assert_eq!(Lsn::from(i), lsn);
@@ -1049,7 +1050,7 @@ mod tests {
         let (_node_2, _node2_reactor) =
             node_2.process_with_message_handler(get_node_state_handler)?;
 
-        let mut appender = bifrost.create_appender(LOG_ID)?;
+        let mut appender = bifrost.create_appender(LOG_ID, ErrorRecoveryStrategy::default())?;
         for i in 1..=20 {
             let lsn = appender.append(format!("record{i}")).await?;
             assert_eq!(Lsn::from(i), lsn);
@@ -1112,7 +1113,7 @@ mod tests {
         })
         .await?;
 
-        let mut appender = bifrost.create_appender(LOG_ID)?;
+        let mut appender = bifrost.create_appender(LOG_ID, ErrorRecoveryStrategy::default())?;
         for i in 1..=5 {
             let lsn = appender.append(format!("record{i}")).await?;
             assert_eq!(Lsn::from(i), lsn);

crates/bifrost/Cargo.toml

@@ -12,6 +12,8 @@ default = []
 replicated-loglet = []
 memory-loglet = ["restate-types/memory-loglet"]
 test-util = ["memory-loglet", "dep:googletest", "dep:restate-test-util"]
+# enables bifrost to auto seal and extend. This is a transitional feature that will be removed soon.
+auto-extend = []
 
 [dependencies]
 restate-core = { workspace = true }

crates/bifrost/benches/replicated_loglet_serde.rs

@@ -26,11 +26,10 @@ use restate_types::identifiers::{InvocationId, LeaderEpoch, PartitionProcessorRp
 use restate_types::invocation::{
     InvocationTarget, ServiceInvocation, ServiceInvocationSpanContext,
 };
-use restate_types::logs::{LogId, Record};
+use restate_types::logs::{LogId, LogletId, Record};
 use restate_types::net::codec::{serialize_message, MessageBodyExt, WireDecode};
 use restate_types::net::replicated_loglet::{Append, CommonRequestHeader};
 use restate_types::protobuf::node::Message;
-use restate_types::replicated_loglet::ReplicatedLogletId;
 use restate_types::time::MillisSinceEpoch;
 use restate_types::GenerationalNodeId;
 use restate_wal_protocol::{Command, Destination, Envelope};
@@ -124,7 +123,7 @@ fn serialize_append_message(payloads: Arc<[Record]>) -> anyhow::Result<Message>
         header: CommonRequestHeader {
             log_id: LogId::from(12u16),
             segment_index: 2.into(),
-            loglet_id: ReplicatedLogletId::new(12u16.into(), 4.into()),
+            loglet_id: LogletId::new(12u16.into(), 4.into()),
         },
         payloads,
     };

crates/bifrost/src/appender.rs

@@ -21,7 +21,7 @@ use restate_types::logs::{LogId, Lsn, Record};
 use restate_types::retries::RetryIter;
 use restate_types::storage::StorageEncode;
 
-use crate::bifrost::BifrostInner;
+use crate::bifrost::{BifrostInner, ErrorRecoveryStrategy};
 use crate::loglet::AppendError;
 use crate::loglet_wrapper::LogletWrapper;
 use crate::{Error, InputRecord, Result};
@@ -31,17 +31,25 @@ pub struct Appender {
     log_id: LogId,
     #[debug(skip)]
     pub(super) config: Live<Configuration>,
+    // todo: asoli remove
+    #[allow(unused)]
+    error_recovery_strategy: ErrorRecoveryStrategy,
     loglet_cache: Option<LogletWrapper>,
     #[debug(skip)]
     bifrost_inner: Arc<BifrostInner>,
 }
 
 impl Appender {
-    pub(crate) fn new(log_id: LogId, bifrost_inner: Arc<BifrostInner>) -> Self {
+    pub(crate) fn new(
+        log_id: LogId,
+        error_recovery_strategy: ErrorRecoveryStrategy,
+        bifrost_inner: Arc<BifrostInner>,
+    ) -> Self {
         let config = Configuration::updateable();
         Self {
             log_id,
             config,
+            error_recovery_strategy,
             loglet_cache: Default::default(),
             bifrost_inner,
         }