From 0256d7ce08be0bb260e2e38fe04d904793f6bae4 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 4 Sep 2026 05:44:23 -0600 Subject: [PATCH 1/3] fix(precompute): stage config at window boundaries --- asap-precompute-go/precompute.go | 72 ++++++++++++++++++++++++-------- asap-precompute-go/scope_test.go | 27 +++++++----- asap-precompute-go/window.go | 14 +++++++ 3 files changed, 85 insertions(+), 28 deletions(-) diff --git a/asap-precompute-go/precompute.go b/asap-precompute-go/precompute.go index ff4c229a..18a7890e 100644 --- a/asap-precompute-go/precompute.go +++ b/asap-precompute-go/precompute.go @@ -283,6 +283,7 @@ func (p *precompute) ResetDeltaBase() { // No global mutex around the Precompute itself. type precompute struct { cfg atomic.Pointer[PrecomputeConfig] + pendingCfg atomic.Pointer[PrecomputeConfig] sketchFactory SketchFactory observer SketchObserver window *windowState @@ -319,8 +320,8 @@ func New(initialCfg *PrecomputeConfig, sketchFactory SketchFactory, observer Ske stats: NewPrecomputeStats(), } if initialCfg != nil { - cfgCopy := *initialCfg - p.cfg.Store(&cfgCopy) + cfgCopy := clonePrecomputeConfig(initialCfg) + p.cfg.Store(cfgCopy) p.sketchType = initialCfg.SketchType } return p @@ -494,7 +495,11 @@ func (p *precompute) Tick(nowMs uint64) []*SketchEnvelope { return nil } closed, rng := p.window.rotate(nowMs, cfg) - return p.takePendingOutput(p.finishRotate(closed, rng, nowMs)) + envelopes := p.finishRotate(closed, rng, nowMs) + if rng != [2]uint64{} { + p.activatePendingConfig() + } + return p.takePendingOutput(envelopes) } // Drain implements Precompute.Drain. Unconditionally rotates the @@ -509,12 +514,17 @@ func (p *precompute) Drain() []*SketchEnvelope { return nil } closed, rng := p.window.drain(cfg) - return p.takePendingOutput(p.finishRotate(closed, rng, rng[1])) + envelopes := p.finishRotate(closed, rng, rng[1]) + p.activatePendingConfig() + return p.takePendingOutput(envelopes) } func (p *precompute) rotateForFuture(timestampMs uint64, cfg *PrecomputeConfig) { closed, rng := p.window.rotate(timestampMs, cfg) envelopes := p.finishRotate(closed, rng, timestampMs) + if rng != [2]uint64{} { + p.activatePendingConfig() + } if len(envelopes) == 0 { return } @@ -1024,20 +1034,16 @@ func (p *precompute) UpdateConfig(cs *PrecomputeConfigSet) { if chosen == nil { chosen = &cs.Configs[0] } - cfgCopy := *chosen - // A scope flip (PerSeries <-> WholeStream) cannot hot-swap in place: the - // active window's series map is keyed incompatibly under the two scopes - // (one bucket per AggID vs one per series). Drop the in-flight partial - // window before installing the new config so the next observations - // accumulate under the new scope's keying. Same-scope changes (matchers, - // aggregateBy, delta toggles, etc.) leave the window untouched, preserving - // the bytes already accumulated this window (the documented UpdateConfig - // contract). `active` is read above (before the store), so this compares - // the scope that produced the current window against the incoming one. - if active != nil && active.effectiveScope() != cfgCopy.effectiveScope() { - p.window.resetForScopeChange() - } - p.cfg.Store(&cfgCopy) + cfgCopy := clonePrecomputeConfig(chosen) + // An in-flight window is owned by its current immutable config. Stage the + // replacement until Tick/Drain closes that generation; otherwise old sketch + // bytes could be serialized with new family, grouping, or delta semantics. + if active != nil && p.window.hasAccumulatedState() { + p.pendingCfg.Store(cfgCopy) + return + } + p.cfg.Store(cfgCopy) + p.pendingCfg.Store(nil) p.sketchType = cfgCopy.SketchType // Re-evaluate the monitor hooks against the newly installed config so a // control-plane toggle of Monitor.Enabled (or a functional/key change) @@ -1045,6 +1051,36 @@ func (p *precompute) UpdateConfig(cs *PrecomputeConfigSet) { p.rewireMonitorHooks() } +func clonePrecomputeConfig(source *PrecomputeConfig) *PrecomputeConfig { + if source == nil { + return nil + } + cloned := *source + cloned.Matchers = append([]LabelMatcher(nil), source.Matchers...) + cloned.AggregateBy = append([]string(nil), source.AggregateBy...) + cloned.Quantiles = append([]float64(nil), source.Quantiles...) + if source.SketchParams != nil { + cloned.SketchParams = make(SketchParams, len(source.SketchParams)) + for key, value := range source.SketchParams { + cloned.SketchParams[key] = value + } + } + cloned.Monitor.Key = append([]byte(nil), source.Monitor.Key...) + cloned.Monitor.Coeffs = append([]float64(nil), source.Monitor.Coeffs...) + return &cloned +} + +func (p *precompute) activatePendingConfig() { + pending := p.pendingCfg.Swap(nil) + if pending == nil { + return + } + p.cfg.Store(pending) + p.sketchType = pending.SketchType + p.snapshotCache.Reset() // a new generation must start from a full checkpoint + p.rewireMonitorHooks() +} + // Stats implements Precompute.Stats. func (p *precompute) Stats() *PrecomputeStats { return p.stats diff --git a/asap-precompute-go/scope_test.go b/asap-precompute-go/scope_test.go index 9e07c3cc..372edc72 100644 --- a/asap-precompute-go/scope_test.go +++ b/asap-precompute-go/scope_test.go @@ -237,10 +237,9 @@ func TestWholeStreamIgnoresMaxSeries(t *testing.T) { } } -// TestUpdateConfigScopeChangeResetsWindow confirms a live scope flip discards -// the incompatible in-flight window rather than mis-keying it, while a -// same-scope change preserves accumulated state. -func TestUpdateConfigScopeChangeResetsWindow(t *testing.T) { +// TestUpdateConfigStagesAtWindowBoundary confirms a live plan change drains +// the old generation before the replacement becomes active. +func TestUpdateConfigStagesAtWindowBoundary(t *testing.T) { t.Parallel() // Start PerSeries, accumulate two series, then flip to WholeStream. @@ -249,20 +248,25 @@ func TestUpdateConfigScopeChangeResetsWindow(t *testing.T) { _ = p.Observe(&Observation{TimestampMs: 1000, Labels: []KeyValue{{Key: "host", Value: h}}, Value: FloatValue(1)}) } p.UpdateConfig(&PrecomputeConfigSet{Version: 2, Configs: []PrecomputeConfig{*scopeCfg(ModeWholeStream)}}) - // A drain right after the flip emits nothing (window was reset). - if envs := p.Drain(); len(envs) != 0 { - t.Errorf("after scope flip the in-flight window should be empty, got %d envelopes", len(envs)) + if got := p.(*precompute).activeConfig().effectiveScope(); got != ModePerSeries { + t.Fatalf("replacement activated before boundary: %v", got) + } + // Drain emits the old per-series generation, then promotes WholeStream. + if envs := p.Drain(); len(envs) != 2 { + t.Errorf("old generation should drain 2 envelopes, got %d", len(envs)) } // New observations now accumulate under WholeStream ⇒ one envelope. for _, h := range []string{"x", "y", "z"} { - _ = p.Observe(&Observation{TimestampMs: 1000, Labels: []KeyValue{{Key: "host", Value: h}}, Value: FloatValue(1)}) + if err := p.Observe(&Observation{TimestampMs: 2_500, Labels: []KeyValue{{Key: "host", Value: h}}, Value: FloatValue(1)}); err != nil { + t.Fatal(err) + } } if envs := p.Drain(); len(envs) != 1 { t.Fatalf("post-flip WholeStream: want 1 envelope, got %d", len(envs)) } - // Same-scope UpdateConfig must NOT reset: accumulate, reconfigure with the - // same scope, then confirm the prior observation still flushes. + // Same-scope changes are staged too because grouping/encoding and other + // semantics can still change independently of scope. p2 := New(scopeCfg(ModeWholeStream), newFakeFactory(), &fakeObserver{}) _ = p2.Observe(&Observation{TimestampMs: 1000, Labels: []KeyValue{{Key: "host", Value: "a"}}, Value: FloatValue(1)}) sameScope := *scopeCfg(ModeWholeStream) @@ -271,4 +275,7 @@ func TestUpdateConfigScopeChangeResetsWindow(t *testing.T) { if envs := p2.Drain(); len(envs) != 1 { t.Errorf("same-scope change should preserve the window, want 1 envelope, got %d", len(envs)) } + if got := p2.(*precompute).activeConfig().AggregateBy; len(got) != 1 || got[0] != "zone" { + t.Fatalf("pending config not promoted: %v", got) + } } diff --git a/asap-precompute-go/window.go b/asap-precompute-go/window.go index d3d54a86..592e1c0c 100644 --- a/asap-precompute-go/window.go +++ b/asap-precompute-go/window.go @@ -139,6 +139,20 @@ type windowState struct { wakeHook func() } +func (w *windowState) hasAccumulatedState() bool { + w.mu.RLock() + defer w.mu.RUnlock() + if len(w.series) != 0 { + return true + } + for _, pane := range w.panes { + if len(pane.series) != 0 { + return true + } + } + return false +} + // wakeSignaler is implemented by a Sketch that can trigger an out-of-cycle // flush from insert-time GOS threshold detection // (design-gos-unified-edge-telemetry.md §11). ConsumeWakeSignal reports From 58bd918bd4746545e060e99bf455ba7e6a6cbc39 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 4 Sep 2026 05:59:53 -0600 Subject: [PATCH 2/3] fix(precompute): rotate and retry future samples --- asap-precompute-go/precompute.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/asap-precompute-go/precompute.go b/asap-precompute-go/precompute.go index 18a7890e..eddb0242 100644 --- a/asap-precompute-go/precompute.go +++ b/asap-precompute-go/precompute.go @@ -371,6 +371,7 @@ func (p *precompute) Observe(obs *Observation) error { err := p.window.observe(obs, cfg, p.sketchFactory, p.observer, p.stats) if errors.Is(err, ErrFutureData) { p.rotateForFuture(obs.TimestampMs, cfg) + cfg = p.activeConfig() err = p.window.observe(obs, cfg, p.sketchFactory, p.observer, p.stats) } if err != nil { @@ -418,6 +419,8 @@ func (p *precompute) ObserveKeyed(key string, obs *Observation) error { err := p.window.observeKeyed(key, obs, cfg, p.sketchFactory, p.observer, p.stats) if errors.Is(err, ErrFutureData) { p.rotateForFuture(obs.TimestampMs, cfg) + cfg = p.activeConfig() + key = cfg.SeriesKeyFor(obs) err = p.window.observeKeyed(key, obs, cfg, p.sketchFactory, p.observer, p.stats) } if err != nil { From 6ec19d3b2126f84d478020fbef5ea843580e7335 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 4 Sep 2026 06:02:26 -0600 Subject: [PATCH 3/3] fix(precompute): synchronize config and window generations --- asap-precompute-go/precompute.go | 87 ++++++++++++++++++-------------- 1 file changed, 49 insertions(+), 38 deletions(-) diff --git a/asap-precompute-go/precompute.go b/asap-precompute-go/precompute.go index eddb0242..a15486d2 100644 --- a/asap-precompute-go/precompute.go +++ b/asap-precompute-go/precompute.go @@ -297,6 +297,7 @@ type precompute struct { envelopeMu sync.Mutex pendingMu sync.Mutex pendingOutput []*SketchEnvelope + generationMu sync.RWMutex // monitorEngine is the continuous-monitoring (Discipline B) engine. nil // until SetMonitorEngine is called by the adapter; when set AND the active // config has Monitor.Enabled, the window's per-observation hook routes the @@ -345,10 +346,6 @@ func (p *precompute) Observe(obs *Observation) error { if p.closed.Load() { return errors.New("precompute: instance is closed") } - cfg := p.activeConfig() - if cfg == nil { - return ErrNoConfig - } p.stats.InputObservations.Add(1) // Envelope-valued observations route through the dedicated @@ -357,22 +354,10 @@ func (p *precompute) Observe(obs *Observation) error { return p.ObserveEnvelope(obs.Value.Envelope) } - if !cfg.Matches(obs) { - return nil - } - - if p.sketchFactory == nil { - return errors.New("precompute: sketch factory not configured") - } - if p.observer == nil { - return errors.New("precompute: sketch observer not configured") - } - - err := p.window.observe(obs, cfg, p.sketchFactory, p.observer, p.stats) + err := p.observeOnce("", obs, false) if errors.Is(err, ErrFutureData) { - p.rotateForFuture(obs.TimestampMs, cfg) - cfg = p.activeConfig() - err = p.window.observe(obs, cfg, p.sketchFactory, p.observer, p.stats) + p.rotateForFuture(obs.TimestampMs) + err = p.observeOnce("", obs, false) } if err != nil { switch { @@ -399,29 +384,14 @@ func (p *precompute) ObserveKeyed(key string, obs *Observation) error { if p.closed.Load() { return errors.New("precompute: instance is closed") } - cfg := p.activeConfig() - if cfg == nil { - return ErrNoConfig - } p.stats.InputObservations.Add(1) if obs.Value.Kind == KindEnvelope && obs.Value.Envelope != nil { return p.ObserveEnvelope(obs.Value.Envelope) } - if !cfg.Matches(obs) { - return nil - } - if p.sketchFactory == nil { - return errors.New("precompute: sketch factory not configured") - } - if p.observer == nil { - return errors.New("precompute: sketch observer not configured") - } - err := p.window.observeKeyed(key, obs, cfg, p.sketchFactory, p.observer, p.stats) + err := p.observeOnce(key, obs, true) if errors.Is(err, ErrFutureData) { - p.rotateForFuture(obs.TimestampMs, cfg) - cfg = p.activeConfig() - key = cfg.SeriesKeyFor(obs) - err = p.window.observeKeyed(key, obs, cfg, p.sketchFactory, p.observer, p.stats) + p.rotateForFuture(obs.TimestampMs) + err = p.observeOnce("", obs, true) } if err != nil { switch { @@ -435,10 +405,37 @@ func (p *precompute) ObserveKeyed(key string, obs *Observation) error { return nil } +func (p *precompute) observeOnce(key string, obs *Observation, keyed bool) error { + p.generationMu.RLock() + defer p.generationMu.RUnlock() + cfg := p.activeConfig() + if cfg == nil { + return ErrNoConfig + } + if !cfg.Matches(obs) { + return nil + } + if p.sketchFactory == nil { + return errors.New("precompute: sketch factory not configured") + } + if p.observer == nil { + return errors.New("precompute: sketch observer not configured") + } + if keyed { + if key == "" { + key = cfg.SeriesKeyFor(obs) + } + return p.window.observeKeyed(key, obs, cfg, p.sketchFactory, p.observer, p.stats) + } + return p.window.observe(obs, cfg, p.sketchFactory, p.observer, p.stats) +} + // ObserveEnvelope implements Precompute.ObserveEnvelope. func (p *precompute) ObserveEnvelope(env *SketchEnvelope) error { p.envelopeMu.Lock() defer p.envelopeMu.Unlock() + p.generationMu.RLock() + defer p.generationMu.RUnlock() if p.closed.Load() { return errors.New("precompute: instance is closed") } @@ -493,6 +490,8 @@ func (p *precompute) ObserveEnvelope(env *SketchEnvelope) error { // series covering the trailing window (panesPerWindow × slide) — see // windowState.rotateSlidingLocked. func (p *precompute) Tick(nowMs uint64) []*SketchEnvelope { + p.generationMu.Lock() + defer p.generationMu.Unlock() cfg := p.activeConfig() if cfg == nil { return nil @@ -512,6 +511,8 @@ func (p *precompute) Tick(nowMs uint64) []*SketchEnvelope { // Implementation: delegates to windowState.drain which mirrors // rotate's body but skips the `nowMs < activeEndMs` gate. func (p *precompute) Drain() []*SketchEnvelope { + p.generationMu.Lock() + defer p.generationMu.Unlock() cfg := p.activeConfig() if cfg == nil { return nil @@ -522,7 +523,13 @@ func (p *precompute) Drain() []*SketchEnvelope { return p.takePendingOutput(envelopes) } -func (p *precompute) rotateForFuture(timestampMs uint64, cfg *PrecomputeConfig) { +func (p *precompute) rotateForFuture(timestampMs uint64) { + p.generationMu.Lock() + defer p.generationMu.Unlock() + cfg := p.activeConfig() + if cfg == nil { + return + } closed, rng := p.window.rotate(timestampMs, cfg) envelopes := p.finishRotate(closed, rng, timestampMs) if rng != [2]uint64{} { @@ -553,6 +560,8 @@ func (p *precompute) takePendingOutput(current []*SketchEnvelope) []*SketchEnvel // delta for each active series that has DIVERGED past the per-family threshold, // under the window lock, without rotating — so accumulation continues. func (p *precompute) EmitSubWindow(nowMs uint64) []*SketchEnvelope { + p.generationMu.RLock() + defer p.generationMu.RUnlock() cfg := p.activeConfig() if cfg == nil || !cfg.DeltaTransmission { // Without delta encoding there is no in-window base to diff against — @@ -1021,6 +1030,8 @@ func sketchRelativeAccuracy(s Sketch) float64 { // the active config (or the first one if no active config). A // future refactor will route between multiple configs by AggID. func (p *precompute) UpdateConfig(cs *PrecomputeConfigSet) { + p.generationMu.Lock() + defer p.generationMu.Unlock() if cs == nil || len(cs.Configs) == 0 { return }