diff --git a/docs/planning/ROADMAP.md b/docs/planning/ROADMAP.md index 39716a6..3be3f3d 100644 --- a/docs/planning/ROADMAP.md +++ b/docs/planning/ROADMAP.md @@ -34,7 +34,7 @@ A single embeddable database can replace the 6-system content ranking stack by t | M5 | Hybrid Search | Text + semantic + signal-ranked search in one query | UC-02, UC-10, UC-11 | | M6 | Full Surface Coverage | Every use case, every sort mode, every filter, every feedback loop | UC-01 through UC-14 complete | | M7 | Production Hardening | Crash safety, graceful degradation, operational readiness | All UCs at production quality | -| M8 | Distributed Fabric | Multi-region, multi-tenant replication keeps agent-memory semantics intact | Hosted tidalDB, cloud/edge deployments, shared agent substrate (in-process primitives COMPLETE; multi-node NOT STARTED) | +| M8 | Distributed Fabric | Multi-region, multi-tenant replication keeps agent-memory semantics intact | Hosted tidalDB, cloud/edge deployments, shared agent substrate (in-process primitives COMPLETE; multi-node PARTIAL — transport + HTTP surface done, GrpcTransport wiring + tier-3 tests pending) | | M9 | Community Sync & Revocation | Local embeddable profiles can opt into community personalization and safely leave/purge contributions | Community personalization, federated taste graphs, shared feeds | | M10 | Governance & Agent Rights | Community rules and agent-scoped permissions control what signals influence ranking | User-owned AI personalization at scale, policy-compliant agents | @@ -120,9 +120,9 @@ The roadmap now has two tracks: | **m8p5: Control Plane + Multi-Tenancy + Routing** | COMPLETE | 1194 lib + 5 m8p5_multitenancy integration tests; TenantId/TenantConfig, TenantRateLimiter (AtomicU64 CAS token-bucket), TenantRouter (Jump Consistent Hash), ControlPlane (shard heartbeat + health), TenantMigration state machine, RollingUpgradeCoordinator | | **m8p6: End-to-End UAT (In-Process)** | COMPLETE | 1,206 lib + 8 m8_uat tests (0.11s); SimulatedCluster (signal-replay harness, 3 regions), NetworkPartition/ShardCrash RAII fault injection, 5 UAT steps + 3 perf assertions; p99 replication < 2s, failover < 10s, CRDT reconciliation < 100ms | | **m8p7: Network Transport (gRPC)** | COMPLETE | `tidal-net` crate: `GrpcTransport` implementing `Transport` trait via tonic 0.12; `GrpcTransportFactory`; per-peer circuit breaker (Closed/Open/HalfOpen); mutual TLS via rustls; proto `WalShipping` service (ShipSegment, StreamSegments, Heartbeat); 16 tests (contract, mTLS, reconnection); benchmark harness | -| **m8p8: Multi-Node tidal-server** | COMPLETE | `cluster` subcommand with topology YAML; `ClusterState` wrapping `SimulatedCluster`; cluster HTTP routes (`/cluster/status`, `/cluster/promote`, `/cluster/partition`, `/cluster/heal`); region-aware reads (`?region=`); leader-routed writes; `docker/cluster/Dockerfile` rebuilt (ENTRYPOINT+CMD) | +| **m8p8: Multi-Node tidal-server** | PARTIAL | `cluster` subcommand with topology YAML; `ClusterState` wrapping `SimulatedCluster` (in-process crossbeam channels — GrpcTransport not yet wired); cluster HTTP routes; region-aware reads; leader-routed writes; `docker/cluster/Dockerfile` rebuilt (ENTRYPOINT+CMD) | | **m8p9: Cross-Node Query Routing** | COMPLETE | `scatter_gather` module: entity-sharded writes via `hash(entity_id) % num_shards`; scatter-gather RETRIEVE/SEARCH across all shards; K-way merge by score; deadline propagation (50ms budget - 5ms overhead); partial failure with `degraded: true` + `unavailable_shards`; sharded HTTP routes (`/sharded/feed`, `/sharded/search`, `/sharded/items`, `/sharded/signals`) | -| **m8p10: Multi-Node UAT** | COMPLETE | 6 multi-node UAT tests over real gRPC (tidal-net/tests/multi_node_uat.rs): cross-region replication, idempotent replay, mixed signal types, 3-node convergence, runbook seed-and-converge; perf: 100 signals replicated in ~230ms over localhost gRPC (< 2s target); `apply_payload` made public for external use | +| **m8p10: Multi-Node UAT** | PARTIAL | Tier-2 (same-process, real gRPC): 8 tests in tidal-net/tests/multi_node_uat.rs — replication, idempotency, mixed signals, 3-node convergence, partition/heal, perf (100 signals in ~230ms). Tier-3 (multi-process harness with OS process isolation, iptables partition injection): NOT STARTED | **M0 Embeddable Runtime: COMPLETE** — m0p1 (skeleton), m0p2 (tooling/diagnostics), m0p3 (samples/docs). Zero-config in-process runtime with WAL, fjall backend, and tidalctl CLI operational. @@ -140,7 +140,7 @@ The roadmap now has two tracks: **M7 Production Hardening: COMPLETE** — m7p1–m7p4 + Enterprise Readiness all done. Crash recovery (BLAKE3 integrity, WAL compaction), 4-stage graceful degradation, per-agent rate limiting, session TTL sweeper, scale to 1M items, Prometheus metrics, tidalctl diagnostics, RLHF export. -**M8 Distributed Fabric: COMPLETE** — All 10 phases (m8p1–m8p10) delivered. In-process primitives (m8p1–p6): shard routing, WAL shipping, CRDT reconciliation, session continuity, multi-tenancy, control plane. Multi-node (m8p7–p10): `tidal-net` crate with `GrpcTransport` (tonic, mTLS, circuit breaker); `tidal-server cluster` subcommand with topology YAML, cluster HTTP routes, region-aware reads; scatter-gather query routing across entity-sharded nodes with deadline propagation and partial failure handling; 6 multi-node UAT tests proving gRPC replication pipeline (100 signals in ~230ms over localhost). 1,209 lib + 22 tidal-net + 10 tidal-server tests passing. +**M8 Distributed Fabric: NEAR-COMPLETE** — In-process primitives (m8p1–p6) COMPLETE: shard routing, WAL shipping, CRDT reconciliation, session continuity, multi-tenancy, control plane. Multi-node (m8p7–p10) PARTIAL: `tidal-net` crate with `GrpcTransport` (tonic, mTLS, circuit breaker) COMPLETE; `tidal-server cluster` subcommand with HTTP routes, region-aware reads COMPLETE but uses in-process SimulatedCluster transport (GrpcTransport not yet wired — m8p8 gap); scatter-gather query routing COMPLETE; tier-2 gRPC UAT tests (8 tests, same-process) COMPLETE; tier-3 multi-process test harness NOT STARTED (m8p10 gap). 1,209 lib + 22 tidal-net + 23 tidal-server tests passing. **Forage: COMPLETE** — All 5 phases done (P0: demo loop, P1: real signal surface, P2: semantic embeddings, P3: adaptive MAB, P4: bridge/surprise moment). Chrome extension + forage-server + forage-engine + forage-embedder sidecar all operational. @@ -2866,6 +2866,14 @@ These phases take the proven in-process primitives and deliver actual multi-node **Complexity:** XL **Task Files:** `docs/planning/milestone-8/phase-10/` (to be created) +### M8 Known Gaps (Verified 2026-04-11) + +| Gap | Phase | Severity | Description | +|-----|-------|----------|-------------| +| G1 | m8p8 | Medium | `ClusterState` wraps `SimulatedCluster` using crossbeam channels; spec requires `GrpcTransport` from `tidal-net` for real network replication between cluster nodes | +| G2 | m8p10 | Medium | Tier-3 multi-process test harness not built; tier-2 (same-process gRPC) covers transport correctness but not process crash, network partition injection, or rolling upgrade | +| G3 | m8p9 | Low | `entity_shard()` uses Knuth multiplicative hash while `TenantRouter` uses Jump Consistent Hash for the same entity→shard routing problem; must unify before entity-sharded production | + ### Done When (M8 Full) A developer can: diff --git a/tidal-server/Cargo.toml b/tidal-server/Cargo.toml index 2e2a68f..c29e5e5 100644 --- a/tidal-server/Cargo.toml +++ b/tidal-server/Cargo.toml @@ -24,5 +24,11 @@ tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } tidaldb = { path = "../tidal", features = ["test-utils"] } +[features] +cluster-e2e = [] + [dev-dependencies] tempfile = "3" +reqwest = { version = "0.12", features = ["json", "blocking"] } +serde_json = "1" +libc = "0.2" diff --git a/tidal-server/src/cluster.rs b/tidal-server/src/cluster.rs index 9c685b0..b0133fa 100644 --- a/tidal-server/src/cluster.rs +++ b/tidal-server/src/cluster.rs @@ -88,6 +88,7 @@ impl ClusterState { leader_region: leader_id, schema, profiles, + transports: None, // TODO(m8p8): wire GrpcTransport when topology has grpc_addr }; let cluster = SimulatedCluster::build(config); diff --git a/tidal-server/src/scatter_gather.rs b/tidal-server/src/scatter_gather.rs index 795b7c7..0f67877 100644 --- a/tidal-server/src/scatter_gather.rs +++ b/tidal-server/src/scatter_gather.rs @@ -372,6 +372,7 @@ mod tests { leader_region: RegionId(0), schema: test_schema(), profiles: Vec::new(), + transports: None, }; let cluster = SimulatedCluster::build(config); let names: HashMap = regions diff --git a/tidal-server/tests/cluster_e2e.rs b/tidal-server/tests/cluster_e2e.rs new file mode 100644 index 0000000..5812e43 --- /dev/null +++ b/tidal-server/tests/cluster_e2e.rs @@ -0,0 +1,390 @@ +//! Tier-3 multi-process cluster E2E tests. +//! +//! Spawns real `tidal-server cluster` OS processes, connects via HTTP, +//! and verifies distributed behavior across process boundaries. +//! +//! Gated behind `--features cluster-e2e` (disabled by default; these tests +//! are slow and require a built binary). +//! +//! ```bash +//! cargo test -p tidal-server --features cluster-e2e --test cluster_e2e -- --nocapture +//! ``` +#![cfg(feature = "cluster-e2e")] + +use std::io::Write; +use std::net::{SocketAddr, TcpListener}; +use std::path::PathBuf; +use std::process::{Child, Command}; +use std::time::{Duration, Instant}; + +/// A running tidal-server cluster node. +struct NodeHandle { + #[allow(dead_code)] + name: String, + http_addr: SocketAddr, + process: Child, +} + +impl Drop for NodeHandle { + fn drop(&mut self) { + // Best-effort graceful shutdown. + #[cfg(unix)] + { + unsafe { + libc::kill(self.process.id() as i32, libc::SIGTERM); + } + // Wait up to 2s for graceful exit. + let deadline = Instant::now() + Duration::from_secs(2); + loop { + match self.process.try_wait() { + Ok(Some(_)) => break, + _ if Instant::now() > deadline => { + let _ = self.process.kill(); + break; + } + _ => std::thread::sleep(Duration::from_millis(50)), + } + } + } + #[cfg(not(unix))] + { + let _ = self.process.kill(); + } + let _ = self.process.wait(); + } +} + +/// Multi-process cluster test harness. +struct ClusterHarness { + nodes: Vec, + _config_dir: tempfile::TempDir, +} + +impl ClusterHarness { + /// Spawn a multi-region cluster with `n` nodes. + /// + /// Allocates dynamic ports, generates topology + schema YAML, spawns + /// `tidal-server cluster` processes, and waits for all to become healthy. + fn start(n: usize) -> Self { + assert!(n >= 2, "need at least 2 nodes for a cluster"); + + let region_names: Vec = (0..n).map(|i| format!("region-{i}")).collect(); + let http_addrs: Vec = (0..n).map(|_| free_addr()).collect(); + + // Write config files. + let config_dir = tempfile::tempdir().expect("create temp dir"); + + let topology = Self::write_topology(&config_dir.path().to_path_buf(), ®ion_names); + let schema = Self::write_schema(&config_dir.path().to_path_buf()); + + // Find the tidal-server binary. + let bin = tidal_server_bin(); + + // Spawn processes. + let mut nodes = Vec::with_capacity(n); + for i in 0..n { + let child = Command::new(&bin) + .arg("cluster") + .arg("--listen") + .arg(http_addrs[i].to_string()) + .arg("--schema") + .arg(&schema) + .arg("--topology") + .arg(&topology) + .env("TIDAL_SERVER_LOG", "warn") + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .unwrap_or_else(|e| panic!("failed to spawn tidal-server: {e}")); + + nodes.push(NodeHandle { + name: region_names[i].clone(), + http_addr: http_addrs[i], + process: child, + }); + } + + let harness = Self { + nodes, + _config_dir: config_dir, + }; + + // Wait for all nodes to become healthy. + harness.wait_all_healthy(Duration::from_secs(15)); + harness + } + + fn write_topology(dir: &PathBuf, regions: &[String]) -> PathBuf { + let path = dir.join("topology.yaml"); + let mut f = std::fs::File::create(&path).expect("create topology.yaml"); + writeln!(f, "regions:").unwrap(); + for name in regions { + writeln!(f, " - name: {name}").unwrap(); + } + writeln!(f, "leader: {}", regions[0]).unwrap(); + path + } + + fn write_schema(dir: &PathBuf) -> PathBuf { + let path = dir.join("schema.yaml"); + std::fs::write( + &path, + r#"signals: + - name: view + entity: item + decay: + exponential: + half_life_seconds: 604800 + windows: [one_hour] + velocity: false + - name: like + entity: item + decay: + exponential: + half_life_seconds: 86400 + windows: [one_hour] + velocity: false +text_fields: + - name: title + kind: text +embedding_slots: + - name: content_vector + entity: item + dimensions: 4 +"#, + ) + .expect("write schema.yaml"); + path + } + + fn wait_all_healthy(&self, timeout: Duration) { + let deadline = Instant::now() + timeout; + for node in &self.nodes { + wait_for_health(&node.http_addr, deadline); + } + } + + fn http_get(&self, node_idx: usize, path: &str) -> reqwest::blocking::Response { + let url = format!("http://{}{path}", self.nodes[node_idx].http_addr); + reqwest::blocking::get(&url).unwrap_or_else(|e| panic!("GET {url} failed: {e}")) + } + + fn http_post_json( + &self, + node_idx: usize, + path: &str, + body: &serde_json::Value, + ) -> reqwest::blocking::Response { + let url = format!("http://{}{path}", self.nodes[node_idx].http_addr); + let client = reqwest::blocking::Client::new(); + client + .post(&url) + .json(body) + .send() + .unwrap_or_else(|e| panic!("POST {url} failed: {e}")) + } + + /// Poll `/cluster/status` until all follower regions show lag_events == 0, or timeout. + /// + /// The leader is excluded because it writes directly (not via replication) + /// and its `applied_events` counter tracks replication from other shards. + fn wait_converged(&self, timeout: Duration) { + let deadline = Instant::now() + timeout; + loop { + if Instant::now() > deadline { + panic!("convergence timeout after {timeout:?}; cluster did not converge"); + } + let resp = self.http_get(0, "/cluster/status"); + if resp.status().is_success() { + let body: serde_json::Value = resp.json().unwrap(); + let leader = body["leader"].as_str().unwrap_or(""); + if let Some(regions) = body["regions"].as_array() { + let followers_converged = regions + .iter() + .filter(|r| r["name"].as_str().unwrap_or("") != leader) + .all(|r| r["lag_events"].as_u64().unwrap_or(u64::MAX) == 0); + if followers_converged { + return; + } + } + } + std::thread::sleep(Duration::from_millis(200)); + } + } +} + +/// Allocate a free port by binding to :0 and recording the assigned port. +fn free_addr() -> SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.local_addr().unwrap() +} + +/// Find the tidal-server binary. Build it first if needed. +fn tidal_server_bin() -> PathBuf { + // Try the standard cargo target directory. + let manifest_dir = env!("CARGO_MANIFEST_DIR"); + let workspace_root = std::path::Path::new(manifest_dir) + .parent() + .expect("workspace root"); + + // Build the binary to make sure it's up to date. + let status = Command::new("cargo") + .arg("build") + .arg("-p") + .arg("tidal-server") + .current_dir(workspace_root) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status() + .expect("cargo build"); + assert!(status.success(), "cargo build -p tidal-server failed"); + + let bin = workspace_root.join("target/debug/tidal-server"); + assert!(bin.exists(), "tidal-server binary not found at {bin:?}"); + bin +} + +/// Poll a node's `/health/startup` until it returns 200. +fn wait_for_health(addr: &SocketAddr, deadline: Instant) { + let url = format!("http://{addr}/health/startup"); + loop { + if Instant::now() > deadline { + panic!("health check timeout for {addr}"); + } + match reqwest::blocking::get(&url) { + Ok(resp) if resp.status().is_success() => return, + _ => std::thread::sleep(Duration::from_millis(100)), + } + } +} + +// ── E2E Tests ───────────────────────────────────────────────────────────── + +/// Smoke test: start 3-node cluster, write data, verify convergence. +#[test] +fn cluster_smoke_test() { + let harness = ClusterHarness::start(3); + + // 1. Health check. + let resp = harness.http_get(0, "/health"); + assert!(resp.status().is_success()); + let body: serde_json::Value = resp.json().unwrap(); + assert_eq!(body["mode"], "cluster"); + + // 2. Cluster status — all regions healthy, leader is region-0. + let resp = harness.http_get(0, "/cluster/status"); + assert!(resp.status().is_success()); + let status: serde_json::Value = resp.json().unwrap(); + assert_eq!(status["leader"], "region-0"); + assert_eq!(status["regions"].as_array().unwrap().len(), 3); + + // 3. Write items. + for i in 1..=5u64 { + let resp = harness.http_post_json( + 0, + "/items", + &serde_json::json!({ + "entity_id": i, + "metadata": { "title": format!("item {i}") } + }), + ); + assert!( + resp.status().is_success(), + "POST /items failed: {}", + resp.status() + ); + } + + // 4. Write signals. + for i in 1..=5u64 { + let resp = harness.http_post_json( + 0, + "/signals", + &serde_json::json!({ + "entity_id": i, + "signal": "view", + "weight": 1.0 + }), + ); + assert!( + resp.status().is_success(), + "POST /signals failed: {}", + resp.status() + ); + } + + // 5. Wait for convergence. + harness.wait_converged(Duration::from_secs(5)); + + // 6. Verify follower regions show zero lag. + let resp = harness.http_get(0, "/cluster/status"); + let status: serde_json::Value = resp.json().unwrap(); + let leader = status["leader"].as_str().unwrap(); + for region in status["regions"].as_array().unwrap() { + if region["name"].as_str().unwrap() == leader { + continue; // Leader writes directly, not via replication. + } + assert_eq!( + region["lag_events"].as_u64().unwrap(), + 0, + "follower {} has lag", + region["name"] + ); + } + + // 7. Retrieve from a follower region. + let resp = harness.http_get(0, "/feed?profile=trending&limit=5®ion=region-1"); + assert!(resp.status().is_success()); + let feed: serde_json::Value = resp.json().unwrap(); + assert!( + !feed["items"].as_array().unwrap().is_empty(), + "follower region should have items" + ); +} + +/// Promote a follower and verify writes route to new leader. +#[test] +fn cluster_promote_leader() { + let harness = ClusterHarness::start(3); + + // Write some initial data. + harness.http_post_json( + 0, + "/items", + &serde_json::json!({ "entity_id": 1, "metadata": { "title": "before promote" } }), + ); + harness.http_post_json( + 0, + "/signals", + &serde_json::json!({ "entity_id": 1, "signal": "view", "weight": 1.0 }), + ); + harness.wait_converged(Duration::from_secs(5)); + + // Promote region-1 to leader. + let resp = harness.http_post_json( + 0, + "/cluster/promote", + &serde_json::json!({ "region": "region-1" }), + ); + assert!(resp.status().is_success()); + + // Verify status shows new leader. + let resp = harness.http_get(0, "/cluster/status"); + let status: serde_json::Value = resp.json().unwrap(); + assert_eq!(status["leader"], "region-1"); + + // Write more signals — should route to new leader. + let resp = harness.http_post_json( + 0, + "/signals", + &serde_json::json!({ "entity_id": 1, "signal": "like", "weight": 2.0 }), + ); + assert!( + resp.status().is_success(), + "POST /signals after promote failed: {}", + resp.status() + ); + // Note: convergence after leader promotion is tracked per-source-shard in + // ReplicationState, but /cluster/status reports lag against ShardId(0) only. + // Full multi-shard lag tracking is a known gap (see ROADMAP.md G1). +} diff --git a/tidal/src/testing/cluster.rs b/tidal/src/testing/cluster.rs index dcd37b2..9ab9a87 100644 --- a/tidal/src/testing/cluster.rs +++ b/tidal/src/testing/cluster.rs @@ -1,11 +1,14 @@ //! Simulated multi-region cluster for M8 UAT testing. //! //! Creates a set of ephemeral [`TidalDb`] instances wired with the real M8 -//! distributed fabric: in-process transports, `spawn_receiver`, and -//! `ReplicationState`. Signal replication now traverses the full path: +//! distributed fabric. The transport layer is pluggable: by default it uses +//! in-process crossbeam channels ([`ChannelTransport`]), but callers can +//! provide external transports (e.g., [`GrpcTransport`] from `tidal-net`). +//! +//! Signal replication traverses the full path: //! //! ```text -//! write_signal → encode_batch → channel send +//! write_signal → encode_batch → transport.send_segment() //! ↓ //! spawn_receiver thread //! ↓ @@ -22,7 +25,7 @@ //! * Every non-initial-leader region starts a `spawn_receiver` thread (via //! `db.start_replication(transport)`) that processes incoming WAL batches. //! * `write_signal` encodes the event as a one-event WAL batch and ships it -//! immediately to all non-partitioned followers. +//! immediately to all non-partitioned followers via their transport. //! * A `batch_log` records every shipped batch so `await_convergence` can //! re-deliver missed batches after a partition is healed. //! * `await_convergence` ships any pending batches, then polls @@ -39,13 +42,13 @@ use crate::db::config::{NodeConfig, NodeRole}; use crate::query::retrieve::Retrieve; use crate::query::search::Search; use crate::replication::shard::{RegionId, ShardId}; -use crate::replication::transport::WalSegmentPayload; +use crate::replication::transport::{Transport, WalSegmentPayload}; use crate::replication::{WalSegmentId, spawn_receiver}; use crate::schema::{EntityId, Schema, Timestamp}; use crate::signals::{NoopWalWriter, SignalLedger}; use crate::wal::format::batch::{EventRecord, encode_batch}; -use super::cluster_transport::{BatchEntry, ReceiveOnlyTransport, redeliver_missed}; +use super::cluster_transport::{BatchEntry, ChannelTransport, redeliver_missed}; // ── Public API ──────────────────────────────────────────────────────────── @@ -67,6 +70,12 @@ pub struct ClusterConfig { pub schema: Schema, /// Ranking profiles shared by all nodes (optional, defaults to builtins only). pub profiles: Vec, + /// External transports keyed by follower region ID. + /// + /// When `None` (the default), built-in crossbeam `ChannelTransport` is used + /// for each follower. When `Some`, the provided transports are used instead, + /// enabling real network transport (e.g., `GrpcTransport` from `tidal-net`). + pub transports: Option>>, } /// A simulated multi-region tidalDB cluster using the real M8 distributed @@ -75,16 +84,18 @@ pub struct ClusterConfig { /// Signal replication traverses the real WAL-batch encode → transport → /// `apply_payload` → `ReplicationState::advance` pipeline instead of calling /// `db.signal()` directly on followers. +/// +/// The transport layer is pluggable via [`ClusterConfig::transports`]. pub struct SimulatedCluster { /// Current leader region ID. Mutable via `promote_leader`. leader_region: Mutex, /// All nodes indexed by region. nodes: HashMap, - /// Per-follower channel senders (region → sender for WAL payloads). + /// Per-follower transports (region → transport for WAL segment delivery). /// /// Only regions that have an active `spawn_receiver` thread appear here. /// When dropped, the corresponding receiver thread exits cleanly. - follower_senders: HashMap>, + follower_transports: HashMap>, /// All batches ever shipped, for partition-recovery re-delivery. batch_log: Mutex>, /// Per-leader signal count used as the WAL seqno for that leader's batches. @@ -106,6 +117,10 @@ impl SimulatedCluster { /// All nodes are created immediately in ephemeral mode. Non-leader regions /// have a `spawn_receiver` thread started automatically. /// + /// When `config.transports` is `None`, built-in crossbeam channels are used + /// (fast, in-process). When `Some`, the provided transports are used for + /// segment delivery, enabling real network transport like `GrpcTransport`. + /// /// # Panics /// /// Panics if any `TidalDb` fails to open or if `start_replication` fails. @@ -128,8 +143,7 @@ impl SimulatedCluster { let all_shards: Vec = config.regions.iter().map(|r| ShardId(r.0)).collect(); let mut nodes: HashMap = HashMap::new(); - let mut follower_senders: HashMap> = - HashMap::new(); + let mut follower_transports: HashMap> = HashMap::new(); let mut leader_seqnos: HashMap = HashMap::new(); for ®ion in &config.regions { @@ -155,22 +169,32 @@ impl SimulatedCluster { // Wire a receiver for every non-leader region. if region != config.leader_region { - let (tx, rx) = crossbeam::channel::bounded::(1024); - let transport = Arc::new(ReceiveOnlyTransport { - local_shard: shard_id, - rx, - }); - // spawn_receiver directly: we already have Arc - // which implements Transport. Use the replication state from the db. + let transport: Arc = if let Some(ref ext) = config.transports { + // Use externally-provided transport (e.g., GrpcTransport). + ext.get(®ion) + .unwrap_or_else(|| { + panic!("no external transport for follower region {region}") + }) + .clone() + } else { + // Default: built-in crossbeam channel transport. + let (tx, rx) = crossbeam::channel::bounded::(1024); + Arc::new(ChannelTransport { + local_shard: shard_id, + tx, + rx, + }) + }; + let ledger = db .ledger() .expect("ephemeral db with schema must have ledger") .clone(); let rep_state = db.replication_state().clone(); - let _handle = spawn_receiver(transport, ledger, rep_state); + let _handle = spawn_receiver(transport.clone(), ledger, rep_state); // Note: the JoinHandle is intentionally not stored — the receiver thread - // will exit cleanly when `tx` (and all senders to `rx`) are dropped. - follower_senders.insert(region, tx); + // will exit cleanly when the transport is dropped. + follower_transports.insert(region, transport); } nodes.insert( @@ -186,7 +210,7 @@ impl SimulatedCluster { Self { leader_region: Mutex::new(config.leader_region), nodes, - follower_senders, + follower_transports, batch_log: Mutex::new(Vec::new()), leader_seqnos: Mutex::new(leader_seqnos), total_signals: AtomicU64::new(0), @@ -233,7 +257,7 @@ impl SimulatedCluster { /// Write a signal to the cluster leader and ship it to active followers. /// /// The signal is immediately applied to the leader's `TidalDb`. A one-event - /// WAL batch is encoded and shipped via the channel transport to all + /// WAL batch is encoded and shipped via each follower's transport to all /// non-partitioned followers with active receivers. The batch is also /// recorded in the `batch_log` for partition-recovery re-delivery. /// @@ -284,14 +308,14 @@ impl SimulatedCluster { let bytes = encode_batch(&events, seqno, ts.as_nanos()).expect("WAL batch encoding must not fail"); - // Ship immediately to non-partitioned followers that have active receivers. + // Ship immediately to non-partitioned followers via their transport. let partitioned = self .partitioned_regions .read() .unwrap_or_else(std::sync::PoisonError::into_inner) .clone(); - for (®ion, tx) in &self.follower_senders { + for (®ion, transport) in &self.follower_transports { if region == leader_region || partitioned.contains(®ion) { continue; } @@ -301,7 +325,7 @@ impl SimulatedCluster { event_count: 1, }; // Ignore send errors: the receiver may have exited (e.g. after a crash). - let _ = tx.send(payload); + let _ = transport.send_segment(ShardId(region.0), payload); } // Record in batch_log for partition-recovery re-delivery. @@ -383,11 +407,11 @@ impl SimulatedCluster { .batch_log .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - for (&rgn, tx) in &self.follower_senders { + for (&rgn, transport) in &self.follower_transports { if rgn == leader_region || partitioned.contains(&rgn) { continue; } - redeliver_missed(tx, &self.nodes[&rgn].db, &log); + redeliver_missed(transport.as_ref(), &self.nodes[&rgn].db, &log); } } @@ -398,7 +422,7 @@ impl SimulatedCluster { continue; } // Only check regions that have an active receiver thread. - if !self.follower_senders.contains_key(®ion) { + if !self.follower_transports.contains_key(®ion) { continue; } let applied = node @@ -510,12 +534,12 @@ impl SimulatedCluster { .unwrap_or_else(std::sync::PoisonError::into_inner); partitions.remove(®ion); } - if let Some(tx) = self.follower_senders.get(®ion) { + if let Some(transport) = self.follower_transports.get(®ion) { let log = self .batch_log .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - redeliver_missed(tx, &self.nodes[®ion].db, &log); + redeliver_missed(transport.as_ref(), &self.nodes[®ion].db, &log); } } @@ -541,9 +565,6 @@ impl SimulatedCluster { /// leader promotion, since each signal produces exactly one WAL batch. #[must_use] pub fn applied_count(&self, region: RegionId) -> u64 { - // ShardId(0) is always the initial leader's shard (RegionId(0)). - // Tests that check `applied_count` do not involve leader promotion, - // so this is always the correct source shard. self.nodes.get(®ion).map_or(0, |n| { n.db.replication_state() .applied_seqno(ShardId(0)) diff --git a/tidal/src/testing/cluster_transport.rs b/tidal/src/testing/cluster_transport.rs index 68fcf07..f0c5827 100644 --- a/tidal/src/testing/cluster_transport.rs +++ b/tidal/src/testing/cluster_transport.rs @@ -5,20 +5,22 @@ use crate::replication::WalSegmentId; use crate::replication::shard::{RegionId, ShardId}; use crate::replication::transport::{Transport, TransportError, WalSegmentPayload}; -// ── Internal: receive-only transport ───────────────────────────────────── +// ── ChannelTransport: bidirectional crossbeam transport ───────────────── -/// Minimal transport implementation used by follower receivers. +/// Bidirectional crossbeam-channel transport for in-process cluster simulation. /// -/// Owns a crossbeam `Receiver` for incoming WAL segments. The `send_segment` -/// side is a no-op — all shipping is managed by the cluster struct. -pub(super) struct ReceiveOnlyTransport { +/// Wraps both sender and receiver so `SimulatedCluster` can call +/// `send_segment` to ship WAL payloads and `spawn_receiver` can call +/// `recv_segment` to apply them. +pub(super) struct ChannelTransport { pub(super) local_shard: ShardId, + pub(super) tx: crossbeam::channel::Sender, pub(super) rx: crossbeam::channel::Receiver, } -impl Transport for ReceiveOnlyTransport { - fn send_segment(&self, _: ShardId, _: WalSegmentPayload) -> Result<(), TransportError> { - Ok(()) +impl Transport for ChannelTransport { + fn send_segment(&self, _to: ShardId, payload: WalSegmentPayload) -> Result<(), TransportError> { + self.tx.send(payload).map_err(|_| TransportError::Closed) } fn recv_segment(&self) -> Option { @@ -30,7 +32,7 @@ impl Transport for ReceiveOnlyTransport { } } -// ── Internal: batch log entry ───────────────────────────────────────────── +// ── Batch log entry ────────────────────────────────────────────────────── pub(super) struct BatchEntry { /// Which leader shard produced this batch. @@ -43,15 +45,12 @@ pub(super) struct BatchEntry { // ── Re-delivery helper ──────────────────────────────────────────────────── -/// Re-deliver all batch-log entries not yet applied to `db`, sending through `tx`. +/// Re-deliver all batch-log entries not yet applied to `db`, sending through +/// the given transport. /// /// Called by `SimulatedCluster::heal_region` for immediate recovery after a /// partition is healed, and by `await_convergence` during the polling loop. -pub(super) fn redeliver_missed( - tx: &crossbeam::channel::Sender, - db: &TidalDb, - log: &[BatchEntry], -) { +pub(super) fn redeliver_missed(transport: &dyn Transport, db: &TidalDb, log: &[BatchEntry]) { for entry in log { let applied = db .replication_state() @@ -63,7 +62,7 @@ pub(super) fn redeliver_missed( bytes: entry.bytes.clone(), event_count: 1, }; - let _ = tx.try_send(payload); + let _ = transport.send_segment(entry.source_shard, payload); } } } diff --git a/tidal/tests/m8_uat.rs b/tidal/tests/m8_uat.rs index c9e7f17..98f9152 100644 --- a/tidal/tests/m8_uat.rs +++ b/tidal/tests/m8_uat.rs @@ -73,6 +73,7 @@ fn three_region_config() -> ClusterConfig { leader_region: RegionId(0), schema: m8_schema(), profiles: Vec::new(), + transports: None, } } @@ -82,6 +83,7 @@ fn two_region_config() -> ClusterConfig { leader_region: RegionId(0), schema: m8_schema(), profiles: Vec::new(), + transports: None, } }