[WIP] Secondary indexes column family basics in partition-store

Cargo.toml

@@ -6,6 +6,7 @@ members = [
     "crates/encoding/derive",
     "crates/codederror/derive",
     "server",
+    "lite",
     "benchmarks",
     "tools/bifrost-benchpress",
     "tools/mock-service-endpoint",
@@ -50,6 +51,7 @@ restate-encoding = { path = "crates/encoding" }
 restate-errors = { path = "crates/errors" }
 restate-fs-util = { path = "crates/fs-util" }
 restate-futures-util = { path = "crates/futures-util" }
+restate-hyper-uds = { path = "crates/hyper-uds" }
 restate-ingress-http = { path = "crates/ingress-http" }
 restate-ingress-kafka = { path = "crates/ingress-kafka" }
 restate-invoker-api = { path = "crates/invoker-api" }
@@ -68,6 +70,7 @@ restate-queue = { path = "crates/queue" }
 restate-rocksdb = { path = "crates/rocksdb" }
 restate-serde-util = { path = "crates/serde-util" }
 restate-server = { path = "server" }
+restate-lite = { path = "lite" }
 restate-service-client = { path = "crates/service-client" }
 restate-service-protocol = { path = "crates/service-protocol" }
 restate-service-protocol-v4 = { path = "crates/service-protocol-v4" }
@@ -190,7 +193,7 @@ rangemap = "1.5.1"
 rayon = { version = "1.10" }
 regex = { version = "1.11" }
 regress = { version = "0.10" }
-reqwest = { version = "0.12.5", default-features = false, features = [
+reqwest = { version = "0.12", default-features = false, features = [
     "json",
     "rustls-tls",
     "stream",

benchmarks/src/lib.rs

@@ -9,13 +9,21 @@
 // by the Apache License, Version 2.0.
 
 #![allow(clippy::async_yields_async)]
-
 //! Utilities for benchmarking the Restate runtime
+
+use std::net::SocketAddr;
+use std::time::Duration;
+
 use anyhow::anyhow;
 use futures_util::{TryFutureExt, future};
 use http::Uri;
 use http::header::CONTENT_TYPE;
 use pprof::flamegraph::Options;
+use tokio::net::TcpListener;
+use tokio::runtime::Runtime;
+use tokio::sync::oneshot;
+use tracing::warn;
+
 use restate_core::{TaskCenter, TaskCenterBuilder, TaskKind, cancellation_token, task_center};
 use restate_node::Node;
 use restate_rocksdb::RocksDbManager;
@@ -26,11 +34,8 @@ use restate_types::config::{
 };
 use restate_types::config_loader::ConfigLoaderBuilder;
 use restate_types::logs::metadata::ProviderKind;
+use restate_types::net::listener::AddressBook;
 use restate_types::retries::RetryPolicy;
-use std::time::Duration;
-use tokio::runtime::Runtime;
-use tokio::sync::oneshot;
-use tracing::warn;
 
 pub fn discover_deployment(current_thread_rt: &Runtime, address: Uri) {
     let client = reqwest::Client::builder()
@@ -102,16 +107,23 @@ pub fn spawn_restate(config: Configuration) -> task_center::Handle {
         .build()
         .expect("task_center builds")
         .into_handle();
+
     let mut prometheus = Prometheus::install(&config.common);
     restate_types::config::set_current_config(config.clone());
+
+    let mut address_book = AddressBook::new(restate_types::config::node_filepath(""));
     let live_config = Configuration::live();
 
     tc.block_on(async {
         RocksDbManager::init();
         prometheus.start_upkeep_task();
 
+        if let Err(err) = address_book.bind_from_config(&live_config.pinned()).await {
+            panic!("Failed to bind address book: {err}");
+        }
+
         TaskCenter::spawn(TaskKind::SystemBoot, "restate", async move {
-            let node = Node::create(live_config, prometheus)
+            let node = Node::create(live_config, prometheus, address_book)
                 .await
                 .expect("Restate node must build");
             node.start().await
@@ -126,9 +138,12 @@ pub fn spawn_mock_service_endpoint(task_center_handle: &task_center::Handle) {
     task_center_handle.block_on(async {
         let (running_tx, running_rx) = oneshot::channel();
         TaskCenter::spawn(TaskKind::TestRunner, "mock-service-endpoint", async move {
+            let addr: SocketAddr = ([127, 0, 0, 1], 9080).into();
+            let listener = TcpListener::bind(addr).await?;
+
             cancellation_token()
                 .run_until_cancelled(mock_service_endpoint::listener::run_listener(
-                    "127.0.0.1:9080".parse().expect("valid socket addr"),
+                    listener,
                     || {
                         let _ = running_tx.send(());
                     },

cli/Cargo.toml

@@ -17,7 +17,8 @@ dist = true
 formula = "restate"
 
 [features]
-default = ["cloud", "no-trace-logging"]
+default = ["cloud", "no-trace-logging", "dev-cmd"]
+dev-cmd = ["restate-lite", "mock-service-endpoint"]
 cloud = []
 no-trace-logging = ["tracing/max_level_trace", "tracing/release_max_level_debug"]
 
@@ -30,6 +31,8 @@ restate-cloud-tunnel-client = { workspace = true }
 restate-serde-util = { workspace = true }
 restate-time-util = { workspace = true }
 restate-types = { workspace = true }
+restate-lite = { workspace = true, optional = true }
+mock-service-endpoint = { workspace = true, optional = true }
 
 anyhow = { workspace = true }
 arc-swap = { workspace = true }

cli/src/app.rs

@@ -49,6 +49,10 @@ pub struct GlobalOpts {
 
 #[derive(Run, Subcommand, Clone)]
 pub enum Command {
+    #[cfg(feature = "dev-cmd")]
+    #[clap(name = "dev", visible_alias = "up")]
+    Dev(dev::Dev),
+
     /// Prints general information about the configured environment
     #[clap(name = "whoami")]
     WhoAmI(whoami::WhoAmI),