Preview Repair Support for Tracked Keyspaces

mt-preview-repair.md

REPAIRED preview for fully-migrated tracked keyspaces — design summary

Goal and scope

Provide a REPAIRED preview that audits already-reconciled data across replicas for a tracked keyspace, validating that data backing reconciled offsets is byte-identical across replicas. Reject the operation if the keyspace isn't fully migrated to mutation tracking or if topology changes during any phase. ALL and UNREPAIRED previews on tracked keyspaces are a separate (simpler) design — out of scope here.

Operator-facing interface

Identical to classic REPAIRED preview:

nodetool repair --validate <keyspace>

--validate maps to PreviewKind.REPAIRED (tools/nodetool/Repair.java:96-97, 137-139). The operator does not learn a second command. Routing between classic PreviewRepairTask and the tracked-keyspace path is invisible to them.

Core insight

The audit needs a cut: a per-log offset frontier C such that every live participant is known to have reconciled at least up to C. Below C, replicas must agree by contract; any observed disagreement is a real bug (MT bookkeeping, journal corruption, SSTable corruption, or operator intervention).

Establishing such a cut is exactly what a real MT incremental repair does as a side effect: MutationTrackingIncrementalRepairTask completes only when every live participant's reconciled offsets contain the target captured at IR start (see MutationTrackingSyncCoordinator.ShardSyncState.isComplete — MutationTrackingSyncCoordinator.java:385-402). We abstract this behind a CutEstablisher interface and reuse the IR machinery as the default implementation, but the audit depends on the cut, not on IR per se.

MT IR does not touch SSTable metadata. SSTableWriter.finalizeMetadata:362-378 sets repairedAt only at flush time based on MutationTrackingService.isDurablyReconciled(coordinatorLogOffsets). Compaction can re-check via CompactionTask.getCoordinatorLogOffsets:446-453, but MT IR skips anti-compaction and does no streaming, so it does not indirectly trigger this either. What IR does do — and this is what makes it useful for the audit — is advance every live replica's reconciled watermark to include every offset that any live replica had witnessed at IR start. That is: IR raises the cut, without touching a single SSTable byte.

Why SSTable straddling almost disappears under IR-first

Every SSTable's coordinatorLogOffsets (SSTableWriter.java:88, 117, 385; read via SSTableReader.getCoordinatorLogOffsets() at line 1233) records exactly which (log, offset) pairs were applied to its source memtable. For an SSTable S that existed on any replica at IR start, every offset in S.coordinatorLogOffsets was witnessed by that replica (that's how it got into the memtable), so it's in T = union(witnessed). IR waits until every live replica reconciles ⊇ T, so post-IR C = min(reconciled) ⊇ T. Therefore every offset in S.coordinatorLogOffsets is ≤ C. S is fully below cut — not straddling.

The only SSTables that can straddle are those flushed after IR start (memtables that absorbed post-IR-start writes). These are addressed by snapshotting the SSTable set at IR start on each participant, so any post-IR-start flush produces an SSTable that is not in the audit's set at all.

New classes introduced

• CutEstablisher — interface (in org.apache.cassandra.replication). Contract: given a range set and participant set, produce an AuditCut such that for every live participant p, p.reconciled ⊇ cut, and ensure each participant has captured a stable snapshot of its SSTable set + journal state before the returned future completes. Fail loud on topology change (never return a stale cut) and on participant timeout / unreachable.

  interface CutEstablisher {
      Future<AuditCut> establishCut(RepairJobDesc desc,
                                     NeighborsAndRanges neighborsAndRanges,
                                     Epoch expectedEpoch);
  }
  

  A future implementation could establish the cut without running full IR (e.g. by intersecting reconciled offsets sampled at a single point in time). One implementation for now.

