feat(sharding): place regions and primaries onto joining nodes

[austin-barrington]

Sharding partitions each measurement's series_id space into regions, but a
process joining an existing cluster never received any of them. try_split
clones the parent peer set and try_rebalance only moves a primary among
peers already in the region, so an existing measurement's peer set could not
grow to include a new node. A joiner added membership and HA, and no write
capacity.

This is Phase 1 of first-class sharding: a one-member cluster owns regions, a
joining node receives existing region data, and then a primary moves onto it
so writes land there locally.

What changes

• RF is min(configured, live members). A 1-member cluster gets
  peers = [self] instead of an impossible peer set at configured RF 3.
• A joiner installs the committed shard map before any data moves, so
  staging can never run against a lagging epoch (ShardMapPort::replace_map).
• ShardMapOp::AddPeer grows a region's replica set. Apply refuses a
  duplicate peer, a stale epoch, and any region carrying transfer debt.
• Three scheduler placement steps, all stage-then-commit:
  • try_place_live_member — stages a region onto a live member while the
    region is below effective RF, then commits AddPeer.
  • try_place_idle_member — covers 3→4, where no region is under RF and
    AddPeer correctly declines; an existing replica steps aside via
    MovePeer.
  • try_place_primary — hands the primary to the region's newest peer while
    that peer holds strictly fewer primaries than the incumbent.

Staging before committing is the contract here. The heal path's
commit-then-stage ordering would publish a peer holding no rows, and scatter
reads fall back to replicas, so that surfaces as silent empty results.

Primary placement terminates by construction rather than by cooldown. Pure
count balancing cannot satisfy the goal: with one region and two members,
moving the only primary does not improve balance, it swaps which node is idle
— so a balancer either refuses to move or oscillates forever. Keying on the
region's newest peer means the rule stops applying the moment that peer is
primary, and the count guard stops every region stampeding onto one joiner.

Reviewer note: rolling upgrades

AddPeer is a new ShardMapOp variant and ClusterRequest is serialised into
the Raft log. A node on a build that does not know the variant fails to
deserialise the whole append RPC — axum rejects the body before the handler
runs — so it stops replicating, and a leader mid-rollout can lose commit
quorum on the control plane. It resolves as soon as every node is upgraded,
and ingest does not go through Raft.

add_peer_proposals_enabled defaults on, matching
transfer_clear_proposals_enabled, which guards the identical hazard. The flag
is an opt-out for the duration of a rolling upgrade, not an opt-in for the
feature. Disable it before a rollout of a sharded cluster; re-enable when every
node is on the new build.

Review fixes included

The branch was reviewed before this PR; 14 findings, 12 fixed here.

Correctness:

• Rollup (SummingMergeTree) destinations were eligible for placement.
  apply_transfer_push is not idempotent, so a region staged onto a joiner
  whose map op then failed was staged again next tick and summed twice —
  permanent silent corruption. All three paths now share the guard
  enqueue_reconciliation already used, failing closed on lock contention.
• sync_shard_map could install an older peer map. The sync peer is
  whichever Active node came first out of a HashMap, not the leader; an
  older map dropped ops Raft never redelivers, because replace_map bypasses
  the state machine and leaves last_applied untouched. Strictly-newer only.
• A 404 from /internal/shard/map aborted the whole join before WAL
  catch-up, and after the retries the node went Active having synced nothing.
  Best-effort now: with no map there are no regions to move anyway.
• Placement could stall cluster-wide, permanently and quietly: catch-up
  required exact map_version equality while the candidate was chosen before
  any probe and never advanced, so one lagging node blocked every region
  forever with only a debug! line. Catch-up now accepts a candidate that is
  ahead, placement walks candidates, and a persistent stall warns and
  increments hyperbytedb_shard_placement_stalled_total.
• try_rebalance scored a peer with no heartbeat row as 0 bytes. Unmeasured
  is not empty — a freshly staged replica looked infinitely lighter than the
  primary and would have taken ownership on missing telemetry.
• Placement steps acted on the tick's region snapshot, so a placement earlier
  in the same tick made later steps spend a full region copy before failing
  the epoch CAS.

Efficiency:

• /internal/shard/map serialises every space and region to answer one u64
  and was probed once per region per tick. Cached per peer per tick.

Testing

• cargo test --lib — 521 pass
• cargo test --test '*' — 288 pass across 11 suites
• cargo test -p hyperbytedb-cli — 38 pass
• cargo clippy --all-targets -- -D warnings, cargo fmt --check — clean

Integration coverage asserts the contract, not the midpoint: a write aimed at
a replica-only joiner forwards and leaves its WAL empty, and the same write
applies locally once the primary has moved. The tick wiring, the stall skip
and the probe cache are each mutation-tested — reverting them reproduces the
original bug rather than merely failing.

+2570 / −59, of which +846 is tests.

Supersedes

Closes the earlier stack, whose commits are the first two on this branch:

• Allow a one-member sharded cluster to own regions #113 feat/sharding-n1-legal-cluster (f7b550a)
• Install the shard map on join before region movement #114 feat/sharding-joiner-map-catchup (8bdbd99)

Not in this PR

• Server and operator defaults stay off (that is Phase 5). Sharding itself
  remains experimental and opt-in.
• Raft snapshot-install map carry (Phase 2). Join catch-up is met by map
  pull, not by reconstruction after log purge. Related: whether the catch-up
  gate should compare the Raft applied index instead of map_version — it is
  already exposed at /internal/raft/metrics, but replace_map installs a
  map without advancing it, so an applied-index gate would reject exactly the
  joiner it should admit. Belongs with the Phase 2 snapshot work.
• No kind/e2e run. G0–G9 does not cover this phase and the dedicated 1→2
  gate has not been built; placement is proven by in-process tests only.
• Hexagonal cleanup. Placement policy is pure domain logic living in the
  application layer, and peer RPC is raw HTTP there. Planned separately; this
  PR adds no new use crate::adapters imports to application/.
• try_place_idle_member is not join-triggered. It targets the member
  with the greatest joined_at, so a stable-but-unbalanced cluster rebalances
  on the first tick after upgrade. Bounded, convergent, verified before
  commit; documented and accepted while sharding is beta.

Companion: hyperbytedb-operator hyperbyte-cloud/hyperbytedb-operator#15
drops the webhook rule forbidding sharding.enabled at replicas: 1 and
emits [cluster] enabled = true for a 1-replica CR with explicit sharding.
Merge order does not matter; neither side depends on the other at runtime.

🤖 Generated with Claude Code

https://claude.ai/code/session_019Te8YUXjjLssk3hxwoE5Db

[cursor]

Bugbot couldn't run - usage limit reached

Bugbot is counted against Cursor usage for this user or team, and this run hit a usage or spend limit.

A user or team admin can review and increase usage limits in the Cursor dashboard.

(requestId: serverGenReqId_04f83f13-d9f9-4015-a75d-535c538c47b0)

[cursor]

PR Summary

^{Cursor Bugbot is generating a summary for commit e96f7a0. Configure here.}