diff --git a/tidal-server/src/cluster/node.rs b/tidal-server/src/cluster/node.rs index 4275f16..f6d2270 100644 --- a/tidal-server/src/cluster/node.rs +++ b/tidal-server/src/cluster/node.rs @@ -123,6 +123,12 @@ pub struct RegionClusterState { /// if a second transport (gRPC reads, native protocol) lands, extract a /// `ShardTransport` trait in `scatter_gather` rather than branching here. blocking_client: reqwest::blocking::Client, + /// Bearer key for this node's OWN inter-sibling POSTs (the heal item/embedding + /// backfill in [`Self::post_marked_blocking`]). The internal marker is a trust + /// signal, NOT an auth bypass — the bearer middleware still runs first — so a + /// backfill POST 401s unless it carries `Authorization: Bearer` when + /// `TIDAL_API_KEY` is set. Same key on every region (required in multi-process). + api_key: Option>, /// Signal type name → `u8` WAL id (the same u8-guarded map `SimulatedCluster` /// builds). The relay stays schema-agnostic; the handler resolves the id. signal_type_ids: HashMap, @@ -170,6 +176,7 @@ impl RegionClusterState { profiles: Vec, data_dir: Option, hlc_offset_ms: i64, + api_key: Option>, ) -> Result { super::topology::validate_multiproc(topology, region_name)?; @@ -284,6 +291,7 @@ impl RegionClusterState { client, broadcast_peer_timeout: topology.broadcast_peer_timeout(), blocking_client, + api_key, signal_type_ids, write_pool, shutting_down: AtomicBool::new(false), @@ -561,7 +569,7 @@ impl RegionClusterState { /// the next heal). Carries the internal marker so the peer applies locally and /// does not re-broadcast. fn post_marked_blocking(&self, url: &str, body: &serde_json::Value) -> bool { - match self + let mut req = self .blocking_client .post(url) .timeout(super::forward::STATUS_PEER_TIMEOUT) @@ -569,9 +577,15 @@ impl RegionClusterState { super::forward::INTERNAL_MARKER, super::forward::INTERNAL_MARKER_VALUE, ) - .json(body) - .send() - { + .json(body); + // The internal marker is a trust signal, not an auth bypass: when the + // cluster runs with TIDAL_API_KEY set, the peer's bearer middleware runs + // first and 401s an unauthenticated backfill. Carry this node's own key + // (identical on every region) so the heal item/embedding backfill lands. + if let Some(key) = &self.api_key { + req = req.bearer_auth(key); + } + match req.send() { Ok(resp) if resp.status().is_success() => true, Ok(resp) => { tracing::warn!(url, status = %resp.status(), "heal backfill peer non-success"); @@ -1442,14 +1456,21 @@ pub async fn create_item( offload_region_read(move || RegionClusterState::apply_item_local(&db, entity, &metadata)) .await?; - if internal { - // A marked broadcast terminates here: no fan-out. + if !state.is_leader() { + // A FOLLOWER applying a marked broadcast/heal terminates here: no + // fan-out. Gating on `internal` alone was a bug: a forwarded item write + // reaches the LEADER with the marker set (forward_write marks it for + // loop-prevention) and so terminated without ever broadcasting — items + // written through a non-leader gateway never replicated to peers (only + // the WAL-relayed signal did). Gate on leadership: a follower with the + // marker terminates; the leader (external OR forwarded) always fans out. return Ok(StatusCode::CREATED.into_response()); } - // Leader external request: broadcast to peers (best-effort, per-peer - // report). 201 asserts the LOCAL create — the single-node /items contract — - // while the body carries the broadcast report (cf. /embeddings' 200, which - // exists only because a 204 cannot carry that body). + // Leader request (external client OR a write forwarded from a follower): + // broadcast to peers (best-effort, per-peer report). 201 asserts the LOCAL + // create — the single-node /items contract — while the body carries the + // broadcast report (cf. /embeddings' 200, which exists only because a 204 + // cannot carry that body). let outcome = broadcast_to_peers(&state, "/items", &req, &headers).await; Ok((StatusCode::CREATED, Json(broadcast_body(&outcome))).into_response()) } @@ -1485,12 +1506,15 @@ pub async fn write_embedding( offload_region_read(move || RegionClusterState::apply_embedding_local(&db, entity, &values)) .await?; - if internal { + if !state.is_leader() { + // Follower applying a marked broadcast/heal: terminate, no fan-out. A + // forwarded embedding write reaches the LEADER with the marker set and + // must still broadcast (see create_item for the full rationale). return Ok(StatusCode::NO_CONTENT.into_response()); } - // Leader external request: broadcast. The per-peer report needs a body, so we - // return 200 with the report (a 204 cannot carry one). A forwarded/internal - // write keeps the bodyless 204 above. + // Leader request (external OR forwarded from a follower): broadcast. The + // per-peer report needs a body, so we return 200 with the report (a 204 + // cannot carry one). A follower's marked write keeps the bodyless 204 above. let outcome = broadcast_to_peers(&state, "/embeddings", &req, &headers).await; Ok((StatusCode::OK, Json(broadcast_body(&outcome))).into_response()) } diff --git a/tidal-server/src/main.rs b/tidal-server/src/main.rs index 44648b3..b8ed858 100644 --- a/tidal-server/src/main.rs +++ b/tidal-server/src/main.rs @@ -241,6 +241,12 @@ async fn run_region_cluster(args: ClusterArgs, region: String) -> Result<()> { ); let data_dir = args.data_dir.clone(); + // Read the bearer key up front: the region node needs its OWN copy to + // authenticate inter-sibling POSTs (the heal item/embedding backfill). The + // internal marker is a trust signal, not an auth bypass, so an unauthenticated + // backfill 401s when TIDAL_API_KEY is set. Same key serves the router below. + let api_key = read_api_key(); + let node_key = api_key.clone(); // `RegionClusterState::new` builds the GrpcTransport via `GrpcTransport::new`, // which blocks on its own tokio runtime — must run off this reactor. Build it // on a dedicated thread, exactly like the single-process path. @@ -254,13 +260,13 @@ async fn run_region_cluster(args: ClusterArgs, region: String) -> Result<()> { profiles, data_dir, hlc_offset_ms, + node_key, ) }) .map_err(|e| ServerError::Cluster(format!("spawn region builder thread: {e}")))? .join() .map_err(|_| ServerError::Cluster("region builder thread panicked".into()))??; - let api_key = read_api_key(); serve_state(state, &args.listen, api_key, build_region_router).await } diff --git a/tidal-server/tests/cluster_region.rs b/tidal-server/tests/cluster_region.rs index 2241662..db8906b 100644 --- a/tidal-server/tests/cluster_region.rs +++ b/tidal-server/tests/cluster_region.rs @@ -111,7 +111,7 @@ impl Pair { fn build_region(topology: TopologySpec, region: &str) -> RegionClusterState { let region = region.to_string(); std::thread::spawn(move || { - RegionClusterState::new(&topology, ®ion, region_schema(), Vec::new(), None, 0) + RegionClusterState::new(&topology, ®ion, region_schema(), Vec::new(), None, 0, None) }) .join() .unwrap() diff --git a/tidal-server/tests/cluster_routes.rs b/tidal-server/tests/cluster_routes.rs index d10f525..0906165 100644 --- a/tidal-server/tests/cluster_routes.rs +++ b/tidal-server/tests/cluster_routes.rs @@ -102,7 +102,7 @@ impl Cluster3 { fn build_region(topology: TopologySpec, region: &str) -> RegionClusterState { let region = region.to_string(); std::thread::spawn(move || { - RegionClusterState::new(&topology, ®ion, region_schema(), Vec::new(), None, 0) + RegionClusterState::new(&topology, ®ion, region_schema(), Vec::new(), None, 0, None) }) .join() .unwrap()