• IncrementalRepairCutEstablisher — the default (and currently only) CutEstablisher implementation, alongside MutationTrackingIncrementalRepairTask in org.apache.cassandra.repair. Orchestrates a two-phase preparation on each participant:

  1. Sends MT_PREVIEW_PREPARE_REQ to every participant. Each participant captures its current SSTable set (via SSTableReader.Refs so compaction can't reclaim the files during audit) and pins a MutationJournal.Snapshot, caching both against parentSession (see AuditPreparationCache below). Responds when the snapshot is stable.
  2. After all participants ack, runs an internal MutationTrackingIncrementalRepairTask for the same range/participant set. On completion, extracts targets from the sync coordinators' ShardSyncState.targets (via the accessor we need to add — currently private at MutationTrackingSyncCoordinator.java:371) and wraps them in an AuditCut.
  3. Completes the returned future with the AuditCut.

  On failure, completes with a CutEstablishmentException wrapping the underlying cause. Failure also triggers snapshot release on every participant that had prepared.

• AuditCut — immutable value object wrapping ImmutableMap<Range<Token>, ImmutableMap<CoordinatorLogId, Offsets.Immutable>> (shard range → per-log offset frontier). Exposes:

  • containsAll(Range<Token> shardRange, ImmutableCoordinatorLogOffsets sstableOffsets) → boolean — used to test whether an SSTable is fully ≤ cut for its overlapping shard.
  • containsMutation(ShortMutationId id) → boolean — used to filter journal-replay mutations. Same shape as ReconciledLogSnapshot.isFullyReconciled(keyspace, ShortMutationId) (ReconciledLogSnapshot.java:49-55), just sourced from the IR-captured cut rather than the local reconciled state.

• AuditPreparationCache — participant-side state, one entry per parentSession, holding (Refs<SSTableReader> sstables, MutationJournal.Snapshot journal, Epoch epoch). Populated on MT_PREVIEW_PREPARE_REQ, read on MT_AUDIT_VALIDATE_REQ, released on MT_AUDIT_VALIDATE_RSP, on MT_PREVIEW_RELEASE_REQ, or on timeout. Bounded by session lifetime; each entry has an idle-timeout (say 2 × mutation_tracking_sync_timeout) so a crashed coordinator doesn't leak refs forever.

• AuditValidationSource — implements the existing ValidationPartitionIterator contract that Validator consumes (repair/Validator.java). Consumes an AuditPreparationCache entry (the snapshotted SSTables + pinned journal) and an AuditCut. Yields rows from SSTables passing cut.containsAll(...) interleaved with journal-replay rows passing cut.containsMutation(...), all filtered to the requested token range.

• MutationTrackingRepairedPreviewTask extends AbstractRepairTask — top-level orchestrator for the audit. Sibling of MutationTrackingIncrementalRepairTask. Uses CutEstablisher for phase 1; does not reference MutationTrackingIncrementalRepairTask directly.

• MutationTrackingPreviewPrepareRequest / Response — RepairMessage payloads for the new verb pair MT_PREVIEW_PREPARE_REQ / MT_PREVIEW_PREPARE_RSP. Sent by IncrementalRepairCutEstablisher to each participant before triggering IR. Request carries (parentSession, range, table, Epoch expectedEpoch). Response is an ack once the snapshot is in place.

• MutationTrackingAuditRequest / Response — RepairMessage payloads for MT_AUDIT_VALIDATE_REQ / MT_AUDIT_VALIDATE_RSP. Request carries (parentSession, range, table, AuditCut, Epoch expectedEpoch). Participant looks up its prepared snapshot by parentSession. Response is an ack; the merkle tree itself ships over the existing Validator → ValidationComplete path.

• MutationTrackingPreviewReleaseRequest — one-way payload for MT_PREVIEW_RELEASE_REQ. Coordinator sends to every participant in step 7 to explicitly release cached snapshots.

Protocol

Step 1 — Gate

At the start of MutationTrackingRepairedPreviewTask.performUnsafe:

• Reject if !ksm.useMutationTracking() or metadata.mutationTrackingMigrationState.isMigrating(keyspace) (existing helpers MutationTrackingIncrementalRepairTask.shouldUseMutationTrackingRepair at MutationTrackingIncrementalRepairTask.java:195-212 and isMutationTrackingMigrationInProgress at lines 223-226 already encode these tests).
• Capture Epoch epochAtStart = ClusterMetadata.current().epoch for topology checks in steps 3 and 4.

Step 2 — Prepare snapshots and establish the cut

Invoke the default CutEstablisher (currently IncrementalRepairCutEstablisher):

CutEstablisher establisher = MutationTrackingService.instance().cutEstablisher();
Future<AuditCut> cutFuture = establisher.establishCut(desc, neighborsAndRanges, epochAtStart);

Emit "establishing cut" progress notification before the call.

Under the hood, IncrementalRepairCutEstablisher does:

1. Prepare: send MT_PREVIEW_PREPARE_REQ to every participant. Each captures Refs<SSTableReader> for the target table + a MutationJournal.Snapshot, caches by parentSession in AuditPreparationCache, and responds. This freezes the auditable state on each participant.
2. Run IR: invoke an internal MutationTrackingIncrementalRepairTask for the same range/participants. Waits for reconciled ⊇ T = union(witnessed) on every live participant.
3. Assemble: extract targets from ShardSyncState.targets, build AuditCut, complete the returned future.

Emit "cut established" on success.

On failure at any sub-step: complete the future with CutEstablishmentException wrapping the underlying cause; also send MT_PREVIEW_RELEASE_REQ to any participants that had already prepared. Surface to the operator as "REPAIRED preview failed: unable to establish audit cut (<cause>)".

Step 3 — Verify topology unchanged between cut establishment and audit

Before dispatching audit-validate RPCs, assert ClusterMetadata.current().epoch.equals(epochAtStart). Fail with a clear message if not, and send MT_PREVIEW_RELEASE_REQ to every participant to free the cached snapshots.

Step 4 — Dispatch audit-validate to all participants

Emit "dispatching audit" progress notification. Send MutationTrackingAuditRequest(parentSession, range, table, AuditCut, epochAtStart) over MT_AUDIT_VALIDATE_REQ. Each participant, on receipt:

1. Verifies ClusterMetadata.current().epoch.equals(expectedEpoch); rejects with TopologyChanged if not.
2. Looks up its AuditPreparationCache entry for parentSession. If missing (e.g., participant restarted between prepare and validate), rejects with PreparationMissing.
3. Constructs an AuditValidationSource over (cache.sstables, cache.journal, AuditCut), feeds it to a Validator, and periodically re-checks the TCM epoch — aborting on any change.
4. Returns the merkle tree via the existing ValidationComplete path.

Step 5 — Audit source semantics (AuditValidationSource)

Tree input is the union of:

• SSTables: iterate the snapshotted Refs<SSTableReader> set (captured at IR start). Include sstable iff auditCut.containsAll(sstable.range, sstable.getCoordinatorLogOffsets()) — strict containment, fully below cut. Because the SSTable set was frozen at IR start and IR raises the cut past every offset witnessed at IR start, every snapshotted SSTable's offsets are ≤ cut. Straddling within the snapshotted set is therefore not possible under normal operation; any straddling SSTables would represent bugs (e.g. a snapshotted SSTable somehow contains offsets that weren't witnessed by any live replica at IR start), so the strict check remains as a defensive guard rather than an expected filter path.
• Journal: replay the pinned MutationJournal.Snapshot, including mutation m iff auditCut.containsMutation(m.id()).

