diff --git a/distribution/autosplit/detector.go b/distribution/autosplit/detector.go index cf1c6f12a..32af77991 100644 --- a/distribution/autosplit/detector.go +++ b/distribution/autosplit/detector.go @@ -79,9 +79,10 @@ type RouteLoad struct { // 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. +// Runtime integration passes only committed windows with a proven duration. +// keyviz.MatrixColumn.WindowStart is authoritative when present; legacy +// in-memory rows may be accepted only when the previous contiguous column proves +// the lower boundary. type ColumnWindow struct { Column keyviz.MatrixColumn Duration time.Duration diff --git a/distribution/autosplit/sampler_reader.go b/distribution/autosplit/sampler_reader.go new file mode 100644 index 000000000..84460033c --- /dev/null +++ b/distribution/autosplit/sampler_reader.go @@ -0,0 +1,142 @@ +package autosplit + +import ( + "sort" + "time" + + "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/keyviz" +) + +// SnapshotSource is the narrow keyviz snapshot surface the observe-only +// autosplit reader needs. +type SnapshotSource interface { + Snapshot(from, to time.Time) []keyviz.MatrixColumn +} + +// SnapshotReadConfig controls one off-path autosplit sampler read. +type SnapshotReadConfig struct { + Step time.Duration + CandidateWindows int + LastProcessedAt time.Time + Now time.Time +} + +// SnapshotReadResult is the committed keyviz material consumed by the detector. +type SnapshotReadResult struct { + Windows []ColumnWindow + NewestCommittedAt time.Time + SnapshotFrom time.Time + SnapshotTo time.Time + SkippedInvalid int +} + +// ObserveSnapshot reads committed keyviz windows and runs the pure detector. +// Callers may log Result.Decisions in observe-only mode; this helper never +// mutates the route catalog and never calls SplitRange. +func ObserveSnapshot( + cfg Config, + state *DetectorState, + routes []distribution.RouteDescriptor, + source SnapshotSource, + readCfg SnapshotReadConfig, +) (Result, SnapshotReadResult) { + read := ReadCommittedWindows(source, readCfg) + result := Evaluate(cfg, state, Input{ + Routes: routes, + Windows: read.Windows, + Now: read.SnapshotTo, + }) + return result, read +} + +// ReadCommittedWindows converts a time-range keyviz snapshot into detector +// windows, excluding columns that have already been processed and columns whose +// committed lower boundary is not proven. +func ReadCommittedWindows(source SnapshotSource, cfg SnapshotReadConfig) SnapshotReadResult { + if source == nil { + return SnapshotReadResult{} + } + step := cfg.Step + if step <= 0 { + step = keyviz.DefaultStep + } + candidateWindows := cfg.CandidateWindows + if candidateWindows <= 0 { + candidateWindows = defaultCandidateWindows + } + now := cfg.Now + if now.IsZero() { + now = time.Now() + } + + from := now.Add(-time.Duration(candidateWindows+1) * step) + if !cfg.LastProcessedAt.IsZero() { + from = cfg.LastProcessedAt.Add(-step) + } + cols := source.Snapshot(from, now) + windows, newest, skipped := CommittedWindowsFromColumns(cols, cfg.LastProcessedAt) + return SnapshotReadResult{ + Windows: windows, + NewestCommittedAt: newest, + SnapshotFrom: from, + SnapshotTo: now, + SkippedInvalid: skipped, + } +} + +// CommittedWindowsFromColumns normalizes raw keyviz columns into detector +// windows. WindowStart is authoritative when present. For older in-memory +// columns without WindowStart, the immediately previous column boundary is the +// only accepted fallback. Columns whose lower boundary is not proven are +// returned as zero-duration reset sentinels so the detector clears stale +// confidence instead of carrying it across an unknown interval. +func CommittedWindowsFromColumns(cols []keyviz.MatrixColumn, lastProcessedAt time.Time) ([]ColumnWindow, time.Time, int) { + ordered := append([]keyviz.MatrixColumn(nil), cols...) + sort.SliceStable(ordered, func(i, j int) bool { + return ordered[i].At.Before(ordered[j].At) + }) + + var newest time.Time + windows := make([]ColumnWindow, 0, len(ordered)) + skipped := 0 + lastBoundary := lastProcessedAt + for i, col := range ordered { + if col.At.After(newest) { + newest = col.At + } + if !col.At.After(lastProcessedAt) { + continue + } + start := committedWindowStart(ordered, i) + if start.IsZero() || !start.Before(col.At) { + skipped++ + windows = append(windows, ColumnWindow{Column: keyviz.MatrixColumn{At: col.At}}) + lastBoundary = col.At + continue + } + if needsBoundaryReset(lastBoundary, start) { + windows = append(windows, ColumnWindow{Column: keyviz.MatrixColumn{At: col.At}}) + lastBoundary = col.At + continue + } + windows = append(windows, ColumnWindow{ + Column: col, + Duration: col.At.Sub(start), + }) + lastBoundary = col.At + } + return windows, newest, skipped +} + +func committedWindowStart(cols []keyviz.MatrixColumn, i int) time.Time { + start := cols[i].WindowStart + if start.IsZero() && i > 0 && cols[i-1].At.Before(cols[i].At) { + start = cols[i-1].At + } + return start +} + +func needsBoundaryReset(lastBoundary, start time.Time) bool { + return !lastBoundary.IsZero() && !start.Equal(lastBoundary) +} diff --git a/distribution/autosplit/sampler_reader_test.go b/distribution/autosplit/sampler_reader_test.go new file mode 100644 index 000000000..c8acf9b2b --- /dev/null +++ b/distribution/autosplit/sampler_reader_test.go @@ -0,0 +1,268 @@ +package autosplit + +import ( + "testing" + "time" + + "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/keyviz" + "github.com/stretchr/testify/require" +) + +type fakeSnapshotSource struct { + cols []keyviz.MatrixColumn + from time.Time + to time.Time +} + +func (f *fakeSnapshotSource) Snapshot(from, to time.Time) []keyviz.MatrixColumn { + f.from = from + f.to = to + out := make([]keyviz.MatrixColumn, 0, len(f.cols)) + for _, col := range f.cols { + if !from.IsZero() && col.At.Before(from) { + continue + } + if !to.IsZero() && !col.At.Before(to) { + continue + } + out = append(out, col) + } + return out +} + +func TestCommittedWindowsPreferExplicitWindowStart(t *testing.T) { + t.Parallel() + at := time.Unix(1_700_000_100, 0) + cols := []keyviz.MatrixColumn{ + readerColumn(at.Add(time.Minute), at, nil), + readerColumn(at.Add(3*time.Minute), at.Add(time.Minute), nil), + } + + windows, newest, skipped := CommittedWindowsFromColumns(cols, time.Time{}) + + require.Equal(t, at.Add(3*time.Minute), newest) + require.Zero(t, skipped) + require.Len(t, windows, 2) + require.Equal(t, time.Minute, windows[0].Duration) + require.Equal(t, 2*time.Minute, windows[1].Duration) +} + +func TestCommittedWindowsFallbackRequiresPreviousBoundary(t *testing.T) { + t.Parallel() + at := time.Unix(1_700_000_200, 0) + cols := []keyviz.MatrixColumn{ + {At: at, Rows: []keyviz.MatrixRow{readerRow(20)}}, + {At: at.Add(time.Minute), Rows: []keyviz.MatrixRow{readerRow(20)}}, + } + + windows, newest, skipped := CommittedWindowsFromColumns(cols, time.Time{}) + + require.Equal(t, at.Add(time.Minute), newest) + require.Equal(t, 1, skipped) + require.Len(t, windows, 2) + require.Equal(t, at, windows[0].Column.At) + require.Zero(t, windows[0].Duration) + require.Equal(t, at.Add(time.Minute), windows[1].Column.At) + require.Equal(t, time.Minute, windows[1].Duration) +} + +func TestReadCommittedWindowsFetchesFromLastProcessedMinusStep(t *testing.T) { + t.Parallel() + at := time.Unix(1_700_000_300, 0) + source := &fakeSnapshotSource{cols: []keyviz.MatrixColumn{ + readerColumn(at, at.Add(-time.Minute), nil), + readerColumn(at.Add(time.Minute), at, nil), + readerColumn(at.Add(2*time.Minute), at.Add(time.Minute), nil), + }} + + result := ReadCommittedWindows(source, SnapshotReadConfig{ + Step: time.Minute, + CandidateWindows: 3, + LastProcessedAt: at.Add(time.Minute), + Now: at.Add(3 * time.Minute), + }) + + require.Equal(t, at, source.from) + require.Equal(t, at.Add(3*time.Minute), source.to) + require.Equal(t, at.Add(2*time.Minute), result.NewestCommittedAt) + require.Len(t, result.Windows, 1) + require.Equal(t, at.Add(2*time.Minute), result.Windows[0].Column.At) +} + +func TestObserveSnapshotProcessesSilentGapAsReset(t *testing.T) { + t.Parallel() + at := time.Unix(1_700_000_400, 0) + source := &fakeSnapshotSource{cols: []keyviz.MatrixColumn{ + readerColumn(at, at.Add(-time.Minute), []keyviz.MatrixRow{ + readerRow(60), + }), + readerColumn(at.Add(time.Minute), at, nil), + readerColumn(at.Add(2*time.Minute), at.Add(time.Minute), []keyviz.MatrixRow{ + readerRow(60), + }), + }} + state := NewDetectorState() + cfg := testConfig() + cfg.CandidateWindows = 2 + + result, read := ObserveSnapshot(cfg, state, []distribution.RouteDescriptor{ + testRoute(1, 1, "a", "z"), + }, source, SnapshotReadConfig{ + Step: time.Minute, + CandidateWindows: 2, + Now: at.Add(3 * time.Minute), + }) + + require.Len(t, read.Windows, 3) + require.Empty(t, result.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) + require.Equal(t, at.Add(2*time.Minute), state.RouteStatus(1).LastProcessedAt) +} + +func TestObserveSnapshotResetsWhenHistoryNoLongerReachesWatermark(t *testing.T) { + t.Parallel() + at := time.Unix(1_700_000_500, 0) + route := testRoute(1, 1, "a", "z") + state := NewDetectorState() + cfg := testConfig() + cfg.CandidateWindows = 3 + warmup := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{ + hotWindow(at.Add(-time.Minute)), + hotWindow(at), + }, + Now: at, + }) + require.Empty(t, warmup.Decisions) + require.Equal(t, 2, state.RouteStatus(1).ConsecutiveOver) + + source := &fakeSnapshotSource{cols: []keyviz.MatrixColumn{ + readerColumn(at.Add(10*time.Minute), at.Add(9*time.Minute), []keyviz.MatrixRow{ + readerRow(60), + }), + readerColumn(at.Add(11*time.Minute), at.Add(10*time.Minute), []keyviz.MatrixRow{ + readerRow(60), + }), + }} + + result, read := ObserveSnapshot(cfg, state, []distribution.RouteDescriptor{route}, source, SnapshotReadConfig{ + Step: time.Minute, + CandidateWindows: 3, + LastProcessedAt: at, + Now: at.Add(12 * time.Minute), + }) + + require.Len(t, read.Windows, 2) + require.Zero(t, read.Windows[0].Duration) + require.Equal(t, at.Add(10*time.Minute), read.Windows[0].Column.At) + require.Empty(t, result.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) + require.Equal(t, at.Add(11*time.Minute), state.RouteStatus(1).LastProcessedAt) + requireEvent(t, result.Events, 0, SkipReasonInvalidWindow) +} + +func TestObserveSnapshotResetsWhenWindowOverlapsWatermark(t *testing.T) { + t.Parallel() + at := time.Unix(1_700_000_550, 0) + route := testRoute(1, 1, "a", "z") + state := NewDetectorState() + cfg := testConfig() + cfg.CandidateWindows = 3 + warmup := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{ + hotWindow(at.Add(-time.Minute)), + hotWindow(at), + }, + Now: at, + }) + require.Empty(t, warmup.Decisions) + require.Equal(t, 2, state.RouteStatus(1).ConsecutiveOver) + + source := &fakeSnapshotSource{cols: []keyviz.MatrixColumn{ + readerColumn(at.Add(2*time.Minute), at.Add(-time.Minute), []keyviz.MatrixRow{ + readerRow(60), + }), + readerColumn(at.Add(3*time.Minute), at.Add(2*time.Minute), []keyviz.MatrixRow{ + readerRow(60), + }), + }} + + result, read := ObserveSnapshot(cfg, state, []distribution.RouteDescriptor{route}, source, SnapshotReadConfig{ + Step: time.Minute, + CandidateWindows: 3, + LastProcessedAt: at, + Now: at.Add(4 * time.Minute), + }) + + require.Len(t, read.Windows, 2) + require.Zero(t, read.Windows[0].Duration) + require.Equal(t, at.Add(2*time.Minute), read.Windows[0].Column.At) + require.Empty(t, result.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) + require.Equal(t, at.Add(3*time.Minute), state.RouteStatus(1).LastProcessedAt) + requireEvent(t, result.Events, 0, SkipReasonInvalidWindow) +} + +func TestObserveSnapshotResetsOnInvalidCommittedColumn(t *testing.T) { + t.Parallel() + at := time.Unix(1_700_000_600, 0) + route := testRoute(1, 1, "a", "z") + state := NewDetectorState() + cfg := testConfig() + cfg.CandidateWindows = 3 + warmup := Evaluate(cfg, state, Input{ + Routes: []distribution.RouteDescriptor{route}, + Windows: []ColumnWindow{ + hotWindow(at.Add(-time.Minute)), + hotWindow(at), + }, + Now: at, + }) + require.Empty(t, warmup.Decisions) + require.Equal(t, 2, state.RouteStatus(1).ConsecutiveOver) + + source := &fakeSnapshotSource{cols: []keyviz.MatrixColumn{ + {At: at.Add(time.Minute), Rows: []keyviz.MatrixRow{readerRow(60)}}, + readerColumn(at.Add(2*time.Minute), at.Add(time.Minute), []keyviz.MatrixRow{ + readerRow(60), + }), + }} + + result, read := ObserveSnapshot(cfg, state, []distribution.RouteDescriptor{route}, source, SnapshotReadConfig{ + Step: time.Minute, + CandidateWindows: 3, + LastProcessedAt: at, + Now: at.Add(3 * time.Minute), + }) + + require.Equal(t, 1, read.SkippedInvalid) + require.Len(t, read.Windows, 2) + require.Zero(t, read.Windows[0].Duration) + require.Empty(t, result.Decisions) + require.Equal(t, 1, state.RouteStatus(1).ConsecutiveOver) + require.Equal(t, at.Add(2*time.Minute), state.RouteStatus(1).LastProcessedAt) + requireEvent(t, result.Events, 0, SkipReasonInvalidWindow) +} + +func readerColumn(at, start time.Time, rows []keyviz.MatrixRow) keyviz.MatrixColumn { + return keyviz.MatrixColumn{ + WindowStart: start, + At: at, + Rows: rows, + } +} + +func readerRow(writes uint64) keyviz.MatrixRow { + return keyviz.MatrixRow{ + RouteID: 1, + RaftGroupID: 1, + Start: []byte("a"), + End: []byte("m"), + SubBucket: 0, + SubBucketCount: 2, + Writes: writes, + } +} 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 6c9243715..fae7d93e2 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 @@ -3,11 +3,13 @@ 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-PR2a adds -the pure autosplit detector core. M3-PR1b, M3-PR2b+, and scheduler wiring remain +the pure autosplit detector core. M3-PR2b-a adds the committed-window +`MemSampler.Snapshot` reader and observe-only detector bridge. M3-PR1b, the +remaining M3-PR2b Top-K / leadership-watermark work, and scheduler wiring remain open. Author: bootjp Date: 2026-06-11 -Updated: 2026-07-18 +Updated: 2026-07-22 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). @@ -547,7 +549,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-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-PR2b-a** | **Committed-window reader and observe-only detector bridge.** Add `MatrixColumn.WindowStart` at `MemSampler.Flush`, preserve it through `Snapshot` deep copies, and add a `distribution/autosplit` reader that consumes `MemSampler.Snapshot`, fetches from `lastProcessedAt - Step`, excludes already-processed columns, accepts only proven committed window boundaries, falls back only to the immediately previous column for legacy in-memory rows, and runs the pure detector without calling `SplitRange`. | `keyviz` tests for persisted/deep-copied `WindowStart`; autosplit tests for committed-window derivation, legacy previous-boundary fallback, `lastProcessedAt - Step` fetches, and missing sampler rows resetting confidence in observe-only evaluation. | Yes — no runtime scheduler, no catalog mutation, no split RPC. | +| **M3-PR2b** | Remaining observe-only detector integration: 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 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. | diff --git a/keyviz/ring_buffer.go b/keyviz/ring_buffer.go index a3dcc6162..8e81b3c23 100644 --- a/keyviz/ring_buffer.go +++ b/keyviz/ring_buffer.go @@ -83,7 +83,7 @@ func cloneColumn(col MatrixColumn) MatrixColumn { rows[i].MemberRoutes = append([]uint64(nil), row.MemberRoutes...) } } - return MatrixColumn{At: col.At, Rows: rows} + return MatrixColumn{WindowStart: col.WindowStart, At: col.At, Rows: rows} } // snapshotOrdered returns a chronologically ordered (oldest first) diff --git a/keyviz/sampler.go b/keyviz/sampler.go index 3bf6a70f9..20aa01a93 100644 --- a/keyviz/sampler.go +++ b/keyviz/sampler.go @@ -212,10 +212,19 @@ type MemSampler struct { // Held only by non-hot-path callers. routesMu sync.Mutex + // flushMu serialises the full Flush flow from timestamp capture + // through history append. That keeps column order and window + // boundaries aligned when callers accidentally run concurrent + // flushers. + flushMu sync.Mutex + // historyMu guards the ring buffer. Reads (Snapshot) and writes // (Flush) acquire it; Observe never touches it. historyMu sync.Mutex history *ringBuffer + // lastFlushAt is the lower boundary for the next committed matrix + // column. Guarded by historyMu with history. + lastFlushAt time.Time // retiredSlots holds slots that RemoveRoute removed from the live // table. Each entry is drained for `remaining` Flushes — a grace @@ -425,10 +434,13 @@ func (s *routeSlot) snapshotMeta() (start, end []byte, aggregate bool, members [ return } -// MatrixColumn is one slice of the heatmap at a single flush time. +// MatrixColumn is one committed heatmap window. type MatrixColumn struct { - At time.Time - Rows []MatrixRow + // WindowStart and At delimit the committed half-open sampler + // interval. Rows contain counters drained for (WindowStart, At]. + WindowStart time.Time + At time.Time + Rows []MatrixRow } // MatrixRow is a single route or virtual bucket's counter snapshot @@ -509,10 +521,11 @@ func NewMemSampler(opts MemSamplerOptions) *MemSampler { now = time.Now } s := &MemSampler{ - opts: opts, - now: now, - history: newRingBuffer(opts.HistoryColumns), - groupTerms: map[uint64]uint64{}, + opts: opts, + now: now, + history: newRingBuffer(opts.HistoryColumns), + lastFlushAt: now(), + groupTerms: map[uint64]uint64{}, } s.table.Store(newEmptyRouteTable()) if opts.HotKeysEnabled { @@ -1205,6 +1218,9 @@ func (s *MemSampler) Flush() { if s == nil { return } + s.flushMu.Lock() + defer s.flushMu.Unlock() + col := MatrixColumn{At: s.now()} // Snapshot the per-group leader-term map once at the top of // Flush so every row in this column sees a consistent view, even @@ -1229,6 +1245,11 @@ func (s *MemSampler) Flush() { }) s.historyMu.Lock() + col.WindowStart = s.lastFlushAt + if !col.WindowStart.Before(col.At) { + col.WindowStart = col.At.Add(-s.opts.Step) + } + s.lastFlushAt = col.At s.history.Push(col) s.historyMu.Unlock() } diff --git a/keyviz/sampler_test.go b/keyviz/sampler_test.go index 674c757dc..127d46635 100644 --- a/keyviz/sampler_test.go +++ b/keyviz/sampler_test.go @@ -78,6 +78,40 @@ func TestObserveAndFlushBasic(t *testing.T) { } } +func TestFlushStampsWindowStart(t *testing.T) { + t.Parallel() + s, clk := newTestSampler(t, MemSamplerOptions{Step: time.Second, HistoryColumns: 4}) + start := clk.Now() + if !s.RegisterRoute(1, []byte("a"), []byte("b"), 0) { + t.Fatal("RegisterRoute(1) returned false") + } + + clk.Advance(time.Second) + s.Observe(1, []byte("a"), OpWrite, 0, LabelLegacy) + s.Flush() + + clk.Advance(2 * time.Second) + s.Observe(1, []byte("a"), OpWrite, 0, LabelLegacy) + s.Flush() + + cols := s.Snapshot(time.Time{}, time.Time{}) + if len(cols) != 2 { + t.Fatalf("got %d columns, want 2", len(cols)) + } + if !cols[0].WindowStart.Equal(start) { + t.Fatalf("first WindowStart = %v, want %v", cols[0].WindowStart, start) + } + if !cols[1].WindowStart.Equal(cols[0].At) { + t.Fatalf("second WindowStart = %v, want previous At %v", cols[1].WindowStart, cols[0].At) + } + + cols[0].WindowStart = time.Time{} + fresh := s.Snapshot(time.Time{}, time.Time{}) + if !fresh[0].WindowStart.Equal(start) { + t.Fatalf("snapshot did not deep-copy WindowStart: got %v want %v", fresh[0].WindowStart, start) + } +} + // TestNoCountsLostAcrossFlush asserts the SwapUint64 flush protocol // doesn't drop counts even when many goroutines hammer Observe across // the flush boundary. Repeats the exercise N times to flush out