tidaldb/docs/planning/milestone-8/phase-10/task-03-cross-process-routes.md

Task 03: Cross-process routes — forwarding, promote fan-out, status aggregation, reconcile, sharded

Delivers

The multi-process cluster becomes a single coherent HTTP surface: any node accepts any runbook operation and the cluster behaves as one system. Concretely: leader forwarding for writes, region forwarding for reads, promote propagation, aggregated /cluster/status, cross-process CRDT reconciliation (/cluster/reconcile), item/embedding broadcast, and multi-process /sharded/* scatter-gather with honest degraded semantics.

Complexity: XL

Dependencies

Task 02 (RegionClusterState, typed NotLeader/NotLocal errors, peer_http map).

Technical Design

Forwarding client (cluster/forward.rs)

Promote reqwest from dev-dependency to dependency (workspace-consistent version 0.12, features = ["json"], default-features = false + rustls if the workspace builds clean that way; otherwise default features — implementer verifies). One shared reqwest::Client on RegionClusterState (connection pooling), short timeouts: connect 1s, request 5s (status aggregation uses 500ms overall per peer).

Internal-propagation marker: header x-tidal-internal: 1. A request carrying it is NEVER re-forwarded/re-broadcast (loop prevention). Forwarded requests pass through the Authorization header verbatim (all nodes share one TIDAL_API_KEY).

Write forwarding

POST /signals|/items|/embeddings|/hardnegs on a non-leader (without the internal marker): forward body to http://{leader_http}{path}, relay status + body back. With the marker: apply locally only (this is how broadcasts and forwarded writes terminate). Forward happens in the async handler (reqwest is async; no write-pool involvement — the pool is only for local gRPC ship work). Leader unreachable → 503 with a JSON body naming the leader and the connect error.

Item/embedding broadcast (m8p8 parity)

On the leader, POST /items|/embeddings (external request): apply locally, then best-effort fan out the same body to every peer http_addr with the internal marker, concurrently, 2s per-peer timeout. Per-peer failures are WARN-logged and reported in the response body ({"replicated_to": n, "failed": ["ap-south"]} with 201/204 preserved as status); a partitioned peer catches up via operator re-broadcast or the heal-era re-write — document the contract in the handler rustdoc and runbook (items are broadcast config-data, signals are the replicated stream; same split as m8p8).

Promote fan-out

POST /cluster/promote {region} external: validate, apply locally, fan out with internal marker to ALL peers concurrently (2s timeout each), then 200 with {ok, leader, acked: [..], failed: [..]}. Internal: apply locally only. The promote is correct even if some peers miss it (their next forwarded write returns NotLeader→ the forwarding node knows the leader; stale views self-correct on the next status poll — document this as eventual leadership propagation; the UAT asserts all-3 convergence).

/cluster/partition + /cluster/heal external on non-leader: forward to current leader (they mutate leader-side ship-skip state).

Status aggregation

GET /cluster/status on ANY node: query every region's /cluster/status/local (including own, in-process) with 500ms per-peer budget, concurrently. Response keeps the documented shape and adds reachable:

{
  "leader": "us-east",
  "relay_log_len": 125,
  "regions": [
    {"name":"us-east","applied_events":125,"lag_events":0,"partitioned":false,"reachable":true},
    {"name":"ap-south","applied_events":0,"lag_events":125,"partitioned":true,"reachable":false}
  ]
}

• leader = this node's view; relay_log_len = leader's last_seq (from the leader's local status; if THIS node leads, read it directly).
• Per region: applied_events from its local status; lag_events = leader_last_seq.saturating_sub(applied); leader row mirrors today's semantics (applied = its replication-from-others counter, lag 0).
• Unreachable peer: reachable: false, partitioned: true, applied_events: 0, lag_events = leader_last_seq (worst-case honest).

Cross-process reconciliation

Two routes (both utoipa-annotated, bearer-protected):

• POST /cluster/reconcile/snapshot (internal, marker required): body = wire StateSnapshot. Node takes its OWN snapshot first, then reconcile_with(remote), responds with its pre-merge snapshot + elapsed_ms (merge+apply time only).
• POST /cluster/reconcile {region} (operator-facing): this node snapshots itself, POSTs to the target's /cluster/reconcile/snapshot, applies the returned pre-merge snapshot via its own reconcile_with, responds {ok, region, local_elapsed_ms, remote_elapsed_ms, ops_applied}. CRDT merge determinism guarantees both sides converge to the same state.

Multi-process /sharded/*

Introduce a shard-executor seam in scatter_gather.rs: the merge/dedup/diversity logic already operates on per-shard results; factor the per-shard fetch into

trait ShardExec: Send + Sync {
    fn retrieve(&self, shard: RegionId, q: &Retrieve, deadline: Duration) -> Result<Results>;
    fn search(&self, shard: RegionId, q: &Search, deadline: Duration) -> Result<SearchResults>;
}

with the existing in-process impl for ClusterState (zero behavior change — existing scatter-gather tests must pass untouched) and an HTTP impl for RegionClusterState: local region executes locally; remote regions via GET {peer_http}/feed|/search?...&limit=... with the per-shard deadline as the request timeout, marker header set. Timeout/unreachable/error ⇒ that shard lands in unavailable_shards, degraded: true — never an error, never silently dropped (preserve the documented semantics verbatim).

Sharded writes (/sharded/items|embeddings|signals): owning shard via the engine ShardRouter (same hash the in-process path uses); owner == self → local apply; else forward to the owner's http_addr /sharded/* with marker (owner applies locally on marker receipt). Note the data-model split (documented in runbook rewrite): /sharded/* hash-partitions, the non-sharded surface replicates.

Test Strategy

In-process integration tests with 3 RegionClusterStates + real axum servers on loopback (tokio runtimes per node, pattern from cluster_grpc.rs::http_signal_replicates…), no OS processes (tier-3 OS-process coverage is tasks 04/05):

• forwarded_write_lands_on_leader_and_replicates: POST /signals to follower → 204 → converges on all nodes.
• promote_fans_out: promote via follower B → all 3 status views agree; writes to old leader forward to new.
• status_aggregates_all_regions + unreachable peer (stop one node) → reachable: false, worst-case lag.
• reconcile_converges_and_times: divergent hardnegs (hide on A during simulated divergence) → /cluster/reconcile → both sides converged (hide present on both), local_elapsed_ms/remote_elapsed_ms reported; second reconcile is a no-op (idempotent).
• sharded_feed_degrades_honestly: 3 nodes, one stopped → degraded: true, unavailable_shards names it, merged results from the live shards.
• sharded_write_routes_to_owner: entity whose ShardRouter owner is a remote region → forwarded, readable via that region's local feed.
• Existing single-process scatter-gather + cluster tests green (executor seam neutral).

Acceptance Criteria

• Writes to ANY node succeed via leader forwarding; internal marker prevents loops
• Items/embeddings broadcast leader→peers with per-peer failure reporting
• Promote propagates to all peers; partition/heal forward to the leader
• /cluster/status on any node aggregates all regions with reachable honesty
• /cluster/reconcile exchanges snapshots, converges both sides, reports elapsed_ms, idempotent on repeat
• /sharded/feed|search fan out across processes with preserved degraded semantics; sharded writes route to the owning region
• Existing in-process scatter-gather behavior unchanged (seam is refactor-neutral)
• All new routes utoipa-annotated and present in the cluster OpenAPI document
• cargo clippy -p tidal-server -- -D warnings clean; full workspace tests pass