Shard Splitting

[merlimat]

Shard Split Design

Purpose

As an Oxia cluster grows, individual shards can become hot spots — either due to
data volume or request rate. Shard splitting allows an operator to divide a parent
shard into two child shards, each covering half of the parent's hash range. The
parent continues serving traffic until the children are fully caught up, at which
point traffic is atomically cut over to the children.

Goals:

• No downtime: the parent serves traffic throughout the split. The cutover
  causes a brief unavailability window (~100ms), masked by client SDK transparent
  retries — similar to a clean leader election.
• No data loss: children reach quorum commit before the parent is removed.
• Rollback-able: the split can be aborted at any point before cutover.
• Crash-safe: coordinator restarts resume the split from persisted state.

Terminology

Term	Definition
Parent shard	The shard being split
Child shards	The two new shards created by the split (left and right)
Split point	The hash boundary dividing the parent's range into two halves
Observer cursor	A non-voting FollowerCursor on the parent leader that streams data to a child leader without participating in the parent's replication quorum

Split Trigger

Splits are triggered manually via an admin RPC on the coordinator:

service OxiaCoordination {
  rpc SplitShard(SplitShardRequest) returns (SplitShardResponse);
}

message SplitShardRequest {
  string namespace = 1;
  int64 shard = 2;
  // Optional: explicit split points. If omitted, split at midpoint.
  // For N-way split, provide N-1 split points (sorted).
  repeated uint32 split_points = 3;
}

message SplitShardResponse {
  repeated int64 child_shards = 1;
}

The oxia CLI exposes this as oxia admin split-shard. Auto-splitting based on
shard size or load metrics is future work.

No New Shard Status

The split lifecycle is not tracked via a new ShardStatus. The existing
status enum (SteadyState, Election, Deleting) remains unchanged.

The parent shard must remain free to transition between SteadyState and
Election during the split (which may take a long time). If the parent leader
crashes mid-split, the coordinator runs a normal leader election for it. A
Splitting status would conflict with Election and require special-casing the
entire election code path.

Instead, the split state is tracked through a separate SplitMetadata field on
ShardMetadata, orthogonal to the operational status.

Cluster Status Extension

type SplitMetadata struct {
    Phase                   SplitPhase
    ParentShardId           int64              // set on child shards
    ChildShardIDs           []int64            // set on parent shard
    ParentTermAtBootstrap   int64              // detect parent leader election
    ChildLeadersAtBootstrap map[int64]string   // detect child leader election
}

The split points are not stored explicitly — they are derived from the children's
Int32HashRange fields on ShardMetadata. This naturally extends to future N-way
splits (multiple children, each with its own hash range).

SplitMetadata is set on both parent and children. Phase values:
Bootstrap, CatchUp, Cutover.

Data Flow

The parent leader pushes data to child leaders via observer cursors. Each child
leader then replicates to its own ensemble followers:

Parent Leader ──(observer cursor)──> Child-L Leader ──(follower cursors)──> Child-L Followers
              ──(observer cursor)──> Child-R Leader ──(follower cursors)──> Child-R Followers

The observer cursor reuses the existing FollowerCursor machinery: it
automatically sends a snapshot if the child is empty, then tails the WAL. The
only differences from a normal follower cursor are:

• Observer acks do not participate in the parent's QuorumAckTracker.
• The data is filtered by hash range (both snapshot and WAL entries).

Proto Changes

// Extended with observer support:
message AddFollowerRequest {
  // ... existing fields ...
  bool observer = 6;          // non-voting follower
  optional int64 target_shard = 7;  // child shard ID
  optional Int32HashRange split_hash_range = 8;  // filter range
}

// New: clean observer removal (no fencing)
message RemoveObserverRequest {
  string namespace = 1;
  int64 shard = 2;
  int64 term = 3;
  string follower_name = 4;
  int64 target_shard = 5;
}

No new RPCs for snapshot transfer or WAL streaming — the existing SendSnapshot
and Replicate RPCs are reused via the observer FollowerCursor.

Client Behavior During Split

During phases 1–3, the parent is included in shard assignments and clients write
to it normally. Children are excluded from assignments.

At the end of cutover, the coordinator updates shard assignments: parent is
removed, children are added. Clients receive the update via their long-poll
WaitForNextUpdate() and the existing overlap detection in shard_manager.go
handles the transition — child hash ranges overlap the parent's, so the parent is
automatically evicted from the client's shard map.

Between fencing the parent and clients receiving the updated assignments, writes
to the parent's hash range get transient errors. This window is bounded by child
election time (< 1s) plus assignment propagation. All other shards are unaffected.

Handling Writes During Split

During Bootstrap and CatchUp, the parent continues accepting writes normally.
New writes go through the parent's WAL and are streamed to children via the
observer cursors. The children filter and apply only keys in their hash range.

---

Coordinator Split Controller Design

Overview

The split controller drives the state machine through 4 phases. It runs alongside
the parent's ShardController and is persisted in the cluster status so it
survives coordinator restarts. The design prioritizes data safety (no data
loss) and rollback-ability (abort before the point of no return).

Actors

• Split Controller: Runs alongside the parent's ShardController in the
  coordinator. Drives the state machine.
• Shard Controller (coordinator): Manages leader election and health for each
  shard. The split controller coordinates with it.
• Parent Leader: The current leader of the parent shard being split.
• Child Leaders: The nodes selected to lead the two child shards.
• Child Ensemble: The 3 (or RF) nodes assigned to each child shard.

Split Phases

