merge: #3482

3482: feat(rebaser): Rebaser uses layerdb activites r=adamhjk a=adamhjk

This refactors the rebaser to use a layerdb activities stream. Under the hood it uses a NATS Jetstream work queue.

* Makes subscriptions to activities multiplexed (unless they are a work queue)
* Adds integration test activities
* Adds rebase and rebase_and_wait, which waits for a finish event

Co-authored-by: Adam Jacob <adam@systeminit.com>

Author: si-bors-ng[bot]

Cargo.lock

@@ -32,7 +32,6 @@ members = [
     "lib/naxum",
     "lib/object-tree",
     "lib/pinga-server",
-    "lib/rebaser-client",
     "lib/rebaser-core",
     "lib/rebaser-server",
     "lib/sdf-server",

Cargo.toml

@@ -9,7 +9,6 @@ rust_binary(
     name = "sdf",
     deps = [
         "//lib/nats-multiplexer:nats-multiplexer",
-        "//lib/rebaser-client:rebaser-client",
         "//lib/sdf-server:sdf-server",
         "//lib/si-std:si-std",
         "//lib/telemetry-application-rs:telemetry-application",

bin/sdf/BUCK

@@ -11,7 +11,6 @@ path = "src/main.rs"
 
 [dependencies]
 nats-multiplexer = { path = "../../lib/nats-multiplexer" }
-rebaser-client = { path = "../../lib/rebaser-client" }
 sdf-server = { path = "../../lib/sdf-server" }
 si-std = { path = "../../lib/si-std" }
 telemetry-application = { path = "../../lib/telemetry-application-rs" }

bin/sdf/Cargo.toml

@@ -13,8 +13,6 @@ use sdf_server::{
 use telemetry_application::prelude::*;
 use tokio_util::{sync::CancellationToken, task::TaskTracker};
 
-use rebaser_client::Config as RebaserClientConfig;
-
 mod args;
 
 type JobProcessor = sdf_server::NatsProcessor;
@@ -111,10 +109,6 @@ async fn async_main() -> Result<()> {
 
     let module_index_url = config.module_index_url().to_string();
 
-    // TODO: accept command line arguments and or environment variables to configure the rebaser
-    // client
-    let rebaser_config = RebaserClientConfig::default();
-
     let (ws_multiplexer, ws_multiplexer_client) =
         Multiplexer::new(&nats_conn, WS_MULTIPLEXER_SUBJECT).await?;
     let (crdt_multiplexer, crdt_multiplexer_client) =
@@ -138,7 +132,6 @@ async fn async_main() -> Result<()> {
         Some(pkgs_path),
         Some(module_index_url),
         symmetric_crypto_service,
-        rebaser_config,
         layer_db,
     );
 

bin/sdf/src/main.rs

@@ -8,7 +8,6 @@ rust_library(
         "//lib/dal:dal",
         "//lib/module-index-client:module-index-client",
         "//lib/pinga-server:pinga-server",
-        "//lib/rebaser-client:rebaser-client",
         "//lib/rebaser-server:rebaser-server",
         "//lib/si-crypto:si-crypto",
         "//lib/si-data-nats:si-data-nats",

lib/dal-test/BUCK

@@ -18,7 +18,6 @@ names = { workspace = true }
 opentelemetry-otlp = { workspace = true }
 opentelemetry_sdk = { workspace = true }
 pinga-server = { path = "../../lib/pinga-server" }
-rebaser-client = { path = "../../lib/rebaser-client"}
 rebaser-server = { path = "../../lib/rebaser-server" }
 remain = { workspace = true }
 serde = { workspace = true }

lib/dal-test/Cargo.toml

@@ -16,7 +16,6 @@ use dal::{
 use derive_builder::Builder;
 use jwt_simple::prelude::RS256KeyPair;
 use lazy_static::lazy_static;
-use rebaser_client::Config as RebaserClientConfig;
 use si_crypto::{
     SymmetricCryptoService, SymmetricCryptoServiceConfig, SymmetricCryptoServiceConfigFile,
 };
@@ -109,8 +108,6 @@ pub struct Config {
     symmetric_crypto_service_config: SymmetricCryptoServiceConfig,
     // TODO(nick): determine why this is unused.
     #[allow(dead_code)]
-    #[builder(default)]
-    rebaser_config: RebaserClientConfig,
     #[builder(default = "si_layer_cache::default_pg_pool_config()")]
     layer_cache_pg_pool: PgPoolConfig,
 }
@@ -219,8 +216,6 @@ pub struct TestContext {
     layer_db_pg_pool: PgPool,
     /// The sled path for the layer db
     layer_db_sled_path: String,
-    /// The configuration for the rebaser client used in tests
-    rebaser_config: RebaserClientConfig,
 }
 
 impl TestContext {
@@ -313,7 +308,6 @@ impl TestContext {
             self.config.pkgs_path.to_owned(),
             None,
             self.symmetric_crypto_service.clone(),
-            self.rebaser_config.clone(),
             layer_db,
         )
     }
@@ -396,17 +390,13 @@ impl TestContextBuilder {
             SymmetricCryptoService::from_config(&self.config.symmetric_crypto_service_config)
                 .await?;
 
-        let mut rebaser_config = RebaserClientConfig::default();
-        rebaser_config.set_subject_prefix(universal_prefix);
-
         Ok(TestContext {
             config,
             pg_pool,
             nats_conn,
             job_processor,
             encryption_key: self.encryption_key.clone(),
             symmetric_crypto_service,
-            rebaser_config,
             layer_db_pg_pool,
             layer_db_sled_path: si_layer_cache::disk_cache::default_sled_path()?.to_string(),
         })
@@ -540,23 +530,13 @@ pub fn pinga_server(services_context: &ServicesContext) -> Result<pinga_server::
 /// Configures and builds a [`rebaser_server::Server`] suitable for running alongside DAL
 /// object-related tests.
 pub fn rebaser_server(services_context: &ServicesContext) -> Result<rebaser_server::Server> {
-    let _config: rebaser_server::Config = {
-        let mut config_file = rebaser_server::ConfigFile::default();
-        rebaser_server::detect_and_configure_development(&mut config_file)
-            .wrap_err("failed to detect and configure Rebaser ConfigFile")?;
-        config_file
-            .try_into()
-            .wrap_err("failed to build Rebaser server config")?
-    };
-
     let server = rebaser_server::Server::from_services(
         services_context.encryption_key(),
         services_context.nats_conn().clone(),
         services_context.pg_pool().clone(),
         services_context.veritech().clone(),
         services_context.job_processor(),
         services_context.symmetric_crypto_service().clone(),
-        services_context.rebaser_config().clone(),
         services_context.layer_db().clone(),
     )
     .wrap_err("failed to create Rebaser server")?;
@@ -702,7 +682,6 @@ async fn global_setup(test_context_builer: TestContextBuilder) -> Result<()> {
             .expect("no pkgs path configured"),
         test_context.config.module_index_url.clone(),
         services_ctx.symmetric_crypto_service(),
-        services_ctx.rebaser_config().clone(),
         services_ctx.layer_db().clone(),
     )
     .await
@@ -745,7 +724,6 @@ async fn migrate_local_builtins(
     pkgs_path: PathBuf,
     module_index_url: String,
     symmetric_crypto_service: &SymmetricCryptoService,
-    rebaser_config: RebaserClientConfig,
     layer_db: DalLayerDb,
 ) -> ModelResult<()> {
     let services_context = ServicesContext::new(
@@ -757,7 +735,6 @@ async fn migrate_local_builtins(
         Some(pkgs_path),
         Some(module_index_url),
         symmetric_crypto_service.clone(),
-        rebaser_config,
         layer_db.clone(),
     );
     let dal_context = services_context.into_builder(true);

lib/dal-test/src/lib.rs

@@ -11,7 +11,6 @@ rust_library(
         "//lib/council-server:council-server",
         "//lib/nats-subscriber:nats-subscriber",
         "//lib/object-tree:object-tree",
-        "//lib/rebaser-client:rebaser-client",
         "//lib/si-crypto:si-crypto",
         "//lib/si-data-nats:si-data-nats",
         "//lib/si-data-pg:si-data-pg",
@@ -83,7 +82,6 @@ rust_test(
     name = "test-integration",
     deps = [
         "//lib/dal-test:dal-test",
-        "//lib/rebaser-client:rebaser-client",
         "//lib/rebaser-core:rebaser-core",
         "//lib/rebaser-server:rebaser-server",
         "//lib/si-pkg:si-pkg",

lib/dal/BUCK

@@ -33,7 +33,6 @@ petgraph = { workspace = true }
 postcard = { version = "1.0.8", features = ["alloc"] }
 postgres-types = { workspace = true }
 rand = { workspace = true }
-rebaser-client = { path = "../../lib/rebaser-client" }
 refinery = { workspace = true }
 regex = { workspace = true }
 remain = { workspace = true }