[Bifrost] Maintain a lookup index in Log metadata for replicated loglets

crates/admin/src/cluster_controller/logs_controller.rs

@@ -739,7 +739,7 @@ impl LogsControllerInner {
                 observed_cluster_state,
                 |seal_lsn, provider_kind, loglet_params| {
                     let mut chain_builder = logs_builder
-                        .chain(log_id)
+                        .chain(*log_id)
                         .expect("Log with '{log_id}' should be present");
 
                     chain_builder.append_segment(seal_lsn, provider_kind, loglet_params)
@@ -761,8 +761,7 @@ impl LogsControllerInner {
         match event {
             Event::WriteLogsSucceeded(version) => {
                 self.on_logs_written(version);
-                // todo prevent clone by adding support for passing in Arcs
-                metadata_writer.submit(self.current_logs.deref().clone());
+                metadata_writer.submit(self.current_logs.clone());
             }
             Event::WriteLogsFailed {
                 logs,
@@ -939,7 +938,7 @@ impl LogsController {
             || metadata_store_client.get_or_insert(BIFROST_CONFIG_KEY.clone(), Logs::default),
         )
         .await?;
-        metadata_writer.update(logs).await?;
+        metadata_writer.update(Arc::new(logs)).await?;
 
         //todo(azmy): make configurable
         let retry_policy = RetryPolicy::exponential(
@@ -1127,7 +1126,7 @@ impl LogsController {
                                 Ok(result) => {
                                     let logs = result.expect("should be present");
                                     // we are only failing if we are shutting down
-                                    let _ = metadata_writer.update(logs).await;
+                                    let _ = metadata_writer.update(Arc::new(logs)).await;
                                     Event::NewLogs
                                 }
                                 Err(err) => {

crates/admin/src/cluster_controller/service.rs

@@ -391,7 +391,9 @@ impl<T: TransportConnect> Service<T> {
         )
         .await?;
 
-        self.metadata_writer.update(partition_table).await?;
+        self.metadata_writer
+            .update(Arc::new(partition_table))
+            .await?;
 
         Ok(())
     }

crates/admin/src/schema_registry/mod.rs

@@ -11,9 +11,15 @@
 pub mod error;
 mod updater;
 
-use crate::schema_registry::error::{SchemaError, SchemaRegistryError, ServiceError};
-use crate::schema_registry::updater::SchemaUpdater;
 use http::Uri;
+
+use std::borrow::Borrow;
+use std::collections::HashMap;
+use std::ops::Deref;
+use std::sync::Arc;
+use std::time::Duration;
+use tracing::subscriber::NoSubscriber;
+
 use restate_core::metadata_store::MetadataStoreClient;
 use restate_core::{metadata, MetadataWriter};
 use restate_service_protocol::discovery::{DiscoverEndpoint, DiscoveredEndpoint, ServiceDiscovery};
@@ -27,11 +33,9 @@ use restate_types::schema::subscriptions::{
     ListSubscriptionFilter, Subscription, SubscriptionResolver, SubscriptionValidator,
 };
 use restate_types::schema::Schema;
-use std::borrow::Borrow;
-use std::collections::HashMap;
-use std::ops::Deref;
-use std::time::Duration;
-use tracing::subscriber::NoSubscriber;
+
+use crate::schema_registry::error::{SchemaError, SchemaRegistryError, ServiceError};
+use crate::schema_registry::updater::SchemaUpdater;
 
 /// Whether to force the registration of an existing endpoint or not
 #[derive(Clone, PartialEq, Eq, Debug)]
@@ -167,7 +171,9 @@ impl<V> SchemaRegistry<V> {
                 .get_deployment_and_services(&new_deployment_id)
                 .expect("deployment was just added");
 
-            self.metadata_writer.update(schema_information).await?;
+            self.metadata_writer
+                .update(Arc::new(schema_information))
+                .await?;
 
             (new_deployment_id, services)
         };
@@ -198,7 +204,9 @@ impl<V> SchemaRegistry<V> {
                 },
             )
             .await?;
-        self.metadata_writer.update(schema_registry).await?;
+        self.metadata_writer
+            .update(Arc::new(schema_registry))
+            .await?;
 
         Ok(())
     }
@@ -235,7 +243,9 @@ impl<V> SchemaRegistry<V> {
             .resolve_latest_service(&service_name)
             .expect("service was just modified");
 
-        self.metadata_writer.update(schema_information).await?;
+        self.metadata_writer
+            .update(Arc::new(schema_information))
+            .await?;
 
         Ok(response)
     }
@@ -267,7 +277,9 @@ impl<V> SchemaRegistry<V> {
             )
             .await?;
 
-        self.metadata_writer.update(schema_information).await?;
+        self.metadata_writer
+            .update(Arc::new(schema_information))
+            .await?;
 
         Ok(())
     }
@@ -367,7 +379,9 @@ where
         let subscription = schema_information
             .get_subscription(subscription_id.expect("subscription was just added"))
             .expect("subscription was just added");
-        self.metadata_writer.update(schema_information).await?;
+        self.metadata_writer
+            .update(Arc::new(schema_information))
+            .await?;
 
         Ok(subscription)
     }

crates/bifrost/Cargo.toml

@@ -9,7 +9,7 @@ publish = false
 
 [features]
 default = []
-replicated-loglet = ["restate-types/replicated-loglet"]
+replicated-loglet = []
 memory-loglet = ["restate-types/memory-loglet"]
 test-util = ["memory-loglet", "dep:googletest", "dep:restate-test-util"]
 

crates/bifrost/benches/util.rs

@@ -7,6 +7,7 @@
 // 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.
+use std::sync::Arc;
 
 use tracing::warn;
 
@@ -53,7 +54,7 @@ pub async fn spawn_environment(
         .put(BIFROST_CONFIG_KEY.clone(), &logs, Precondition::None)
         .await
         .expect("to store bifrost config in metadata store");
-    metadata_writer.submit(logs);
+    metadata_writer.submit(Arc::new(logs));
     spawn_metadata_manager(&tc, metadata_manager).expect("metadata manager starts");
     tc
 }

crates/bifrost/src/bifrost.rs

@@ -804,7 +804,7 @@ mod tests {
             let old_version = bifrost.inner.metadata.logs_version();
 
             let mut builder = bifrost.inner.metadata.logs_ref().clone().into_builder();
-            let mut chain_builder = builder.chain(&LOG_ID).unwrap();
+            let mut chain_builder = builder.chain(LOG_ID).unwrap();
             assert_eq!(1, chain_builder.num_segments());
             let new_segment_params = new_single_node_loglet_params(ProviderKind::InMemory);
             // deliberately skips Lsn::from(6) to create a zombie record in segment 1. Segment 1 now has 4 records.

crates/bifrost/src/bifrost_admin.rs

@@ -9,6 +9,7 @@
 // by the Apache License, Version 2.0.
 
 use std::ops::Deref;
+use std::sync::Arc;
 
 use tracing::{info, instrument};
 
@@ -194,8 +195,7 @@ impl<'a> BifrostAdmin<'a> {
                 let logs = logs.ok_or(Error::UnknownLogId(log_id))?;
 
                 let mut builder = logs.into_builder();
-                let mut chain_builder =
-                    builder.chain(&log_id).ok_or(Error::UnknownLogId(log_id))?;
+                let mut chain_builder = builder.chain(log_id).ok_or(Error::UnknownLogId(log_id))?;
 
                 if chain_builder.tail().index() != last_segment_index {
                     // tail is not what we expected.
@@ -218,7 +218,7 @@ impl<'a> BifrostAdmin<'a> {
             .await
             .map_err(|e| e.transpose())?;
 
-        self.metadata_writer.update(logs).await?;
+        self.metadata_writer.update(Arc::new(logs)).await?;
         Ok(())
     }
 }

crates/bifrost/src/read_stream.rs

@@ -761,7 +761,7 @@ mod tests {
             // when it's implemented)
             let old_version = bifrost.inner.metadata.logs_version();
             let mut builder = bifrost.inner.metadata.logs_ref().clone().into_builder();
-            let mut chain_builder = builder.chain(&LOG_ID).unwrap();
+            let mut chain_builder = builder.chain(LOG_ID).unwrap();
             assert_eq!(1, chain_builder.num_segments());
             let new_segment_params = new_single_node_loglet_params(ProviderKind::InMemory);
             chain_builder.append_segment(
@@ -982,7 +982,7 @@ mod tests {
             // prepare a chain that starts from Lsn 10 (we expect trim from OLDEST -> 9)
             let old_version = bifrost.inner.metadata.logs_version();
             let mut builder = bifrost.inner.metadata.logs_ref().clone().into_builder();
-            let mut chain_builder = builder.chain(&LOG_ID).unwrap();
+            let mut chain_builder = builder.chain(LOG_ID).unwrap();
             assert_eq!(1, chain_builder.num_segments());
             let new_segment_params = new_single_node_loglet_params(ProviderKind::Local);
             chain_builder.append_segment(Lsn::new(10), ProviderKind::Local, new_segment_params)?;