stateDiagram-v2
    [*] --> Initiation
    Initiation --> Bootstrap : MW #1: create children, set Phase=Bootstrap

    Bootstrap --> CatchUp : MW #3,#4: record parentTerm + childLeaders, Phase=CatchUp
    CatchUp --> Bootstrap : parent term changed (parent leader election)
    CatchUp --> Bootstrap : child leader changed (child leader election)
    Bootstrap --> Abort : timeout (pre-cutover)
    CatchUp --> Abort : timeout (pre-cutover)

    state "═══ POINT OF NO RETURN ═══" as PONR
    CatchUp --> PONR : both children caught up (MW #5)
    PONR --> Cutover

    Cutover --> [*] : MW #6-#9: fence parent, elect children, update assignments, trigger DeleteShard

    Abort --> [*] : RemoveObserver, delete children, clear parent Split

    note right of Bootstrap
        MW #2: child Term, Leader
        Fence children (skip if already elected),
        elect leaders, add observer cursors
    end note

    note right of CatchUp
        Round-based: snapshot parent
        commitOffset, wait for children
        commitOffset to reach it.
        Detects parent OR child leader changes.
    end note

    note right of Cutover
        NO ROLLBACK — must complete.
        Fence parent, wait children commit,
        re-elect children in clean term,
        update shard assignments.
        Parent deletion handled by ShardController.
    end note

    note left of Abort
        Safe before Cutover only.
        Parent resumes normal operation.
    end note

Loading

The point of no return is the transition from CatchUp to Cutover. Before that
point, the split can be aborted and the parent restored to normal operation. After
that point, the parent is fenced (stops accepting writes) and the split must
complete.

---

Phase Details

Split Initiation (before the state machine)

The coordinator's InitiateSplit() method is called (e.g. via admin API).

Preconditions (validated before any state changes):

• Parent shard exists and is in SteadyState
• Parent shard has no active split (Split == nil)
• Parent shard has no pending ensemble changes (PendingDeleteShardNodes is empty)
• Parent hash range is large enough to split (max - min >= 1)

Before the split controller starts, the coordinator:

METADATA WRITE #1: Create child shard entries in cluster status with
Split.Phase = Bootstrap, Split.ParentShardId = parent. Set parent's
Split.Phase = Bootstrap, Split.ChildShardIDs = [left, right],
Split.SplitPoint. Create ShardController for each child.

This is the first durable record that a split is in progress. If the coordinator
crashes after this point, restartInProgressSplits() will find the parent's
Split metadata and resume.

---

Phase 1: Bootstrap

Purpose: Validate preconditions, create child shard replicas and start
replicating parent data to them via observer cursors.

Steps:

1. Verify preconditions: Parent shard has a leader and is in SteadyState.
   If not, retry with backoff until healthy or timeout.

For each child shard (left, then right):

2. Fence child ensemble: Send NewTerm(child_shard, term+1) to all child
   ensemble members. Require quorum (majority) to respond.

3. Pick child leader: Select the node with the highest offset from the fencing
   responses. (On first split, all nodes are empty so this is arbitrary.)

METADATA WRITE #2 (per child): Update child shard metadata:
Term = newTerm, Leader = childLeader, Status = SteadyState.

4. Elect child leader: Send BecomeLeader(child, child_leader, followerMap)
   so the child starts replicating to its followers as soon as data arrives. This
   is critical: without an elected leader, only the single child leader node would
   have the data and a crash there loses it. By electing the child leader now, its
   follower cursors replicate received entries to the other ensemble members, and
   commitOffset advances as followers acknowledge.

5. Add observer on parent: Send AddFollower to the parent leader with:

   • observer = true — non-voting, doesn't affect parent's quorum
   • targetShard = child_shard_id — receiving node creates a follower controller
     for the child shard, not the parent
   • term = parent_term — the parent's actual term so the child can validate it
   • splitHashRange = [child.min, child.max] — filter to only replicate keys in
     the child's hash range

METADATA WRITE #3: Record parent.Split.ParentTermAtBootstrap = parentTerm
and parent.Split.ChildLeadersAtBootstrap = {childId → leader.Internal}.

METADATA WRITE #4: Update phase on parent and children:
Split.Phase = CatchUp.

Transition: → CatchUp. There is no need to wait for snapshot completion
here — snapshot delivery and WAL catch-up are treated uniformly by the CatchUp
phase, which waits for children's commitOffset to reach the parent's
commitOffset.

What happens on the data path:

The parent leader pushes 2 copies of its data — one per child leader — via
observer cursors. Each child leader then replicates to its own ensemble followers:

Parent Leader ──(observer cursor)──→ Child-L Leader ──(follower cursors)──→ Child-L Followers
              ──(observer cursor)──→ Child-R Leader ──(follower cursors)──→ Child-R Followers

• The parent leader creates an observer FollowerCursor for each child leader.
• The observer cursor first sends a snapshot (filtered by hash range) to the child,
  then continues streaming WAL entries (also filtered). From the coordinator's
  perspective there is no distinction between these two phases.
• The child's FollowerController applies entries with split filtering:
  • FilterDBForSplit() — post-snapshot, removes out-of-range keys from the DB
  • ApplyLogEntryWithSplitFilter() — during WAL catch-up, skips out-of-range ops
• Since the child is already an elected leader (step 3), it replicates received
  entries to its own followers immediately. The child's commitOffset advances as
  a quorum of the child's followers acknowledge.

Failure handling:

• If NewTerm fails to reach quorum: retry with backoff.
• If BecomeLeader fails: retry with backoff.
• If AddFollower fails (parent leader unreachable): retry with backoff.

Rollback: Remove observer cursors from parent (RemoveObserver RPC, best-effort),
delete child shards from cluster status, clear parent's split metadata.

---

Phase 2: CatchUp

Purpose: Wait for children to be reasonably close to the parent's current
position before cutting over.

Algorithm (round-based):

loop:
    1. Check parentTerm == parentTermAtBootstrap
       If not: parent had a leader election → fall back to Bootstrap
    2. Check childLeaders == childLeadersAtBootstrap
       If not: child had a leader election → RemoveObserver(old), fall back to Bootstrap
    3. Read parent's current commitOffset as the target
    4. Wait (up to 10 seconds) for BOTH children's commitOffset >= target
       - Poll GetStatus(child_leader, child_shard) with backoff
    5. If both children reached the target within 10s → done, advance to Cutover
       If the 10s round expired → go to step 1 (re-read parent commitOffset)

Why round-based: The parent keeps accepting writes during CatchUp. A simple
"lag ≤ N" check is racy because the parent advances continuously. Instead, we
snapshot a target offset and wait for children to reach it. If the round times
out (parent is under heavy write load), we re-read a fresh target and try again.
Eventually, a round will succeed when the children catch up to a snapshot of the
parent's position.

Why commitOffset (not headOffset) for children: The children were elected
leader during Bootstrap (step 5), so they are actively replicating to their
followers. We check commitOffset — not headOffset — because commitOffset
only advances when a quorum of the child's followers have acknowledged the data.
This guarantees that before we cut over, each child shard has a quorum of replicas
with the data, not just a single copy on the child leader. Without this, a child
leader crash before Cutover would lose data.

Transition: When both children's commitOffset >= parent commitOffset within a
round:

METADATA WRITE #5: Update phase on parent and children:
Split.Phase = Cutover.

→ Cutover.

Failure handling:

• Parent leader unreachable: GetStatus fails → retry with backoff.
• Parent term changed (leader election): detected by comparing current term with
  parentTermAtBootstrap:

METADATA WRITE (fallback): Update phase on parent and children:
Split.Phase = Bootstrap (fall back to re-add observers on the new parent leader).

• Child leader changed (child leader election): detected by comparing current
  child leader with childLeadersAtBootstrap:

  1. RemoveObserver(old_child_leader) on parent — best-effort, removes stale cursor
  2. Fall back to Bootstrap — Bootstrap skips re-fencing (children already have
     leaders), re-adds observer cursors for the new child leaders

• Child followers dead (commitOffset stuck): CatchUp round times out, retries.
  If all rounds time out, the overall split timeout triggers abort.

Rollback: Same as Bootstrap — RemoveObserver, delete children, clear parent
split metadata.

---

Phase 3: Cutover

POINT OF NO RETURN: Once this phase starts, the parent is fenced and stops
accepting writes. The split MUST complete. There is no rollback.

Purpose: Freeze the parent, wait for children to commit all remaining data,
then re-elect children in a clean term so they operate independently of the parent.

Steps:

1. Fence parent (NewTerm(parent, term+1) to quorum):
   • Observer cursors on the old leader are terminated (context cancelled).
   • Record parentFinalOffset = max headOffset from fencing responses.

METADATA WRITE #6: Update parent metadata:
Term = newParentTerm, Leader = nil, Status = Election.

2. Wait for children to commit parentFinalOffset (for each child):

   • Poll GetStatus(child_leader, child_shard) until
     commitOffset >= parentFinalOffset.
   • The children were already elected leader during Bootstrap, so they are
     actively replicating to their followers. commitOffset advancing means
     a quorum of child followers have acknowledged all the parent's data.
   • The observer cursors may have already delivered the final entries before
     being terminated, or the child leader may process the last few entries
     from its buffer.

3. Re-elect child leaders in a new term (for each child):

   • NewTerm(child, term+1) to child ensemble (quorum).
   • BecomeLeader(child, existing_leader) — re-elect the same leader.
   • This establishes a clean term for the child shard, independent of the
     parent. The child's follower controller (which was receiving from the
     parent's observer cursor) is replaced by a fresh leader controller.

METADATA WRITE #7 (per child): Update child metadata:
Term = newChildTerm, Leader = childLeader, Status = SteadyState.
Clear Split = nil on children — they are now independent shards.

METADATA WRITE #8: Mark parent Status = Deleting.

4. Update shard assignments: Call computeNewAssignments() and broadcast.
   Since children no longer have Split metadata they are included in the
   assignment map; since parent is Deleting it is excluded. Clients discover
   the two new child shards and start routing to them.

METADATA WRITE #9: Clear parent.Split = nil.

5. Trigger parent deletion: Call SplitComplete() on the event listener.
   The coordinator triggers the parent's ShardController.DeleteShard(), which
   retries DeleteShard RPCs indefinitely with backoff to all ensemble members,
   then removes the parent from cluster status.

Transition: → Done (split controller exits; parent deletion is async).

Failure handling:

• Fencing the parent fails: retry with backoff. If the parent leader is dead,
  the coordinator will elect a new parent leader (higher term), and fencing will
  succeed on the surviving nodes.
• GetStatus on child fails: retry with backoff.
• BecomeLeader fails: retry with backoff.
• Child commit doesn't converge: retry with backoff (the child leader will keep
  replicating to followers).