All output filtered to the requested token range.

Any post-IR-start flush is invisible to the audit because the resulting SSTable is not in Refs<SSTableReader> — it's a genuinely new SSTable that appeared after the snapshot. Same for post-IR-start writes that stay in memtables (never in Refs) and post-IR-start journal segments (the pinned MutationJournal.Snapshot covers only segments that existed at pin time).

Step 6 — Compare and report (coordinator)

Tree exchange and pairwise comparison are unchanged from existing repair. Build a SyncStatSummary from the diffs (same shape as PreviewRepairTask.java:75-93). On non-empty summary, increment RepairMetrics.previewFailures and, for any range mismatch, reuse PreviewRepairTask.maybeSnapshotReplicas (PreviewRepairTask.java:114-168) to take diagnostic snapshots on every participant when snapshot_on_repaired_data_mismatch is enabled. Emit "audit complete" progress notification. Emit the reused "Repaired data is in sync" / "Repaired data is inconsistent\n<summary>" message so operators see one output shape across classic and tracked previews.

Metric semantics note: bytesPreviewedDesynchronized and tokenRangesPreviewedDesynchronized are populated using the same SyncStatSummary accounting as classic REPAIRED preview so operators can dashboard uniformly. For tracked keyspaces, "bytes" here counts the merged SSTable + journal-replay stream rather than pure on-disk bytes; the number is close enough to be operationally useful but is not directly comparable to raw disk bytes. Documented at the task's javadoc and in operator-facing release notes.

Step 7 — Cleanup

