Implement simple networking layer for RaftMetadataStore

crates/admin/src/cluster_controller/service.rs

@@ -241,7 +241,9 @@ where
                     }
                 }
                 Ok(cluster_state) = cluster_state_watcher.next_cluster_state() => {
-                    scheduler.on_cluster_state_update(cluster_state).await?;
+                    if let Err(err) = scheduler.on_cluster_state_update(cluster_state).await {
+                        warn!("Could not perform scheduling operation: {err}");
+                    }
                 }
                 Some(cmd) = self.command_rx.recv() => {
                     self.on_cluster_cmd(cmd, bifrost_admin).await;

crates/metadata-store/Cargo.toml

@@ -18,20 +18,25 @@ restate-rocksdb = { workspace = true }
 restate-types = { workspace = true }
 
 anyhow = { workspace = true }
+assert2 = { workspace = true }
 async-trait = { workspace = true }
 bytes = { workspace = true }
 bytestring = { workspace = true }
 derive_builder = { workspace = true }
+derive_more = { workspace = true }
 futures = { workspace = true }
 http = { workspace = true }
 humantime = { workspace = true }
 hyper = { workspace = true }
 hyper-util = { workspace = true }
 prost = { workspace = true }
 prost-types = { workspace = true }
+protobuf = "2.28.0"
+raft = { version = "0.7.0" }
 rocksdb = { workspace = true }
 schemars = { workspace = true, optional = true }
 serde = { workspace = true }
+slog = { version = "2.7.0" }
 static_assertions = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
@@ -43,6 +48,8 @@ tonic-health = { workspace = true }
 tower = { workspace = true }
 tower-http = { workspace = true, features = ["trace"] }
 tracing = { workspace = true }
+tracing-slog = { version = "0.3.0" }
+ulid = { workspace = true, features = ["serde"] }
 
 [dev-dependencies]
 restate-core = { workspace = true, features = ["test-util"] }

crates/metadata-store/build.rs

@@ -21,5 +21,11 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         .protoc_arg("--experimental_allow_proto3_optional")
         .compile(&["./proto/metadata_store_svc.proto"], &["proto"])?;
 
+    tonic_build::configure()
+        .bytes(["."])
+        .file_descriptor_set_path(out_dir.join("raft_metadata_store_svc.bin"))
+        .protoc_arg("--experimental_allow_proto3_optional")
+        .compile(&["./proto/raft_metadata_store_svc.proto"], &["proto"])?;
+
     Ok(())
 }

crates/metadata-store/proto/raft_metadata_store_svc.proto

@@ -0,0 +1,24 @@
+// Copyright (c) 2024 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate service protocol, which is
+// released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/proto/blob/main/LICENSE
+
+syntax = "proto3";
+
+import "google/protobuf/empty.proto";
+
+package dev.restate.raft_metadata_store_svc;
+
+// Grpc service definition for the RaftMetadataStore implementation.
+service RaftMetadataStoreSvc {
+  rpc Raft(stream RaftMessage) returns (stream RaftMessage);
+}
+
+message RaftMessage {
+  bytes message = 1;
+}
+