• DeleteShard to unreachable ensemble member: the shard controller retries
  indefinitely with backoff (not the split controller's concern).

Rollback: NOT POSSIBLE. The parent is fenced. We must complete the split.
The split controller retries all operations indefinitely (within the overall
split timeout).

---

Abort (Pre-Cutover Only)

When the split times out (default 5 minutes) or is cancelled before reaching
Cutover:

1. Remove observer cursors from parent leader (RemoveObserver RPC for each
   child). Best-effort — if the parent leader is dead, skip.

METADATA WRITE (abort-1): Delete child shard entries from cluster status.

METADATA WRITE (abort-2): Clear parent's Split = nil.

2. Notify coordinator via SplitAborted() callback — coordinator closes
   child ShardControllers and recomputes shard assignments.

The parent resumes normal operation.

---

Failure Scenarios

1. Parent leader dies during Init or Bootstrap

What happens: RPCs (NewTerm, AddFollower, GetStatus) fail with
connection errors.

Recovery: The split controller retries with exponential backoff. Meanwhile,
the coordinator's ShardController for the parent detects the failure and
triggers a new leader election (higher term). Once the new parent leader is
elected, the split controller's retries succeed against the new leader.

If timeout expires: Split is aborted. Parent is restored to normal.

Caveat: If observers were already added to the old leader, they are lost when
the old leader dies. Bootstrap will re-add them on the new leader.

2. Parent leader dies during CatchUp

What happens: GetStatus(parent) fails or returns a higher term.

Recovery: The coordinator elects a new parent leader (new term). The CatchUp
phase detects parentTerm != parentTermAtBootstrap and falls back to
Bootstrap. Bootstrap skips re-fencing (children already have leaders),
re-adds observer cursors on the new parent leader, and the process resumes.

Note: The children may have partial data from the old leader. The new
observer cursors resume from the children's current state — the new parent leader
sends a snapshot if the child is behind the WAL's first offset, otherwise tails
from where the child left off.

3. Parent leader dies during Cutover

What happens: If the parent is already fenced (step 1 complete), the observer
cursors are already terminated and children have the data up to parentFinalOffset.
The remaining steps (elect children, wait for commit) don't involve the parent.

If fence partially succeeded: The split controller retries NewTerm on the
parent ensemble. Since a quorum already accepted the new term, subsequent attempts
will succeed (idempotent with same term).

Known issue: If the old parent leader dies mid-fence and a new leader is
elected with a different term, the split controller may get confused about which
term to use. This needs additional handling.

4. Child leader dies during CatchUp

What happens: The observer cursor on the parent targets the old (dead) child
leader. Data stops flowing to the child.

Recovery: The coordinator's shard controller detects the child leader failure
and elects a new child leader (higher term, updated metadata). The CatchUp phase
detects childLeader != childLeadersAtBootstrap[childId] and:

1. Sends RemoveObserver(old_leader) to the parent (best-effort, cleans up stale
   cursor)
2. Falls back to Bootstrap, which re-adds an observer cursor targeting the new
   child leader

5. Child followers dead (commitOffset stuck)

What happens: The child leader receives data from the parent's observer cursor
(headOffset advances) but the child's followers are unreachable, so commitOffset
stays at -1. The CatchUp round times out repeatedly.

Recovery: If the followers come back, commitOffset resumes and the split
completes. If they don't, the overall split timeout (default 5 minutes) expires
and the split is aborted.

6. Follower/ensemble member dies (any phase)

What happens: Quorum operations (NewTerm, BecomeLeader) succeed as long
as a majority of the ensemble responds. Parent deletion is handled by the shard
controller's deleteShardWithRetries() which retries indefinitely with backoff.

7. Coordinator crashes and restarts (any phase)

The split state machine is designed to survive coordinator failures. All split
state is persisted in the cluster status (stored in the metadata service), and
all operations are idempotent, so a new coordinator instance can resume from
exactly where the old one left off.

Persistence: The cluster status contains, for each shard involved in a split:

• Split.Phase — the current phase (Bootstrap, CatchUp, Cutover)
• Split.ChildShardIDs — on the parent shard, identifies the two children
• Split.ParentShardId — on child shards, points back to the parent
• Split.SplitPoint — the hash boundary
• Split.ParentTermAtBootstrap — used to detect stale parent observer cursors
• Split.ChildLeadersAtBootstrap — used to detect child leader elections

Startup resume flow (NewCoordinator() in coordinator.go):

1. Load cluster status from persistent metadata.
2. Create ShardController for every shard (including children in a split).
   Child ShardControllers detect Split != nil with no ChildShardIDs and
   enter waitForSplitComplete() — they do NOT trigger leader elections or
   balancing, letting the SplitController manage the child's lifecycle.
3. Call restartInProgressSplits(clusterStatus):
   • Scan all shards for those with Split != nil and len(ChildShardIDs) > 0
     (only parent shards match).
   • For each parent with an active split, create a new SplitController.
4. Each SplitController calls currentPhase() to read the persisted phase
   and enters driveStateMachine() which resumes from that phase.

Resume behavior per phase:

Crash during	Persisted phase	Resume behavior
Bootstrap (before observers added)	Bootstrap	Re-fences children (skips if already elected), re-adds observers. Safe: NewTerm with a term the servers already accepted is idempotent.
Bootstrap (after observers added, before phase update)	Bootstrap	Re-runs Bootstrap. Old observer cursors on the parent leader are stale (parent leader may have changed). New AddFollower replaces them.
CatchUp	CatchUp	Re-enters CatchUp polling. Old observer cursors may still be running on the parent leader (if the parent leader survived). If the parent or child leader changed, CatchUp detects the term/leader change and falls back to Bootstrap.
Cutover (before parent fenced)	Cutover	Re-runs fenceEnsemble(parent). If the parent was already fenced with this term, the NewTerm is idempotent.
Cutover (after parent fenced, before child election)	Cutover	Re-runs the full Cutover. parentFinalOffset is re-computed from the fencing responses (same value since parent is fenced and not accepting writes). Child election is idempotent.
Cutover (after child election)	Cutover	Re-runs. BecomeLeader on a node that's already leading is safe. Commit wait resumes. Parent deletion triggered via SplitComplete.

Idempotency guarantees that make resume safe:

• NewTerm(shard, term) — If the server already accepted this term, it returns
  its current head entry. No state change.
• AddFollower(observer) — If an observer cursor for this follower already exists,
  the leader replaces it (closes old, creates new). If the follower name doesn't
  match any existing cursor, a new one is created.
• BecomeLeader(shard, term) — If the node is already leading this shard at this
  term, it returns success.
• DeleteShard(shard, term) — If the shard is already deleted, returns success.

Edge case — coordinator crashes during abort:
If the coordinator crashes while aborting (pre-Cutover), the new coordinator will
see the split metadata still present and attempt to resume the split (not re-abort).
This is acceptable because:

• The new coordinator starts a fresh SplitController with a full timeout.
• If the underlying issue that caused the timeout is resolved (e.g., a dead node
  came back), the split can complete successfully.
• If the issue persists, the new SplitController will time out again and abort.

8. Split timeout

What happens: The split controller's context expires (default 5 minutes).

If phase < Cutover: Abort procedure runs (remove observers, delete children,
restore parent).

If phase >= Cutover: The timeout doesn't trigger an abort. The split must
complete. The operations continue retrying. (In practice, post-cutover operations
should complete quickly since they're just elections and status polling.)

---

Sequence Diagram: Happy Path

sequenceDiagram
    participant SC as Split Controller
    participant PL as Parent Leader
    participant CL as Child-L Leader
    participant CR as Child-R Leader
    participant CE as Child Ensemble

    note over SC,CE: BOOTSTRAP

    SC->>CE: NewTerm(child-L, t+1)
    CE-->>SC: quorum ack
    SC->>CL: BecomeLeader(child-L)
    note right of CL: NOW LEADING — replicates to followers

    SC->>PL: AddFollower(observer, child-L, hashRange=[0,split])
    PL->>CL: snapshot (filtered)
    PL->>CL: WAL entries (filtered)
    CL->>CE: replicate to child-L followers
    note right of CL: commitOffset advances

    SC->>CE: NewTerm(child-R, t+1)
    CE-->>SC: quorum ack
    SC->>CR: BecomeLeader(child-R)
    note right of CR: NOW LEADING

    SC->>PL: AddFollower(observer, child-R, hashRange=(split,max])
    PL->>CR: snapshot (filtered)
    PL->>CR: WAL entries (filtered)
    CR->>CE: replicate to child-R followers

    note over SC,CE: CATCHUP (round-based)

    loop until both children caught up
        SC->>PL: GetStatus → commitOffset = N
        SC->>CL: GetStatus → commitOffset >= N?
        SC->>CR: GetStatus → commitOffset >= N?
        note over SC: If both >= N within 10s → Cutover / Else re-read parent commitOffset
    end

    note over SC,CE: ═══ POINT OF NO RETURN ═══

    note over SC,CE: CUTOVER

    SC->>PL: NewTerm(parent, t+1)
    note right of PL: FENCED — leader=nil, observers killed
    PL-->>SC: headOffset = N (parentFinalOffset)

    SC->>CL: poll GetStatus until commitOffset >= N
    SC->>CR: poll GetStatus until commitOffset >= N

    SC->>CE: NewTerm(child-L, t+1)
    SC->>CL: BecomeLeader(child-L)
    note right of CL: CLEAN TERM

    SC->>CE: NewTerm(child-R, t+1)
    SC->>CR: BecomeLeader(child-R)
    note right of CR: CLEAN TERM

    note over SC: Update shard assignments (children added, parent excluded)
    note over SC: SplitComplete() → ShardController.DeleteShard(parent) [async]
    note over SC: Children now serve traffic

Loading

Sequence Diagram: Parent Leader Dies During CatchUp

sequenceDiagram
    participant SC as Split Controller
    participant ShC as Shard Controller
    participant PL_old as Parent Leader (old)
    participant PL_new as Parent Leader (new)
    participant Ch as Children

    note over SC,Ch: CATCHUP in progress

    SC->>PL_old: GetStatus → commitOffset = N
    SC->>Ch: GetStatus(child-L)

    PL_old-xPL_old: CRASH

    ShC->>ShC: detects failure
    ShC->>PL_new: NewTerm(parent, t+1)
    ShC->>PL_new: BecomeLeader(new)
    note right of PL_new: new parent leader elected

    SC->>SC: reads parent metadata — parentTerm != parentTermAtBootstrap
    SC->>SC: updatePhase(Bootstrap)

    note over SC,Ch: BOOTSTRAP restart

    note over SC: Children already have leaders — skip fencing

    SC->>PL_new: AddFollower(observer, child-L)
    PL_new->>Ch: snapshot + WAL
    SC->>PL_new: AddFollower(observer, child-R)
    PL_new->>Ch: snapshot + WAL

    note over SC,Ch: → CatchUp → Cutover → Done

Loading

Sequence Diagram: Child Leader Dies During CatchUp

sequenceDiagram
    participant SC as Split Controller
    participant ShC as Shard Controller (child)
    participant PL as Parent Leader
    participant CL_old as Child-L Leader (old)
    participant CL_new as Child-L Leader (new)

    note over SC,CL_new: CATCHUP in progress

    PL->>CL_old: WAL entries (observer cursor)

    CL_old-xCL_old: CRASH

    ShC->>ShC: detects child-L failure
    ShC->>CL_new: NewTerm(child-L, t+1)
    ShC->>CL_new: BecomeLeader(child-L)
    note right of CL_new: new child-L leader elected

    SC->>SC: reads child-L metadata — leader != childLeadersAtBootstrap
    SC->>PL: RemoveObserver(old child-L leader)
    SC->>SC: updatePhase(Bootstrap)

    note over SC,CL_new: BOOTSTRAP restart

    note over SC: Children already have leaders — skip fencing

    SC->>PL: AddFollower(observer, new child-L leader)
    PL->>CL_new: snapshot + WAL (to new leader)

    note over SC,CL_new: → CatchUp → Cutover → Done

Loading

Sequence Diagram: Split Timeout (Abort)

sequenceDiagram
    participant SC as Split Controller
    participant PL as Parent Leader
    participant Ch as Children

    note over SC,Ch: BOOTSTRAP or CATCHUP (stuck/slow)

    SC->>SC: context.WithTimeout expires (5min)

    note over SC,Ch: ABORT

    SC->>PL: RemoveObserver(child-L) [best-effort]
    SC->>PL: RemoveObserver(child-R) [best-effort]
    SC->>SC: Delete child-L, child-R from cluster status
    SC->>SC: Clear parent.Split = nil
    SC->>SC: SplitAborted() → close child ShardControllers

    note over PL: Parent resumes normal operation

Loading

Sequence Diagram: Coordinator Crashes During CatchUp and Resumes

sequenceDiagram
    participant C_old as Coordinator (old)
    participant MS as Metadata Store
    participant PL as Parent Leader
    participant Ch as Children

    note over C_old,Ch: CATCHUP

    C_old->>MS: updatePhase(CatchUp)
    note right of MS: persisted: parent.Split.Phase = CatchUp

    C_old->>Ch: polling children...
    PL->>Ch: WAL entries (observer cursors active)

    C_old-xC_old: CRASH

    note over PL,Ch: Observer cursors keep streaming if parent leader is alive

    participant C_new as Coordinator (new)

    C_new->>MS: Load cluster status
    MS-->>C_new: parent.Split.Phase = CatchUp

    C_new->>C_new: Create ShardControllers (children: waitForSplitComplete)
    C_new->>C_new: restartInProgressSplits() → new SplitController

    alt parent leader survived (same term)
        C_new->>PL: GetStatus → commitOffset = N
        C_new->>Ch: GetStatus → commitOffset >= N
        note over C_new: observers still valid, resume polling
    else parent leader died (new term)
        C_new->>C_new: parentTerm != parentTermAtBootstrap → fall back to Bootstrap
    end

    note over C_new,Ch: → Cutover → Done

Loading

Sequence Diagram: Coordinator Crashes During Cutover and Resumes

sequenceDiagram
    participant C_old as Coordinator (old)
    participant MS as Metadata Store
    participant PE as Parent Ensemble
    participant Ch as Children

    note over C_old,Ch: CUTOVER

    C_old->>MS: updatePhase(Cutover)
    C_old->>PE: NewTerm(parent, t+1)
    PE-->>C_old: parentFinalOffset = N (parent fenced)

    C_old->>Ch: Wait children commitOffset >= N
    Ch-->>C_old: OK

    C_old->>Ch: NewTerm(child-L, t+1) + BecomeLeader(child-L)

    C_old-xC_old: CRASH (before child-R election)

    participant C_new as Coordinator (new)

    C_new->>MS: Load cluster status
    MS-->>C_new: parent.Split.Phase = Cutover

    C_new->>C_new: restartInProgressSplits() → SplitController, phase = Cutover

    C_new->>PE: NewTerm(parent, t+1) — IDEMPOTENT, same parentFinalOffset
    C_new->>Ch: children already committed >= N

    C_new->>Ch: NewTerm(child-L) + BecomeLeader → idempotent
    C_new->>Ch: NewTerm(child-R) + BecomeLeader → NEW

    C_new->>Ch: Wait commitOffset >= N
    Ch-->>C_new: OK

    note over C_new,Ch: → Done

Loading

---

Key Invariants

1. Observer cursors use the parent's actual term (not -1). The child validates
   the term — if a parent leader election occurs, the old term entries are rejected,
   and the CatchUp phase detects the term change and falls back to Bootstrap.

2. CatchUp detects both parent and child leader elections. Parent term changes
   and child leader changes are both detected by comparing against values stored at
   Bootstrap time (ParentTermAtBootstrap, ChildLeadersAtBootstrap). Both trigger
   a fallback to Bootstrap with RemoveObserver for stale cursors.

3. Bootstrap is idempotent and incremental. On re-run, it skips fencing children
   that already have leaders (elected by the shard controller). It only re-adds
   observer cursors on the parent for the current child leaders.

4. Abort is safe before Cutover. No data has been moved — children are just
   passive observers. Removing them and clearing metadata restores the system to
   the pre-split state.

5. Post-Cutover is forward-only. Once the parent is fenced, it stops accepting
   writes. The children must be promoted. All post-cutover operations retry
   indefinitely.

6. All operations are idempotent. NewTerm with the same term is accepted.
   AddFollower for an existing observer is handled. BecomeLeader on a node
   that's already leading is safe. This enables safe resume after coordinator
   restart.

7. Child quorum commit before cutover. In CatchUp we wait for children's
   commitOffset >= target (not headOffset). In Cutover step 2 we wait for
   commitOffset >= parentFinalOffset. This ensures data is replicated to a
   quorum of child followers before the parent is deleted, so a child leader
   failure doesn't lose data.

8. Parent deletion is delegated to ShardController. The split controller
   marks the parent as Deleting and calls SplitComplete(). The coordinator
   triggers the parent shard controller's DeleteShard() which retries
   indefinitely with backoff. This reuses the existing deletion mechanism and
   survives coordinator restarts (parent is persisted with Status=Deleting).

---

Future Work

• Auto-split: Trigger splits based on shard size (Pebble DB size) or request
  rate metrics.
• Merge shards: Reverse operation to combine underutilized shards.
• N-way split: Split into more than 2 children in a single operation.

[mattisonchao]

hmm.. It seems like we are losing quorum assurance when cutover.

[mattisonchao]

I'm considering an alternative approach that might be simpler to implement:

1. Support WAL checkpoint via hard link.
2. Fence all replicas, pass the new shard IDs to the ensemble, let each node do a local checkpoint (hard link) to the target directories, then close the parent shard and return.
3. Coordinator starts the two child shard controllers and triggers the election.
4. Split complete.

However, this approach has some drawbacks:

1. Requires compaction filter logic to lazily clean up out-of-range data in the background. We might need to memorise the keySeq or SST names.
2. Requires filtering out-of-range data on read paths (list, range_scan, notifications) until cleanup is complete.
3. partition_key makes this significantly worse — since it's stored in the value, every read requires decoding the value just to determine key ownership, which is expensive.

[mattisonchao]

I draft an alternative one.

Shard Split Design: Local-First Approach

Alternative Proposal to Discussion #909

---

Overview

This document proposes an alternative approach to shard splitting in Oxia that avoids network data replication entirely by leveraging local-first operations on each ensemble member. The core insight is that since all three replicas of a shard already hold a complete copy of the data, we can have each node independently prepare the child shards locally — eliminating snapshot transfer, observer followers, and complex catch-up state machines.

The approach draws inspiration from CockroachDB's range split mechanism, where splits are treated as lightweight metadata operations with no data movement. While Oxia's per-shard Pebble instances and partition_key support prevent an exact replica of that design, the same principle of locality-first can be applied.

---

Limitations of the Existing Proposal

The original proposal is thorough and correct, but introduces significant complexity to achieve low downtime:

• Observer followers require modifying the AddFollower RPC and FollowerCursor machinery to support non-voting participants.
• Multi-phase state machine (Init → Bootstrap → Catch-Up → Cutover → Cleanup) with restart recovery at each phase adds coordinator complexity.
• Network snapshot transfer sends the full Pebble checkpoint across the network to child leaders, even though the same data already exists locally on each node.
• BecomeLeader timing must be carefully sequenced to ensure child followers never receive unfiltered data.
• Leader failover during catch-up requires the coordinator to re-attach observer followers to the new leader after each election.

The fundamental reason for this complexity is that the original proposal treats the problem as a replication problem — getting data from parent to child. This proposal reframes it as a local transformation problem.

---

Proposed Approach: Local-First Background Split

High-Level Flow

┌─────────────┐     ┌───────────────────────┐     ┌──────────────┐
│  1. START   │────>│  2. BACKGROUND SPLIT  │────>│  3. CUTOVER  │
└─────────────┘     └───────────────────────┘     └──────────────┘
  Coordinator          Each node locally              Fence parent,
  sends StartSplit      iterates DB + tails WAL        checkpoint split
  (returns immediately) into child DBs                 DBs, close parent,
                        (no network transfer)           elect children

Phase 1: Background Split

The coordinator sends a StartSplit RPC to all ensemble members (returns immediately). Each node independently starts a background goroutine:

Each ensemble member (Node A, B, C) independently:

  ┌─────────────────────────────────────────────────────┐
  │                 Background Goroutine                 │
  │                                                      │
  │  1. Record current WAL offset = N                    │
  │     (WAL trimmer anchored at N, no deletion past N)  │
  │                                                      │
  │  2. Open two new local Pebble instances:             │
  │       left-child-tmp/  → hash range [min, splitPoint]│
  │       right-child-tmp/ → hash range [splitPoint+1, max]│
  │                                                      │
  │  3. Iterate parent DB                                │
  │     for each key:                                    │
  │       hash = xxh332(partition_key ?? key)            │
  │       if hash in [min, splitPoint]  → write to left  │
  │       if hash in [splitPoint+1, max]→ write to right │
  │       (notifications filtered per batch)             │
  │                                                      │
  │  4. Concurrently tail WAL from offset N              │
  │     for each new entry:                              │
  │       apply + filter by hash range to left/right     │
  │                                                      │
  │  5. When DB iteration complete AND WAL nearly        │
  │     caught up → report SplitReady(commitOffset)      │
  └─────────────────────────────────────────────────────┘

The goroutine acts as both a DB splitter and a WAL consumer simultaneously. Under high write pressure it spends more time on WAL catch-up; under low pressure it makes faster progress on DB iteration. The split child DBs are continuously kept current with parent writes throughout this phase.

Data Flow During Background Split

                    ┌──────────────────────────────────────┐
                    │           COORDINATOR                 │
                    │                                       │
                    │  1. Allocate child shard IDs          │
                    │  2. Send StartSplit to ensemble       │
                    │  3. Wait for SplitReady from all      │
                    │  4. Trigger cutover                   │
                    │  5. Broadcast new ShardAssignments    │
                    └────────────┬─────────────────────────┘
                                 │ StartSplit (fire and forget)
              ┌──────────────────┼──────────────────┐
              ▼                  ▼                   ▼
         ┌─────────┐        ┌─────────┐        ┌─────────┐
         │ Node A  │        │ Node B  │        │ Node C  │
         │         │        │         │        │         │
         │ Parent  │ normal │ Parent  │ normal │ Parent  │
         │ replica │◀──────▶│ replica │◀──────▶│ replica │
         │ (serve) │ replic.│ (serve) │ replic.│ (serve) │
         └────┬────┘        └────┬────┘        └────┬────┘
              │ local             │ local             │ local
              │ split goroutine   │ split goroutine   │ split goroutine
              ▼                   ▼                   ▼
    ┌──────────────────┐  ┌──────────────────┐  ┌──────────────────┐
    │ left-child-tmp/  │  │ left-child-tmp/  │  │ left-child-tmp/  │
    │ [min, splitPoint]│  │ [min, splitPoint]│  │ [min, splitPoint]│
    ├──────────────────┤  ├──────────────────┤  ├──────────────────┤
    │ right-child-tmp/ │  │ right-child-tmp/ │  │ right-child-tmp/ │
    │ [split+1, max]   │  │ [split+1, max]   │  │ [split+1, max]   │
    └──────────────────┘  └──────────────────┘  └──────────────────┘

  No data crosses the network during this phase.
  Each node works entirely from its local disk.

WAL Trimmer Anchor

Parent WAL:
  ──────────────────────────────────────────────────────▶
  0          N (split start)              head (moving)
             │                             │
             │◀──── background goroutine ──┘
             │      tails and applies this
             │      range to child DBs
             │
  Trimmer:   │
  [0, N)  → can be trimmed normally
  [N, →)  → PROTECTED until cutover completes

Phase 2: Cutover

Once all ensemble members report SplitReady, the coordinator triggers cutover. The downtime window is limited to the operations below:

Coordinator signals cutover
         │
         ▼
┌────────────────────────────────────────────────────────────┐
│                   Each ensemble member                      │
│                                                             │
│  1. Fence parent (stop accepting writes)                    │
│                                                             │
│  2. Close parent DB                                         │
│                                                             │
│  3. Checkpoint WAL (hard link, records final offset M)      │
│                                                             │
│  4. Checkpoint the already-split child DBs                  │
│     (hard link left-child-tmp/ → left-child/                │
│      hard link right-child-tmp/ → right-child/)             │
│     These DBs are already filtered — no full scan needed    │
│                                                             │
│  5. Return and close parent shard                           │
└────────────────────────────────────────────────────────────┘
         │
         ▼
Coordinator starts child shard controllers
         │
         ▼
┌────────────────────────────────────────────────────────────┐
│                  Each child shard (on election)             │
│                                                             │
│  1. Open child DB (already filtered, from checkpoint)       │
│                                                             │
│  2. Apply incremental WAL from commitOffset → M             │
│     (tiny window — goroutine was nearly caught up)          │
│     Filter entries by hash range during apply               │
│                                                             │
│  3. Election completes → begin serving traffic              │
└────────────────────────────────────────────────────────────┘
         │
         ▼
Coordinator broadcasts new ShardAssignments
  - Remove parent shard
  - Add left child  [min, splitPoint]
  - Add right child [splitPoint+1, max]
         │
         ▼
Clients switch via existing overlap detection in shard_manager.go

Downtime Window

  ──────────────────────────────────────────────────────────▶ time
  
  Background split running (no downtime, parent serves normally)
  ══════════════════════════════════╗
                                    ║ fence
                                    ╠══════════════════╗
                                    │  close DB         ║
                                    │  checkpoint WAL   ║  downtime
                                    │  checkpoint DBs   ║  window
                                    │  election         ║
                                    ╚══════════════════╝
                                                        │
                                                        ▼
                                                 children serving

The downtime window contains only local operations (no network transfer) plus leader election. It is bounded by the time to checkpoint (hard link, milliseconds) plus election time (typically < 1 second).

---

Key Design Decisions

Child DBs Are Clean at Cutover

By the time cutover happens, each child DB has been continuously updated by the background goroutine throughout the preparation phase. The DB checkpoint at cutover captures this already-filtered state. Child shards only need to apply a tiny incremental WAL window (from commitOffset to the fenced offset M) with hash range filtering — there is no full scan, no lazy cleanup, and no compaction filter needed.

Notifications Require Explicit Filtering

During background preparation, each node filters notification batches as it writes them to the child directories:

• Deserialize the NotificationBatch protobuf
• Remove entries whose key hash falls outside the child's range
• Discard the entire batch if it becomes empty after filtering

Children inherit the parent's offset numbering, allowing clients to resume from their last seen offset after re-subscribing to the child shards.

partition_key Handling

Keys with an explicit partition_key must be routed based on the partition key's hash, not the record key's hash. Both the background split goroutine and the incremental WAL apply step use StorageEntry.partition_key (for DB entries) or PutRequest.partition_key (for WAL entries) when present — consistent with the original proposal's key classification rules.

---

Comparison with Original Proposal

Aspect	Original Proposal	This Proposal
Data transfer	Network snapshot from parent leader to child leaders	Local disk I/O only; no network transfer
Coordinator complexity	5-phase state machine with restart recovery	2-phase: StartSplit + Cutover
New replication machinery	Observer followers, non-voting FollowerCursor, QuorumAckTracker changes	None; existing WAL and DB APIs only
Child DB state at cutover	Filtered during/after snapshot load	Already filtered by background goroutine
Incremental WAL at cutover	Children tail parent WAL via observer FollowerCursor	Tiny local WAL window applied with hash filter
Downtime window	Short (children pre-warmed before cutover)	Short (background goroutine catches up before cutover)
Leader failover during split	Coordinator must re-attach observer to new leader	Each node is independent; failover handled normally
Post-cutover cleanup	Parent deletion	Parent deletion (same); no compaction filter needed
Proto changes	AddFollowerRequest.observer, SplitShard RPC	StartSplit, CancelSplit, SplitReady RPCs only

---

Failure Handling

Node Failure During Background Preparation

Each node operates independently. A single node failure does not affect the others. The failed node restarts, discards its partial child directories, and the coordinator re-sends StartSplit. Other nodes' preparation work is unaffected.

Parent Leader Failure During Preparation

The parent shard controller runs a normal election — no special handling needed. The WAL trimmer anchor at offset N remains in place on all nodes, so no incremental data is lost. After the new leader is elected and resumes normal replication, each node's background goroutine continues consuming WAL entries from the updated replication stream.

Coordinator Failure

The split state is persisted in ClusterStatus. On restart, the coordinator inspects SplitMetadata on the parent and child shards and resumes from the appropriate phase. StartSplit is idempotent — nodes that have already started their background goroutine ignore duplicate requests.

Cutover Failure

Once the parent is fenced, the cutover must complete. Child election failures are retried. If a child node is unavailable, the standard ensemble change mechanism handles it — identical to the original proposal.

Aborting Before Cutover

Coordinator sends CancelSplit
  → Each node stops background goroutine
  → Deletes child tmp directories
  → Releases WAL trimmer anchor
  → Parent continues serving normally
  (no state changes required on parent)

---

Implementation Plan

Milestone 1: Core Infrastructure

1. Add SplitMetadata and SplitPhase to cluster status model (same as original proposal)
2. Add SplitShard admin RPC to coordinator and split-shard CLI command
3. Add StartSplit / CancelSplit RPCs (fire-and-forget)
4. Add SplitReady status reporting from ensemble members to coordinator
5. Add WAL trimmer anchor (prevent segment deletion past a given offset)

Milestone 2: Background Split Goroutine

6. Implement per-node split goroutine: DB iteration and key classification/routing per the key classification rules in the original proposal
7. Implement notification filtering during preparation
8. Implement concurrent WAL consumption in the goroutine to keep child DBs current
9. Implement SplitReady reporting once WAL catch-up is sufficiently close to head

Milestone 3: Cutover

10. Implement fence + close DB + WAL checkpoint + DB checkpoint sequence on each node
11. Start child shard controllers, apply incremental WAL with hash range filter, trigger election
12. Update ShardAssignments broadcast; verify client overlap detection works correctly

Milestone 4: Cleanup and Hardening

13. Parent shard deletion after cutover
14. Coordinator restart recovery for in-progress splits
15. CancelSplit support
16. Integration tests for the full split lifecycle
17. Failure injection tests (node failures, leader failover, coordinator restart at each phase)

---

Future Work

• Auto-split: trigger splits automatically based on shard size (Pebble DB size) or request rate metrics, consistent with the original proposal.
• Merge shards: the reverse operation. Since child shards are co-located on the same nodes immediately after a split, a merge can similarly be done as a local operation before any rebalancing.
• Post-split rebalancing: after cutover, the coordinator migrates child replicas to different nodes to distribute load, using the existing ensemble change mechanism.
• N-way split: split into more than 2 children in a single operation by extending the background goroutine to route into N child directories simultaneously.