On success or failure, the coordinator sends MT_PREVIEW_RELEASE_REQ to every participant so it can:

1. Release Refs<SSTableReader> (allowing compaction to reclaim the pinned SSTable files).
2. Close its pinned MutationJournal.Snapshot (MutationJournal.java:119).
3. Remove the entry from AuditPreparationCache.

Participants also apply an idle-timeout to cache entries (say 2 × mutation_tracking_sync_timeout) so a crashed coordinator doesn't leak refs forever. The coordinator emits a final notification with the result.

Routing change

RepairCoordinator.repair (RepairCoordinator.java:557-...) currently routes preview before checking MT:

if (state.options.isPreview())     → PreviewRepairTask
else if (useMutationTracking)      → MutationTrackingIncrementalRepairTask

Becomes:

if (state.options.isPreview() && useMutationTracking) {
  if (previewKind == REPAIRED && !migrating) → MutationTrackingRepairedPreviewTask
  else                                       → reject with clear error
}
else if (state.options.isPreview())          → PreviewRepairTask    // existing
else if (useMutationTracking)                → MutationTrackingIncrementalRepairTask

Progress notifications

MutationTrackingRepairedPreviewTask emits the following via coordinator.notifyProgress(...) at internal milestones. These are the routing marker and latch-coordination surface for tests and the observable pipeline shape for operators inspecting notification streams:

• "establishing cut" — before invoking CutEstablisher.establishCut(...). Earliest MT-specific marker.
• "cut established" — when the cut future resolves successfully. By this point, every participant's SSTable + journal snapshot has already been captured (that happened during the MT_PREVIEW_PREPARE_REQ sub-step, before IR ran).
• "dispatching audit" — before sending MT_AUDIT_VALIDATE_REQ.
• "audit complete" — after collecting all trees and computing the summary. Immediately followed by the reused "Repaired data is in sync" / "Repaired data is inconsistent\n<summary>" message.

Intentionally rejected

• Migration handling. Operator must wait for full migration before running REPAIRED preview.
• Partial results on topology change. Hard failure with retry guidance — no "audit N of M participants" mode.
• Cheap previews. This is at-least cut-establishment cost, which today is at-least IR cost. The UNREPAIRED MT preview (offset arithmetic only, no cut-establishment IR, no streaming) covers the cheap case separately.
• Force flush before audit. Considered as a way to eliminate straddling SSTables; not needed once the SSTable set is snapshotted at IR start and IR raises the cut past all pre-existing witnessed offsets.
• Row-level mutation-ID tagging on SSTables. Would be needed if we were relying on filtering rows out of straddling SSTables. Not needed because snapshotted SSTables can't straddle under the IR-first + snapshot-at-IR-start design. Left as future work if a scenario ever demands it.
• Operator-visible --skip-establish-cut or --audit-only flags. The cut is a required precondition for a meaningful audit, and there is no operator-facing way to bypass it. A future lighter-weight CutEstablisher implementation would preserve the same operator interface.

Reused infrastructure

Validator and its merkle-tree machinery, ValidationComplete message path, range-difference computation, SyncStatSummary reporting, RepairMetrics.previewFailures, PreviewRepairTask.maybeSnapshotReplicas, and the existing MutationJournal.Snapshot / Refs<SSTableReader> / ImmutableCoordinatorLogOffsets / ShortMutationId types. The task does not directly know about MutationTrackingIncrementalRepairTask — that dependency is confined to IncrementalRepairCutEstablisher.

Novel surface area (minimal)

• CutEstablisher interface (small).
• IncrementalRepairCutEstablisher (orchestrates prepare → IR → cut assembly).
• AuditCut (small immutable value object).
• AuditPreparationCache (participant-side, session-keyed).
• AuditValidationSource (interleaves snapshotted SSTable and journal streams into ValidationPartitionIterator shape).
• MutationTrackingRepairedPreviewTask (orchestration).
• Two new verb pairs: MT_PREVIEW_PREPARE_REQ/RSP and MT_AUDIT_VALIDATE_REQ/RSP, plus the one-way MT_PREVIEW_RELEASE_REQ.
• Corresponding RepairMessage payload classes.
• Accessor on MutationTrackingSyncCoordinator.ShardSyncState to expose captured targets to IncrementalRepairCutEstablisher.
• CutEstablishmentException (thin wrapper for failure attribution).