diff --git a/distribution/autosplit/detector.go b/distribution/autosplit/detector.go new file mode 100644 index 000000000..cf1c6f12a --- /dev/null +++ b/distribution/autosplit/detector.go @@ -0,0 +1,628 @@ +package autosplit + +import ( + "bytes" + "sort" + "time" + + "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/keyviz" +) + +const ( + defaultWriteWeight = 4 + defaultReadWeight = 1 + defaultThresholdOpsMin = 50_000 + defaultCandidateWindows = 3 + defaultMaxRoutes = 1024 + defaultMaxSplitsPerCycle = 1 + opsPerMinute = 60 +) + +// SplitOrigin describes how an automatic split key was selected. +type SplitOrigin string + +const ( + SplitOriginP50Mid SplitOrigin = "p50_mid" + SplitOriginP50LastBucketLo SplitOrigin = "p50_last_bucket_lo" + SplitOriginP50FirstBucketHi SplitOrigin = "p50_first_bucket_hi" +) + +// SkipReason explains why a route did not produce a split decision. +type SkipReason string + +const ( + SkipReasonNoSplitKey SkipReason = "no_split_key" + SkipReasonRouteCap SkipReason = "route_cap" + SkipReasonBudgetExhausted SkipReason = "budget_exhausted" + SkipReasonNonActiveState SkipReason = "non_active_state" + SkipReasonAggregateRow SkipReason = "aggregate_row" + SkipReasonCooldown SkipReason = "cooldown" + SkipReasonInvalidWindow SkipReason = "invalid_window" +) + +// Config controls the pure detector and scheduler admission checks. +type Config struct { + WriteWeight float64 + ReadWeight float64 + ThresholdOpsMin float64 + CandidateWindows int + MaxRoutes int + MaxSplitsPerCycle int +} + +// DefaultConfig returns the M3 detector defaults from the design doc. +func DefaultConfig() Config { + return Config{ + WriteWeight: defaultWriteWeight, + ReadWeight: defaultReadWeight, + ThresholdOpsMin: defaultThresholdOpsMin, + CandidateWindows: defaultCandidateWindows, + MaxRoutes: defaultMaxRoutes, + MaxSplitsPerCycle: defaultMaxSplitsPerCycle, + } +} + +// RouteKey is the per-column aggregation key for keyviz rows. +type RouteKey struct { + RouteID uint64 + RaftGroupID uint64 +} + +// RouteLoad is the route-level load aggregated from a committed keyviz column. +type RouteLoad struct { + Reads uint64 + Writes uint64 + ReadBytes uint64 + WriteBytes uint64 +} + +// ColumnWindow is a committed keyviz column plus its proven committed duration. +// +// keyviz.MatrixColumn does not yet carry WindowStart. Runtime integration can +// derive Duration from the previous contiguous MatrixColumn.At boundary and pass +// only committed windows here. +type ColumnWindow struct { + Column keyviz.MatrixColumn + Duration time.Duration +} + +// Input is one pure detector evaluation. +type Input struct { + Routes []distribution.RouteDescriptor + Windows []ColumnWindow + Now time.Time +} + +// Decision is a scheduler-ready automatic split decision. +type Decision struct { + RouteID uint64 + SplitKey []byte + SplitOrigin SplitOrigin + TargetGroupID uint64 + RouteDelta int + + ScoreOpsMin float64 + ConsecutiveOver int + LeftLoad float64 + RightLoad float64 +} + +// Event records a deterministic skip or reset reason from an evaluation. +type Event struct { + RouteID uint64 + Reason SkipReason + At time.Time +} + +// Result is the complete output of one detector evaluation. +type Result struct { + Decisions []Decision + Events []Event +} + +// RouteStatus is the observable detector state for one route. +type RouteStatus struct { + ConsecutiveOver int + CooldownUntil time.Time + LastProcessedAt time.Time +} + +// DetectorState carries leader-local confidence and cooldown state. +type DetectorState struct { + routes map[uint64]RouteStatus +} + +// NewDetectorState creates empty detector state. +func NewDetectorState() *DetectorState { + return &DetectorState{routes: map[uint64]RouteStatus{}} +} + +// RouteStatus returns the current detector state for routeID. +func (s *DetectorState) RouteStatus(routeID uint64) RouteStatus { + if s == nil { + return RouteStatus{} + } + return s.routes[routeID] +} + +// ResetConfidence clears candidate confidence while preserving cooldown. +func (s *DetectorState) ResetConfidence(routeID uint64) { + if s == nil { + return + } + s.ensure() + status := s.routes[routeID] + status.ConsecutiveOver = 0 + s.routes[routeID] = status +} + +// ApplyRouteState applies a catalog state transition to the detector state. +// +// Non-active route states clear confidence and advance the processed watermark +// through the newest committed column the caller intentionally skipped. +func (s *DetectorState) ApplyRouteState(routeID uint64, state distribution.RouteState, processedThrough time.Time) { + if state != distribution.RouteStateActive { + s.resetConfidenceThrough(routeID, processedThrough) + } +} + +// SetCooldown blocks routeID from promotion until until. +func (s *DetectorState) SetCooldown(routeID uint64, until time.Time) { + if s == nil { + return + } + s.ensure() + status := s.routes[routeID] + status.CooldownUntil = until + status.ConsecutiveOver = 0 + s.routes[routeID] = status +} + +// AggregateColumnRows groups all non-aggregate rows by (RouteID, RaftGroupID). +func AggregateColumnRows(col keyviz.MatrixColumn) map[RouteKey]RouteLoad { + out := make(map[RouteKey]RouteLoad) + for _, row := range col.Rows { + if row.Aggregate { + continue + } + key := RouteKey{RouteID: row.RouteID, RaftGroupID: row.RaftGroupID} + load := out[key] + load.Reads += row.Reads + load.Writes += row.Writes + load.ReadBytes += row.ReadBytes + load.WriteBytes += row.WriteBytes + out[key] = load + } + return out +} + +// Evaluate consumes committed keyviz windows and emits scheduler-ready decisions. +func Evaluate(cfg Config, state *DetectorState, in Input) Result { + cfg = cfg.withDefaults() + if state == nil { + state = NewDetectorState() + } else { + state.ensure() + } + + latestHot := map[uint64]candidate{} + windows := append([]ColumnWindow(nil), in.Windows...) + sort.SliceStable(windows, func(i, j int) bool { + return windows[i].Column.At.Before(windows[j].Column.At) + }) + live, active, result := prepareRoutes(state, in.Routes, newestWindowAt(windows)) + state.gc(live) + + for _, window := range windows { + processWindow(cfg, state, active, window, in.Now, latestHot, &result) + } + selectDecisions(cfg, state, live, latestHot, &result) + + return result +} + +func prepareRoutes( + state *DetectorState, + routes []distribution.RouteDescriptor, + processedThrough time.Time, +) (map[uint64]distribution.RouteDescriptor, []distribution.RouteDescriptor, Result) { + live := make(map[uint64]distribution.RouteDescriptor, len(routes)) + active := make([]distribution.RouteDescriptor, 0, len(routes)) + result := Result{} + for _, route := range routes { + live[route.RouteID] = route + if route.State != distribution.RouteStateActive { + state.ApplyRouteState(route.RouteID, route.State, processedThrough) + result.Events = append(result.Events, Event{RouteID: route.RouteID, Reason: SkipReasonNonActiveState}) + continue + } + active = append(active, route) + } + return live, active, result +} + +func newestWindowAt(windows []ColumnWindow) time.Time { + if len(windows) == 0 { + return time.Time{} + } + return windows[len(windows)-1].Column.At +} + +func processWindow( + cfg Config, + state *DetectorState, + active []distribution.RouteDescriptor, + window ColumnWindow, + now time.Time, + latestHot map[uint64]candidate, + result *Result, +) { + if window.Duration <= 0 { + result.Events = append(result.Events, Event{Reason: SkipReasonInvalidWindow, At: window.Column.At}) + for _, route := range active { + if state.resetConfidenceAt(route.RouteID, window.Column.At) { + delete(latestHot, route.RouteID) + } + } + return + } + aggregated := aggregateColumn(window.Column) + for _, row := range aggregated.aggregateRows { + result.Events = append(result.Events, Event{RouteID: row.RouteID, Reason: SkipReasonAggregateRow, At: window.Column.At}) + } + + for _, route := range active { + processRouteWindow(cfg, state, route, window, now, aggregated, latestHot, result) + } +} + +func processRouteWindow( + cfg Config, + state *DetectorState, + route distribution.RouteDescriptor, + window ColumnWindow, + now time.Time, + aggregated columnAggregation, + latestHot map[uint64]candidate, + result *Result, +) { + status := state.routes[route.RouteID] + if !window.Column.At.After(status.LastProcessedAt) { + return + } + if now.Before(status.CooldownUntil) || windowOverlapsCooldown(window, status.CooldownUntil) { + status.ConsecutiveOver = 0 + status.LastProcessedAt = window.Column.At + state.routes[route.RouteID] = status + delete(latestHot, route.RouteID) + result.Events = append(result.Events, Event{RouteID: route.RouteID, Reason: SkipReasonCooldown, At: window.Column.At}) + return + } + + key := RouteKey{RouteID: route.RouteID, RaftGroupID: route.GroupID} + load := aggregated.loads[key] + score := scoreOpsPerMinute(load, window.Duration, cfg) + if score < cfg.ThresholdOpsMin { + status.ConsecutiveOver = 0 + status.LastProcessedAt = window.Column.At + state.routes[route.RouteID] = status + delete(latestHot, route.RouteID) + return + } + + status.ConsecutiveOver++ + status.LastProcessedAt = window.Column.At + state.routes[route.RouteID] = status + latestHot[route.RouteID] = candidate{ + route: route, + rows: aggregated.rows[key], + scoreOpsMin: score, + consecutiveOver: status.ConsecutiveOver, + } +} + +func selectDecisions( + cfg Config, + state *DetectorState, + live map[uint64]distribution.RouteDescriptor, + latestHot map[uint64]candidate, + result *Result, +) { + ordered := make([]candidate, 0, len(latestHot)) + for routeID, candidate := range latestHot { + if state.routes[routeID].ConsecutiveOver >= cfg.CandidateWindows { + ordered = append(ordered, candidate) + } + } + sort.Slice(ordered, func(i, j int) bool { + if ordered[i].scoreOpsMin == ordered[j].scoreOpsMin { + return ordered[i].route.RouteID < ordered[j].route.RouteID + } + return ordered[i].scoreOpsMin > ordered[j].scoreOpsMin + }) + + reservedDelta := 0 + for _, candidate := range ordered { + if len(result.Decisions) >= cfg.MaxSplitsPerCycle { + result.Events = append(result.Events, Event{RouteID: candidate.route.RouteID, Reason: SkipReasonBudgetExhausted}) + continue + } + decision, ok := selectP50Decision(cfg, candidate) + if !ok { + result.Events = append(result.Events, Event{RouteID: candidate.route.RouteID, Reason: SkipReasonNoSplitKey}) + continue + } + if cfg.MaxRoutes > 0 && len(live)+reservedDelta+decision.RouteDelta > cfg.MaxRoutes { + result.Events = append(result.Events, Event{RouteID: candidate.route.RouteID, Reason: SkipReasonRouteCap}) + continue + } + reservedDelta += decision.RouteDelta + result.Decisions = append(result.Decisions, decision) + } +} + +func windowOverlapsCooldown(window ColumnWindow, cooldownUntil time.Time) bool { + if cooldownUntil.IsZero() { + return false + } + return window.Column.At.Before(cooldownUntil) || window.Column.At.Add(-window.Duration).Before(cooldownUntil) +} + +type candidate struct { + route distribution.RouteDescriptor + rows []keyviz.MatrixRow + scoreOpsMin float64 + consecutiveOver int +} + +type columnAggregation struct { + loads map[RouteKey]RouteLoad + rows map[RouteKey][]keyviz.MatrixRow + aggregateRows []keyviz.MatrixRow +} + +func (s *DetectorState) ensure() { + if s.routes == nil { + s.routes = map[uint64]RouteStatus{} + } +} + +func (s *DetectorState) resetConfidenceThrough(routeID uint64, through time.Time) { + if s == nil { + return + } + s.ensure() + status := s.routes[routeID] + status.ConsecutiveOver = 0 + if through.After(status.LastProcessedAt) { + status.LastProcessedAt = through + } + s.routes[routeID] = status +} + +func (s *DetectorState) resetConfidenceAt(routeID uint64, at time.Time) bool { + if s == nil { + return false + } + s.ensure() + status := s.routes[routeID] + if !at.After(status.LastProcessedAt) { + return false + } + status.ConsecutiveOver = 0 + status.LastProcessedAt = at + s.routes[routeID] = status + return true +} + +func (s *DetectorState) gc(live map[uint64]distribution.RouteDescriptor) { + for routeID := range s.routes { + if _, ok := live[routeID]; !ok { + delete(s.routes, routeID) + } + } +} + +func (cfg Config) withDefaults() Config { + defaults := DefaultConfig() + if cfg.WriteWeight == 0 && cfg.ReadWeight == 0 { + cfg.WriteWeight = defaults.WriteWeight + cfg.ReadWeight = defaults.ReadWeight + } + if cfg.ThresholdOpsMin == 0 { + cfg.ThresholdOpsMin = defaults.ThresholdOpsMin + } + if cfg.CandidateWindows <= 0 { + cfg.CandidateWindows = defaults.CandidateWindows + } + if cfg.MaxRoutes <= 0 { + cfg.MaxRoutes = defaults.MaxRoutes + } + if cfg.MaxSplitsPerCycle <= 0 { + cfg.MaxSplitsPerCycle = defaults.MaxSplitsPerCycle + } + return cfg +} + +func aggregateColumn(col keyviz.MatrixColumn) columnAggregation { + out := columnAggregation{ + loads: make(map[RouteKey]RouteLoad), + rows: make(map[RouteKey][]keyviz.MatrixRow), + } + for _, row := range col.Rows { + if row.Aggregate { + out.aggregateRows = append(out.aggregateRows, row) + continue + } + key := RouteKey{RouteID: row.RouteID, RaftGroupID: row.RaftGroupID} + load := out.loads[key] + load.Reads += row.Reads + load.Writes += row.Writes + load.ReadBytes += row.ReadBytes + load.WriteBytes += row.WriteBytes + out.loads[key] = load + out.rows[key] = append(out.rows[key], row) + } + return out +} + +func scoreOpsPerMinute(load RouteLoad, duration time.Duration, cfg Config) float64 { + seconds := duration.Seconds() + if seconds <= 0 { + return 0 + } + writeRate := float64(load.Writes) / seconds * opsPerMinute + readRate := float64(load.Reads) / seconds * opsPerMinute + return writeRate*cfg.WriteWeight + readRate*cfg.ReadWeight +} + +func selectP50Decision(cfg Config, candidate candidate) (Decision, bool) { + key, origin, leftLoad, rightLoad, ok := selectP50SplitKey(cfg, candidate.route, candidate.rows) + if !ok { + return Decision{}, false + } + return Decision{ + RouteID: candidate.route.RouteID, + SplitKey: key, + SplitOrigin: origin, + TargetGroupID: 0, + RouteDelta: 1, + ScoreOpsMin: candidate.scoreOpsMin, + ConsecutiveOver: candidate.consecutiveOver, + LeftLoad: leftLoad, + RightLoad: rightLoad, + }, true +} + +func selectP50SplitKey(cfg Config, route distribution.RouteDescriptor, rows []keyviz.MatrixRow) ([]byte, SplitOrigin, float64, float64, bool) { + rows = usableSubBucketRows(rows) + if len(rows) == 0 { + return nil, "", 0, 0, false + } + sortSubBucketRows(rows) + rows = coalesceSubBucketRows(rows) + + median, total, leftLoad, loadBeforeMedian, ok := weightedMedianRow(cfg, rows) + if !ok { + return nil, "", 0, 0, false + } + splitKey, origin, leftLoad, rightLoad := splitKeyForMedian(route, median, total, leftLoad, loadBeforeMedian) + if !splitKeyInsideRoute(route, splitKey) { + return nil, "", 0, 0, false + } + return distribution.CloneBytes(splitKey), origin, leftLoad, rightLoad, true +} + +func sortSubBucketRows(rows []keyviz.MatrixRow) { + sort.Slice(rows, func(i, j int) bool { + if rows[i].SubBucket == rows[j].SubBucket { + return bytes.Compare(rows[i].Start, rows[j].Start) < 0 + } + return rows[i].SubBucket < rows[j].SubBucket + }) +} + +func coalesceSubBucketRows(rows []keyviz.MatrixRow) []keyviz.MatrixRow { + if len(rows) <= 1 { + return rows + } + out := rows[:0] + for _, row := range rows { + if len(out) == 0 || !sameSubBucketRow(out[len(out)-1], row) { + out = append(out, row) + continue + } + last := &out[len(out)-1] + last.Reads += row.Reads + last.Writes += row.Writes + last.ReadBytes += row.ReadBytes + last.WriteBytes += row.WriteBytes + } + return out +} + +func sameSubBucketRow(a, b keyviz.MatrixRow) bool { + return a.RouteID == b.RouteID && + a.RaftGroupID == b.RaftGroupID && + a.SubBucket == b.SubBucket && + a.SubBucketCount == b.SubBucketCount && + bytes.Equal(a.Start, b.Start) && + bytes.Equal(a.End, b.End) +} + +func weightedMedianRow(cfg Config, rows []keyviz.MatrixRow) (keyviz.MatrixRow, float64, float64, float64, bool) { + total := 0.0 + for _, row := range rows { + total += rowWeightedLoad(cfg, row) + } + if total <= 0 { + return keyviz.MatrixRow{}, 0, 0, 0, false + } + var median keyviz.MatrixRow + loadBeforeMedian := 0.0 + leftLoad := 0.0 + cumulative := 0.0 + for _, row := range rows { + load := rowWeightedLoad(cfg, row) + cumulative += load + leftLoad += load + if cumulative >= total/2 { + median = row + break + } + loadBeforeMedian = leftLoad + } + return median, total, leftLoad, loadBeforeMedian, true +} + +func splitKeyForMedian( + route distribution.RouteDescriptor, + median keyviz.MatrixRow, + total float64, + leftLoad float64, + loadBeforeMedian float64, +) ([]byte, SplitOrigin, float64, float64) { + rightLoad := total - leftLoad + + splitKey := median.End + origin := SplitOriginP50Mid + if median.SubBucket == 0 && bytes.Equal(median.Start, route.Start) { + origin = SplitOriginP50FirstBucketHi + } + if splitKey == nil || (route.End != nil && bytes.Compare(splitKey, route.End) >= 0) { + splitKey = median.Start + origin = SplitOriginP50LastBucketLo + leftLoad = loadBeforeMedian + rightLoad = total - leftLoad + } + return splitKey, origin, leftLoad, rightLoad +} + +func usableSubBucketRows(rows []keyviz.MatrixRow) []keyviz.MatrixRow { + out := make([]keyviz.MatrixRow, 0, len(rows)) + for _, row := range rows { + if row.Aggregate || row.SubBucketCount <= 1 { + continue + } + out = append(out, row) + } + return out +} + +func rowWeightedLoad(cfg Config, row keyviz.MatrixRow) float64 { + return float64(row.Writes)*cfg.WriteWeight + float64(row.Reads)*cfg.ReadWeight +} + +func splitKeyInsideRoute(route distribution.RouteDescriptor, splitKey []byte) bool { + if len(splitKey) == 0 && len(route.Start) == 0 { + return false + } + if bytes.Compare(splitKey, route.Start) <= 0 { + return false + } + if route.End != nil && bytes.Compare(splitKey, route.End) >= 0 { + return false + } + return true +} diff --git a/distribution/autosplit/detector_test.go b/distribution/autosplit/detector_test.go new file mode 100644 index 000000000..493358c68 --- /dev/null +++ b/distribution/autosplit/detector_test.go @@ -0,0 +1,714 @@ +package autosplit + +import ( + "testing" + "time" + + "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/keyviz" + "github.com/stretchr/testify/require" +) + +func TestAggregateSubBucketRowsBeforeScoring(t *testing.T) { + t.Parallel() + at := time.Unix(100, 0) + route := testRoute(7, 2, "a", "z") + window := testWindow(at, time.Minute, + testRow(7, 2, 0, 4, "a", "b", 20, 0), + testRow(7, 2, 1, 4, "b", "c", 20, 0), + testRow(7, 2, 2, 4, "c", "d", 20, 0), + keyviz.MatrixRow{RouteID: 99, RaftGroupID: 2, Aggregate: true, Writes: 1000}, + ) + + loads := AggregateColumnRows(window.Column) + require.Equal(t, RouteLoad{Writes: 60}, loads[RouteKey{RouteID: 7, RaftGroupID: 2}]) + require.NotContains(t, loads, RouteKey{RouteID: 99, RaftGroupID: 2}) + + state := NewDetectorState() + result := Evaluate(Config{ + WriteWeight: 1, + ReadWeight: 0, + ThresholdOpsMin: 50, + CandidateWindows: 1, + MaxSplitsPerCycle: 1, + }, state, Input{Routes: []distribution.RouteDescriptor{route}, Windows: []ColumnWindow{window}, Now: at}) + + require.Len(t, result.Decisions, 1) + require.Equal(t, uint64(7), result.Decisions[0].RouteID) + require.Equal(t, []byte("c"), result.Decisions[0].SplitKey) + require.InDelta(t, 60, result.Decisions[0].ScoreOpsMin, 0.0001) + requireEvent(t, result.Events, 99, SkipReasonAggregateRow) +} + +func TestReadWeightContributesToScore(t *testing.T) { + t.Parallel() + at := time.Unix(150, 0) + route := testRoute(2, 3, "a", "z") + window := testWindow(at, time.Minute, + testRow(2, 3, 0, 2, "a", "m", 0, 20), + testRow(2, 3, 1, 2, "m", "z", 0, 20), + ) + + result := Evaluate(Config{ + WriteWeight: 0, + ReadWeight: 1, + ThresholdOpsMin: 30, + CandidateWindows: 1, + MaxSplitsPerCycle: 1, + }, NewDetectorState(), Input{Routes: []distribution.RouteDescriptor{route}, Windows: []ColumnWindow{window}, Now: at}) + + require.Len(t, result.Decisions, 1) + require.InDelta(t, 40, result.Decisions[0].ScoreOpsMin, 0.0001) +} + +func TestInvalidWindowIsSkipped(t *testing.T) { + t.Parallel() + at := time.Unix(175, 0) + route := testRoute(1, 1, "a", "z") + window := testWindow(at, 0, + testRow(1, 1, 0, 2, "a", "m", 100, 0), + testRow(1, 1, 1, 2, "m", "z", 100, 0), + ) + + result := Evaluate(testConfig(), NewDetectorState(), Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{window}, + Now: at, + }) + + require.Empty(t, result.Decisions) + requireEvent(t, result.Events, 0, SkipReasonInvalidWindow) +} + +func TestInvalidWindowResetsActiveRouteConfidence(t *testing.T) { + t.Parallel() + at := time.Unix(180, 0) + route := testRoute(1, 1, "a", "z") + cfg := testConfig() + cfg.CandidateWindows = 3 + + result := Evaluate(cfg, NewDetectorState(), Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{ + hotWindow(at), + hotWindow(at.Add(time.Minute)), + testWindow(at.Add(2*time.Minute), 0, + testRow(1, 1, 0, 2, "a", "m", 100, 0), + testRow(1, 1, 1, 2, "m", "z", 100, 0), + ), + hotWindow(at.Add(3 * time.Minute)), + }, + Now: at.Add(3 * time.Minute), + }) + + require.Empty(t, result.Decisions) + requireEvent(t, result.Events, 0, SkipReasonInvalidWindow) +} + +func TestWindowsAreEvaluatedChronologicallyWithoutMutatingInput(t *testing.T) { + t.Parallel() + at := time.Unix(190, 0) + route := testRoute(1, 1, "a", "z") + cfg := testConfig() + cfg.CandidateWindows = 2 + windows := []ColumnWindow{ + hotWindow(at.Add(time.Minute)), + hotWindow(at.Add(2 * time.Minute)), + coldWindow(at), + } + + result := Evaluate(cfg, NewDetectorState(), Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: windows, + Now: at.Add(2 * time.Minute), + }) + + require.Len(t, result.Decisions, 1) + require.Equal(t, at.Add(time.Minute), windows[0].Column.At) + require.Equal(t, at.Add(2*time.Minute), windows[1].Column.At) + require.Equal(t, at, windows[2].Column.At) +} + +func TestNonActiveRouteResetsConfidence(t *testing.T) { + t.Parallel() + at := time.Unix(200, 0) + state := NewDetectorState() + cfg := testConfig() + cfg.CandidateWindows = 2 + + active := testRoute(1, 1, "a", "z") + first := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{active}, + Windows: []ColumnWindow{hotWindow(at)}, + Now: at, + }) + require.Empty(t, first.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) + + migrating := active + migrating.State = distribution.RouteStateMigratingSource + reset := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{migrating}, + Windows: []ColumnWindow{hotWindow(at.Add(time.Minute))}, + Now: at.Add(time.Minute), + }) + require.Empty(t, reset.Decisions) + require.Equal(t, 0, state.RouteStatus(1).ConsecutiveOver) + requireEvent(t, reset.Events, 1, SkipReasonNonActiveState) + + again := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{active}, + Windows: []ColumnWindow{hotWindow(at.Add(2 * time.Minute))}, + Now: at.Add(2 * time.Minute), + }) + require.Empty(t, again.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) +} + +func TestNonActiveRouteAdvancesSkippedBufferedColumns(t *testing.T) { + t.Parallel() + at := time.Unix(225, 0) + state := NewDetectorState() + active := testRoute(1, 1, "a", "z") + cfg := testConfig() + cfg.CandidateWindows = 2 + + first := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{active}, + Windows: []ColumnWindow{hotWindow(at)}, + Now: at, + }) + require.Empty(t, first.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) + + migrating := active + migrating.State = distribution.RouteStateMigratingSource + skipped := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{migrating}, + Windows: []ColumnWindow{hotWindow(at.Add(time.Minute))}, + Now: at.Add(time.Minute), + }) + require.Empty(t, skipped.Decisions) + require.Equal(t, 0, state.RouteStatus(1).ConsecutiveOver) + require.Equal(t, at.Add(time.Minute), state.RouteStatus(1).LastProcessedAt) + requireEvent(t, skipped.Events, 1, SkipReasonNonActiveState) + + replayed := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{active}, + Windows: []ColumnWindow{ + hotWindow(at.Add(time.Minute)), + hotWindow(at.Add(2 * time.Minute)), + }, + Now: at.Add(2 * time.Minute), + }) + require.Empty(t, replayed.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) +} + +func TestApplyRouteStateAdvancesProcessedWatermark(t *testing.T) { + t.Parallel() + at := time.Unix(250, 0) + state := NewDetectorState() + route := testRoute(1, 1, "a", "z") + cfg := testConfig() + cfg.CandidateWindows = 2 + + first := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{hotWindow(at)}, + Now: at, + }) + require.Empty(t, first.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) + + state.ApplyRouteState(1, distribution.RouteStateMigratingSource, at.Add(time.Minute)) + require.Equal(t, 0, state.RouteStatus(1).ConsecutiveOver) + require.Equal(t, at.Add(time.Minute), state.RouteStatus(1).LastProcessedAt) + + result := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{ + hotWindow(at.Add(time.Minute)), + hotWindow(at.Add(2 * time.Minute)), + }, + Now: at.Add(2 * time.Minute), + }) + + require.Empty(t, result.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) +} + +func TestConsecutiveWindowsPromoteAndBelowThresholdResets(t *testing.T) { + t.Parallel() + at := time.Unix(300, 0) + state := NewDetectorState() + route := testRoute(1, 1, "a", "z") + cfg := testConfig() + cfg.CandidateWindows = 3 + + result := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{ + hotWindow(at), + hotWindow(at.Add(time.Minute)), + coldWindow(at.Add(2 * time.Minute)), + hotWindow(at.Add(3 * time.Minute)), + hotWindow(at.Add(4 * time.Minute)), + }, + Now: at.Add(4 * time.Minute), + }) + require.Empty(t, result.Decisions) + require.Equal(t, 2, state.RouteStatus(1).ConsecutiveOver) + + promoted := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{hotWindow(at.Add(5 * time.Minute))}, + Now: at.Add(5 * time.Minute), + }) + require.Len(t, promoted.Decisions, 1) + require.Equal(t, 3, state.RouteStatus(1).ConsecutiveOver) +} + +func TestDuplicateColumnsAreSkippedAcrossEvaluations(t *testing.T) { + t.Parallel() + at := time.Unix(325, 0) + state := NewDetectorState() + route := testRoute(1, 1, "a", "z") + cfg := testConfig() + cfg.CandidateWindows = 2 + + first := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{hotWindow(at)}, + Now: at, + }) + require.Empty(t, first.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) + + duplicate := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{hotWindow(at)}, + Now: at.Add(30 * time.Second), + }) + require.Empty(t, duplicate.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) + + next := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{hotWindow(at.Add(time.Minute))}, + Now: at.Add(time.Minute), + }) + require.Len(t, next.Decisions, 1) + require.Equal(t, 2, state.RouteStatus(1).ConsecutiveOver) +} + +func TestStaleInvalidWindowDoesNotResetConfidence(t *testing.T) { + t.Parallel() + at := time.Unix(330, 0) + state := NewDetectorState() + route := testRoute(1, 1, "a", "z") + cfg := testConfig() + cfg.CandidateWindows = 3 + + warmup := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{ + hotWindow(at), + hotWindow(at.Add(time.Minute)), + }, + Now: at.Add(time.Minute), + }) + require.Empty(t, warmup.Decisions) + require.Equal(t, 2, state.RouteStatus(1).ConsecutiveOver) + + result := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{ + testWindow(at, 0, + testRow(1, 1, 0, 2, "a", "m", 100, 0), + testRow(1, 1, 1, 2, "m", "z", 100, 0), + ), + hotWindow(at.Add(2 * time.Minute)), + }, + Now: at.Add(2 * time.Minute), + }) + + require.Len(t, result.Decisions, 1) + require.Equal(t, 3, state.RouteStatus(1).ConsecutiveOver) + requireEvent(t, result.Events, 0, SkipReasonInvalidWindow) +} + +func TestBufferedColdColumnAfterHotRunSuppressesDecision(t *testing.T) { + t.Parallel() + at := time.Unix(350, 0) + state := NewDetectorState() + route := testRoute(1, 1, "a", "z") + cfg := testConfig() + cfg.CandidateWindows = 3 + + result := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{ + hotWindow(at), + hotWindow(at.Add(time.Minute)), + hotWindow(at.Add(2 * time.Minute)), + coldWindow(at.Add(3 * time.Minute)), + }, + Now: at.Add(3 * time.Minute), + }) + + require.Empty(t, result.Decisions) + require.Equal(t, 0, state.RouteStatus(1).ConsecutiveOver) +} + +func TestCooldownBlocksPromotion(t *testing.T) { + t.Parallel() + at := time.Unix(400, 0) + state := NewDetectorState() + state.SetCooldown(1, at.Add(time.Minute)) + route := testRoute(1, 1, "a", "z") + + result := Evaluate(testConfig(), state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{hotWindow(at)}, + Now: at, + }) + + require.Empty(t, result.Decisions) + require.Equal(t, 0, state.RouteStatus(1).ConsecutiveOver) + requireEvent(t, result.Events, 1, SkipReasonCooldown) +} + +func TestCooldownDropsBufferedWindowsFromCooldownPeriod(t *testing.T) { + t.Parallel() + at := time.Unix(450, 0) + state := NewDetectorState() + state.SetCooldown(1, at.Add(3*time.Minute)) + route := testRoute(1, 1, "a", "z") + cfg := testConfig() + cfg.CandidateWindows = 2 + + result := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{ + hotWindow(at.Add(time.Minute)), + hotWindow(at.Add(2 * time.Minute)), + hotWindow(at.Add(4 * time.Minute)), + }, + Now: at.Add(4 * time.Minute), + }) + + require.Empty(t, result.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) + requireEvent(t, result.Events, 1, SkipReasonCooldown) +} + +func TestCooldownDropsWindowThatStartedBeforeDeadline(t *testing.T) { + t.Parallel() + at := time.Unix(460, 0) + state := NewDetectorState() + state.SetCooldown(1, at.Add(3*time.Minute)) + route := testRoute(1, 1, "a", "z") + cfg := testConfig() + cfg.CandidateWindows = 1 + + result := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{ + testWindow(at.Add(4*time.Minute), 2*time.Minute, + testRow(1, 1, 0, 2, "a", "m", 60, 0), + testRow(1, 1, 1, 2, "m", "z", 60, 0), + ), + }, + Now: at.Add(4 * time.Minute), + }) + + require.Empty(t, result.Decisions) + require.Equal(t, 0, state.RouteStatus(1).ConsecutiveOver) + requireEvent(t, result.Events, 1, SkipReasonCooldown) +} + +func TestStateGCRemovesRetiredRoutes(t *testing.T) { + t.Parallel() + at := time.Unix(475, 0) + state := NewDetectorState() + state.SetCooldown(2, at.Add(time.Hour)) + + Evaluate(testConfig(), state, Input{ + Routes: []distribution.RouteDescriptor{testRoute(1, 1, "a", "z")}, + Windows: nil, + Now: at, + }) + + require.True(t, state.RouteStatus(2).CooldownUntil.IsZero()) +} + +func TestRouteCapUsesCycleLocalReservation(t *testing.T) { + t.Parallel() + at := time.Unix(500, 0) + routes := []distribution.RouteDescriptor{ + testRoute(1, 1, "a", "m"), + testRoute(2, 1, "m", "z"), + } + windows := []ColumnWindow{testWindow(at, time.Minute, + testRow(1, 1, 0, 2, "a", "g", 100, 0), + testRow(1, 1, 1, 2, "g", "m", 100, 0), + testRow(2, 1, 0, 2, "m", "t", 100, 0), + testRow(2, 1, 1, 2, "t", "z", 100, 0), + )} + cfg := testConfig() + cfg.MaxRoutes = 3 + cfg.MaxSplitsPerCycle = 2 + + result := Evaluate(cfg, NewDetectorState(), Input{Routes: routes, Windows: windows, Now: at}) + + require.Len(t, result.Decisions, 1) + require.Equal(t, uint64(1), result.Decisions[0].RouteID) + requireEvent(t, result.Events, 2, SkipReasonRouteCap) +} + +func TestDefaultConfigSetsRouteCap(t *testing.T) { + t.Parallel() + + require.Equal(t, defaultMaxRoutes, DefaultConfig().MaxRoutes) +} + +func TestZeroMaxRoutesUsesDefaultRouteCap(t *testing.T) { + t.Parallel() + at := time.Unix(550, 0) + routes := make([]distribution.RouteDescriptor, 0, defaultMaxRoutes) + for routeID := uint64(1); routeID <= defaultMaxRoutes; routeID++ { + routes = append(routes, testRoute(routeID, 1, "a", "z")) + } + cfg := testConfig() + cfg.MaxRoutes = 0 + + result := Evaluate(cfg, NewDetectorState(), Input{ + Routes: routes, + Windows: []ColumnWindow{hotWindow(at)}, + Now: at, + }) + + require.Empty(t, result.Decisions) + requireEvent(t, result.Events, 1, SkipReasonRouteCap) +} + +func TestMaxSplitsPerCycleBudget(t *testing.T) { + t.Parallel() + at := time.Unix(600, 0) + routes := []distribution.RouteDescriptor{ + testRoute(1, 1, "a", "m"), + testRoute(2, 1, "m", "z"), + } + windows := []ColumnWindow{testWindow(at, time.Minute, + testRow(1, 1, 0, 2, "a", "g", 100, 0), + testRow(1, 1, 1, 2, "g", "m", 100, 0), + testRow(2, 1, 0, 2, "m", "t", 100, 0), + testRow(2, 1, 1, 2, "t", "z", 100, 0), + )} + cfg := testConfig() + cfg.MaxSplitsPerCycle = 1 + + result := Evaluate(cfg, NewDetectorState(), Input{Routes: routes, Windows: windows, Now: at}) + + require.Len(t, result.Decisions, 1) + require.Equal(t, uint64(1), result.Decisions[0].RouteID) + requireEvent(t, result.Events, 2, SkipReasonBudgetExhausted) +} + +func TestNoSplitKeyForSingleBucketEvidence(t *testing.T) { + t.Parallel() + at := time.Unix(700, 0) + route := testRoute(1, 1, "a", "z") + window := testWindow(at, time.Minute, + testRow(1, 1, 0, 1, "a", "z", 100, 0), + ) + + result := Evaluate(testConfig(), NewDetectorState(), Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{window}, + Now: at, + }) + + require.Empty(t, result.Decisions) + requireEvent(t, result.Events, 1, SkipReasonNoSplitKey) +} + +func TestP50BoundarySelection(t *testing.T) { + t.Parallel() + at := time.Unix(800, 0) + tests := []struct { + name string + route distribution.RouteDescriptor + rows []keyviz.MatrixRow + wantKey []byte + wantOrigin SplitOrigin + }{ + { + name: "interior median uses hi", + route: testRoute(1, 1, "a", "e"), + rows: []keyviz.MatrixRow{ + testRow(1, 1, 0, 4, "a", "b", 10, 0), + testRow(1, 1, 1, 4, "b", "c", 90, 0), + }, + wantKey: []byte("c"), + wantOrigin: SplitOriginP50Mid, + }, + { + name: "first bucket uses interior hi", + route: testRoute(1, 1, "a", "e"), + rows: []keyviz.MatrixRow{ + testRow(1, 1, 0, 4, "a", "b", 90, 0), + testRow(1, 1, 1, 4, "b", "c", 10, 0), + }, + wantKey: []byte("b"), + wantOrigin: SplitOriginP50FirstBucketHi, + }, + { + name: "last finite bucket falls back to lo", + route: testRoute(1, 1, "a", "e"), + rows: []keyviz.MatrixRow{ + testRow(1, 1, 0, 4, "a", "b", 10, 0), + testRow(1, 1, 3, 4, "d", "e", 90, 0), + }, + wantKey: []byte("d"), + wantOrigin: SplitOriginP50LastBucketLo, + }, + { + name: "rightmost interior bucket keeps finite hi", + route: testRouteNilEnd(1, 1, "a"), + rows: []keyviz.MatrixRow{ + testRow(1, 1, 0, 4, "a", "b", 10, 0), + testRow(1, 1, 1, 4, "b", "c", 70, 0), + testRowNilEnd(1, 1, 3, 4, "d", 20, 0), + }, + wantKey: []byte("c"), + wantOrigin: SplitOriginP50Mid, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + window := testWindow(at, time.Minute, tc.rows...) + result := Evaluate(testConfig(), NewDetectorState(), Input{ + Routes: []distribution.RouteDescriptor{tc.route}, + Windows: []ColumnWindow{window}, + Now: at, + }) + require.Len(t, result.Decisions, 1) + require.Equal(t, tc.wantKey, result.Decisions[0].SplitKey) + require.Equal(t, tc.wantOrigin, result.Decisions[0].SplitOrigin) + }) + } +} + +func TestP50CoalescesDuplicateSubBucketRows(t *testing.T) { + t.Parallel() + at := time.Unix(850, 0) + route := testRoute(1, 1, "a", "z") + window := testWindow(at, time.Minute, + testRow(1, 1, 0, 2, "a", "m", 1000, 0), + testRow(1, 1, 0, 2, "a", "m", 900, 0), + testRow(1, 1, 1, 2, "m", "z", 100, 0), + ) + + result := Evaluate(testConfig(), NewDetectorState(), Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{window}, + Now: at, + }) + + require.Len(t, result.Decisions, 1) + require.Equal(t, []byte("m"), result.Decisions[0].SplitKey) + require.InDelta(t, 1900, result.Decisions[0].LeftLoad, 0.0001) + require.InDelta(t, 100, result.Decisions[0].RightLoad, 0.0001) +} + +func testConfig() Config { + return Config{ + WriteWeight: 1, + ReadWeight: 0, + ThresholdOpsMin: 50, + CandidateWindows: 1, + MaxSplitsPerCycle: 1, + } +} + +func testRoute(routeID, groupID uint64, start, end string) distribution.RouteDescriptor { + return distribution.RouteDescriptor{ + RouteID: routeID, + Start: []byte(start), + End: []byte(end), + GroupID: groupID, + State: distribution.RouteStateActive, + } +} + +func testRouteNilEnd(routeID, groupID uint64, start string) distribution.RouteDescriptor { + return distribution.RouteDescriptor{ + RouteID: routeID, + Start: []byte(start), + GroupID: groupID, + State: distribution.RouteStateActive, + } +} + +func testWindow(at time.Time, duration time.Duration, rows ...keyviz.MatrixRow) ColumnWindow { + return ColumnWindow{ + Column: keyviz.MatrixColumn{ + At: at, + Rows: rows, + }, + Duration: duration, + } +} + +func hotWindow(at time.Time) ColumnWindow { + return testWindow(at, time.Minute, + testRow(1, 1, 0, 2, "a", "m", 60, 0), + testRow(1, 1, 1, 2, "m", "z", 60, 0), + ) +} + +func coldWindow(at time.Time) ColumnWindow { + return testWindow(at, time.Minute, + testRow(1, 1, 0, 2, "a", "m", 10, 0), + testRow(1, 1, 1, 2, "m", "z", 10, 0), + ) +} + +func testRow(routeID, groupID uint64, subBucket, subBucketCount int, start, end string, writes, reads uint64) keyviz.MatrixRow { + return keyviz.MatrixRow{ + RouteID: routeID, + RaftGroupID: groupID, + Start: []byte(start), + End: []byte(end), + SubBucket: subBucket, + SubBucketCount: subBucketCount, + Writes: writes, + Reads: reads, + } +} + +func testRowNilEnd(routeID, groupID uint64, subBucket, subBucketCount int, start string, writes, reads uint64) keyviz.MatrixRow { + return keyviz.MatrixRow{ + RouteID: routeID, + RaftGroupID: groupID, + Start: []byte(start), + SubBucket: subBucket, + SubBucketCount: subBucketCount, + Writes: writes, + Reads: reads, + } +} + +func requireEvent(t *testing.T, events []Event, routeID uint64, reason SkipReason) { + t.Helper() + for _, event := range events { + if event.RouteID == routeID && event.Reason == reason { + return + } + } + t.Fatalf("missing event route_id=%d reason=%s in %#v", routeID, reason, events) +} diff --git a/docs/design/2026_06_11_partial_hotspot_split_milestone3_automation.md b/docs/design/2026_06_11_partial_hotspot_split_milestone3_automation.md index 89c2e85a4..6c9243715 100644 --- a/docs/design/2026_06_11_partial_hotspot_split_milestone3_automation.md +++ b/docs/design/2026_06_11_partial_hotspot_split_milestone3_automation.md @@ -2,11 +2,12 @@ Status: Partial - M3-PR1a is implemented: the dead `RecordAccess` / `splitRange` / threshold constructor path is removed, and the route-catalog -decoder now keeps v1 strict while accepting forward-version tails. M3-PR1b+ -remain open. +decoder now keeps v1 strict while accepting forward-version tails. M3-PR2a adds +the pure autosplit detector core. M3-PR1b, M3-PR2b+, and scheduler wiring remain +open. Author: bootjp Date: 2026-06-11 -Updated: 2026-07-07 +Updated: 2026-07-18 Parent: [2026_02_18_partial_hotspot_shard_split.md](2026_02_18_partial_hotspot_shard_split.md) §12 "Milestone 3: Automation" (1. Access aggregation, 2. Hotspot detector, 3. Auto-split scheduler with cooldown/hysteresis). Sibling: [PR #945: Hotspot split Milestone 2 migration design](https://github.com/bootjp/elastickv/pull/945) (M2 — migration plane). M3 composes with M2 but does **not** depend on it for first value (see §2, §6). @@ -362,7 +363,7 @@ Budget and cooldown treatment: - `--autoSplit` (bool, default `false`). When false, none of the detector/scheduler goroutines start; behavior is identical to today. When true, the default-group leader runs the detector + scheduler loop, and the keyviz sampler is constructed and flushed if it is not already. - **Sampler-wiring condition.** The sampler is built when `keyvizEnabled || autoSplit` (today it is built only on `keyvizEnabled`; see `main.go:2005` `newKeyvizSampler` and the analogous wiring in `cmd/server/demo.go`). M3-PR3's flag-wiring change touches both `main.go` and `cmd/server/demo.go`. When auto-split forces the sampler on and the operator **did not explicitly set** `--keyvizKeyBucketsPerRoute` (detected via `flag.Visit` after `flag.Parse()`, see §3.1 — plain `flag.Int` cannot distinguish an omitted flag from an explicit `1`), auto-split raises it to `autoSplitDefaultBuckets` (default 16) so split-key evidence exists by default (§3.1 table). An explicit `--keyvizKeyBucketsPerRoute 1` is honored as-is (operator opts into route-granular-only → §5.4.3 decline). `--autoSplitDefaultBuckets` is itself clamped to `[2, MaxKeyBucketsPerRoute]` (256, `keyviz/sampler.go:115`) and validated at startup — a value below 2 would defeat the purpose, and above the cap is rejected like the other keyviz bucket flags. -- Tuning flags (all with parent-doc-derived defaults): `--autoSplitWriteWeight` (4), `--autoSplitReadWeight` (1), `--autoSplitThreshold` (50000 ops/min), `--autoSplitCandidateWindows` (3 — number of **consecutive committed flush columns** whose **per-column** score must clear the threshold to promote, §5.2; it sets the hysteresis length, not the per-column rate), `--autoSplitCooldown` (10m), `--autoSplitMaxRoutes`, `--autoSplitMaxPerCycle` (1 — bounds **logical split decisions** per cycle, not raw `SplitRange` RPCs; a compound single-key isolation is one decision / two RPCs, §6.3), `--autoSplitEvalInterval` (evaluation cycle period, default = `--keyvizStep`), `--autoSplitDefaultBuckets` (sub-range buckets auto-split implies when keyviz bucketing is left at the default, default 16). +- Tuning flags (all with parent-doc-derived defaults): `--autoSplitWriteWeight` (4), `--autoSplitReadWeight` (1), `--autoSplitThreshold` (50000 ops/min), `--autoSplitCandidateWindows` (3 — number of **consecutive committed flush columns** whose **per-column** score must clear the threshold to promote, §5.2; it sets the hysteresis length, not the per-column rate), `--autoSplitCooldown` (10m), `--autoSplitMaxRoutes` (1024), `--autoSplitMaxPerCycle` (1 — bounds **logical split decisions** per cycle, not raw `SplitRange` RPCs; a compound single-key isolation is one decision / two RPCs, §6.3), `--autoSplitEvalInterval` (evaluation cycle period, default = `--keyvizStep`), `--autoSplitDefaultBuckets` (sub-range buckets auto-split implies when keyviz bucketing is left at the default, default 16). ### 7.2 Runtime kill switch @@ -545,7 +546,8 @@ The fence that makes this safe lives on `SplitAtHLC` (a leader-fenced, Raft-comm |---|---|---|---| | **M3-PR1a** (decoder relaxation, ship **first**) | **Implemented.** Retired engine `RecordAccess`/`splitRange` threshold path (§3.4); confirmed keyviz is the sole signal; introduced `catalogRouteCodecVersionMin` (`= 1`), accepted version byte `>= catalogRouteCodecVersionMin`, and made trailing bytes version-conditional: **current `version == 1` remains strict**, while `version > catalogRouteCodecVersion` drains unknown forward-version tail bytes (§7.7c). **No `SplitAtHLC` struct field, no encoder change, no helper change; `catalogRouteCodecVersion` stays `1` and the encoder still writes v1.** | `distribution/engine_test.go` updated (dead-code removal); codec matrix cases authored before the encoder exists — v1 still round-trips, v1 plus extra trailing byte is rejected, hand-built v2 records with finite and nil `End` drain the 8-byte tail, and version `< min` is rejected. | Shipped. | | **M3-PR1b** (v2 emission, deploy **after** PR1a is cluster-wide) | **Add the durable `RouteDescriptor.SplitAtHLC` field**; bump the encoder to `catalogRouteCodecVersion = 2` and append the 8-byte big-endian tail on **both** End paths (§7.7c if/else); add the version-keyed `version >= 2` tail read in the decoder; update `CloneRouteDescriptor` and `routeDescriptorEqual` to include `SplitAtHLC` (§7.7b, Issue 2); add a coordinator-side `SplitRange` catalog transaction builder that allocates the fenced commit ts on the committing default-group leader with `nextTxnTSAfter(snapshot.ReadTS)`, stamps child descriptors with that exact ts, commits at that ts, and returns it; add the `hlcPhysicalMs` conversion helper (§7.7d); add `autosplit_*` metric scaffolding (no detector yet). | **codec round-trip matrix cases 1, 1a, 2 (§7.7c)** — v2 round-trip preserves `SplitAtHLC` (incl. `=0` and incl. at least one `End == nil` route, case 1a, guarding the encoder early-return-on-`End == nil` path); re-run cases 4/4a now against the actual v2 encoder output; **case 7 (truncated tail rejected) — only the PR1b `version >= 2` exact-8-byte read can detect a `< 8`-byte tail (`io.ErrUnexpectedEOF → ErrCatalogInvalidRouteRecord`), so it is authored here, not in PR1a (§7.7c, codex P2)**; helper tests — clone preserves `SplitAtHLC` and `routeDescriptorEqual` distinguishes routes differing only in `SplitAtHLC`; fenced-ts tests — the committing leader's allocator retries after `Observe(ReadTS)` when a first `NextFenced()` would be `<= ReadTS`, forwarded SplitRange requests allocate on the new leader rather than accepting an adapter-supplied `CommitTS`, and `ErrCeilingExpired` fails closed; `hlcPhysicalMs` unit test incl. the `1<<16`-per-ms regression case; metric registration test. | Yes — additive schema field with forward/back-compat decode + metric names; no behavior change. **Deployment-ordered after M3-PR1a** (the only operational gate; see §7.7c failure mode if violated). Carries the deploy-after-PR1a rollout note in its PR description. | -| **M3-PR2** | Aggregation reader (leader-local consumption of `MemSampler.Snapshot`, §4, including **per-column grouping of all non-aggregate rows by `(RouteID, RaftGroupID)` before scoring**) + the pure detector: `RouteStateActive`-only candidacy precondition (§5.0), scoring, hysteresis, cooldown, compound single-key isolation (§5.4.2), split-key selection, guardrails (§5), **state-map GC reconciliation against the live catalog each cycle (§7.7a)**, and the default-group plus shard-group leadership-start processed-column watermark/straddling-column skip (§4.2/§7.6). No scheduler wiring — detector emits decisions to a sink that is a no-op/log in this PR. | Table-driven unit tests for route-row aggregation, candidacy precondition (non-active states never promote)/scoring/hysteresis/cooldown/compound-split cooldown exemption/split-key/guardrails; missing sampler rows synthesize zero **only for candidate hysteresis reset** for every live route in every committed column and are not target-admission evidence (§6.2); pre-election sampler history and the first column whose window opened before default-group or shard-group leadership start are ignored until a fully post-start committed column arrives; Top-K share uses the lower-bound `max(0, Count - ceil(SampledN/Capacity)) × SampleRate / column_i.write_ops`, requires non-degraded snapshots, and is used only when the Top-K sketch is column-attached or has matching shared boundaries; timestamp-only Top-K matching, degraded snapshots, and upper-bound-only Top-K decisions are red controls; rightmost-route single-key lower-edge and p50 cases treat `End == nil` as `+inf`; p50 uses finite `hi` for interior buckets and `lo` only for the actual last bucket; state-map shrinks after a parent `RouteID` retires (the detector is a pure function, §5). | Yes — detector runs and logs would-be decisions; nothing splits. Safe to ship "observe-only." | +| **M3-PR2a** | **Pure autosplit detector core.** Add `distribution/autosplit` with route-row aggregation by `(RouteID, RaftGroupID)`, `RouteStateActive`-only candidacy, per-window weighted ops/min scoring, consecutive-window hysteresis, cooldown state, p50 sub-bucket split-key selection, aggregate-row skips, route cap accounting, per-cycle split budget, and state-map GC reconciliation against the live catalog each evaluation (§7.7a). No sampler reader, no scheduler wiring, no Top-K isolation, and no runtime side effects. | Table-driven unit tests for row aggregation, active vs non-active candidacy reset, consecutive-window promotion/reset, cooldown, state-map shrink after parent retirement, no-evidence decline, route cap reservation, per-cycle budget, and p50 boundary selection including finite and unbounded right edges. | Yes — library-only detector core; no runtime behavior change. | +| **M3-PR2b** | Aggregation reader (leader-local consumption of `MemSampler.Snapshot`, §4, including committed-window boundary handling) + the remaining observe-only detector integration: missing-row zero synthesis for hysteresis reset, compound single-key isolation (§5.4.2), Top-K alignment/error gates, and the default-group plus shard-group leadership-start processed-column watermark/straddling-column skip (§4.2/§7.6). No scheduler wiring — detector emits decisions to a no-op/log sink in this PR. | Unit tests for committed-window derivation, missing sampler rows, leadership watermarks, Top-K lower-bound/share/absolute-floor gates, compound split cooldown exemption, unknown-load target exclusion, and observe-only sink output. | Yes — detector runs and logs would-be decisions; nothing splits. Safe to ship "observe-only." | | **M3-PR3** | Scheduler wiring to `SplitRange` (§6, incl. compound single-key isolation §6.4) + `--autoSplit` flag + sampler-wiring condition `keyvizEnabled \|\| autoSplit` and `autoSplitDefaultBuckets` implied bucketing in `main.go` and `cmd/server/demo.go` (§7.1) + runtime kill switch + slog + `SplitAtHLC`-seeded cooldown reconstruction on election (§7.7b). Same-group target only (cross-group target hook present but disabled). | Integration: detector→`SplitRange` end-to-end in the 3-node demo (`cmd/server/demo.go`); kill-switch + leader-change reset + seeded-cooldown-from-`SplitAtHLC` tests; failed `SplitRange` does **not** seed cooldown; a same-leader compound partial stores the pending second-call details and retries the final split before normal candidates even with no new sampler column, while a restart/leader change reconstructs normal cooldown for that intermediate child. | Yes — completes standalone auto-split. | | **M3-PR4** (post-M2) | Enable least-loaded `target_group_id` selection on the existing scheduler hook once M2's cross-group `SplitRange` lands (§6.2); exclude in-flight `SplitJob` routes. **Compound single-key isolation stays same-group-only** — both its `SplitRange` calls always pass `target_group_id = 0`; cross-group selection applies only when the right child carries the hot load and target knowledge covers every live route in the target group (§6.2/§6.4, codex P2). The singleton `{H}` is non-targetable after isolation until a future `MoveRange`-style primitive exists. | Integration: cross-group auto-split against M2 migrator; target-selection unit tests; **a test that a compound isolation under `--autoSplitCrossGroup` still issues both calls same-group and only a non-singleton final child is migrated in a later independent decision (never call 1's intermediate child, never the `{H}` singleton)**; p50 left-hot and `isolation_single_lower_edge` decisions force same-group; target admission with authoritative fresh rows for every live route admits the group, while a non-authoritative fresh row or any missing live-route row skips as `load_unknown`; zero-live-route groups remain admissible as known idle. | Depends on M2; the M3 scheduler interface is unchanged. | | **M3-PR5** | Lifecycle: this doc is renamed `*_partial_*` after M3-PR1a; → `*_implemented_*` after M3-PR3 (standalone) with M3-PR4 tracked as the cross-group follow-on. | — | Doc-only. |