From bec072dbde672f088a77541c25c3a2df48d44b08 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 4 Sep 2026 05:41:00 -0600 Subject: [PATCH 1/3] fix(precompute): enforce active window timestamps --- asap-precompute-go/precompute.go | 3 +++ asap-precompute-go/window.go | 31 +++++++++++++-------- asap-precompute-go/window_test.go | 45 +++++++++++++++++++++++++++++++ 3 files changed, 68 insertions(+), 11 deletions(-) diff --git a/asap-precompute-go/precompute.go b/asap-precompute-go/precompute.go index 21dd0e55..910c7935 100644 --- a/asap-precompute-go/precompute.go +++ b/asap-precompute-go/precompute.go @@ -149,6 +149,9 @@ var ( // timestamp is older than the active window's lower bound // minus AllowedLateness. ErrLateData = errors.New("precompute: observation timestamp outside allowed lateness") + // ErrFutureData is returned when an observation belongs to a window that + // has not been activated yet. The host must rotate/catch up and retry it. + ErrFutureData = errors.New("precompute: observation timestamp at or beyond active window end") // ErrNoConfig is returned when Precompute has no PrecomputeConfig. ErrNoConfig = errors.New("precompute: no config installed") // ErrAggIDMismatch is returned by ObserveEnvelope when the diff --git a/asap-precompute-go/window.go b/asap-precompute-go/window.go index 97987b8d..0665d468 100644 --- a/asap-precompute-go/window.go +++ b/asap-precompute-go/window.go @@ -261,12 +261,8 @@ func (w *windowState) observe( } w.initWindow(obs.TimestampMs, cfg) - // Late-data check. - if cfg.Window.AllowedLateness > 0 { - latenessMs := uint64(cfg.Window.AllowedLateness / time.Millisecond) - if obs.TimestampMs+latenessMs < w.activeStartMs { - return ErrLateData - } + if err := w.validateTimestampLocked(obs.TimestampMs, cfg); err != nil { + return err } // Build the lookup key into a pooled byte buffer so the common @@ -308,11 +304,8 @@ func (w *windowState) observeKeyed( } w.initWindow(obs.TimestampMs, cfg) - if cfg.Window.AllowedLateness > 0 { - latenessMs := uint64(cfg.Window.AllowedLateness / time.Millisecond) - if obs.TimestampMs+latenessMs < w.activeStartMs { - return ErrLateData - } + if err := w.validateTimestampLocked(obs.TimestampMs, cfg); err != nil { + return err } entry, ok := w.series[key] @@ -325,6 +318,22 @@ func (w *windowState) observeKeyed( return w.recordLocked(entry, obs, observer) } +// validateTimestampLocked prevents host scheduling jitter from changing +// timestamp-defined window semantics. A future-window sample is returned to +// the host for rotate-and-retry; it is never folded into the current sketch. +func (w *windowState) validateTimestampLocked(timestampMs uint64, cfg *PrecomputeConfig) error { + if timestampMs >= w.activeEndMs { + return ErrFutureData + } + if timestampMs < w.activeStartMs { + latenessMs := uint64(cfg.Window.AllowedLateness / time.Millisecond) + if w.activeStartMs-timestampMs > latenessMs { + return ErrLateData + } + } + return nil +} + // admitSeriesLocked creates + registers a new series for key (caller holds // w.mu and confirmed it absent), honoring MaxSeries/OnOverflow and the // parity-mode label-stripping flags. Shared by observe (unkeyed) and diff --git a/asap-precompute-go/window_test.go b/asap-precompute-go/window_test.go index 8fc3800a..106f12a6 100644 --- a/asap-precompute-go/window_test.go +++ b/asap-precompute-go/window_test.go @@ -128,6 +128,51 @@ func TestWindow_LateDataReturnsErrLateData(t *testing.T) { } } +func TestWindow_ZeroLatenessRejectsOlderTimestamp(t *testing.T) { + t.Parallel() + cfg := &PrecomputeConfig{AggID: 1, SketchType: SketchTypeDDSketch, Mode: Tumbling, + Window: WindowSpec{Size: 10 * time.Second}} + w := newWindowState() + if err := w.observe(&Observation{TimestampMs: 10_500, Value: FloatValue(1)}, cfg, newFakeFactory(), &fakeObserver{}, NewStats()); err != nil { + t.Fatal(err) + } + if err := w.observe(&Observation{TimestampMs: 9_999, Value: FloatValue(1)}, cfg, newFakeFactory(), &fakeObserver{}, NewStats()); !errors.Is(err, ErrLateData) { + t.Fatalf("want ErrLateData, got %v", err) + } +} + +func TestWindow_FutureTimestampNeverEntersCurrentWindow(t *testing.T) { + t.Parallel() + cfg := &PrecomputeConfig{AggID: 1, SketchType: SketchTypeDDSketch, Mode: Tumbling, + Window: WindowSpec{Size: 10 * time.Second}} + for _, keyed := range []bool{false, true} { + w := newWindowState() + first := &Observation{TimestampMs: 1_000, Labels: []KeyValue{{Key: "k", Value: "a"}}, Value: FloatValue(1)} + future := &Observation{TimestampMs: 10_000, Labels: first.Labels, Value: FloatValue(2)} + var err error + if keyed { + err = w.observeKeyed(cfg.SeriesKeyFor(first), first, cfg, newFakeFactory(), &fakeObserver{}, NewStats()) + } else { + err = w.observe(first, cfg, newFakeFactory(), &fakeObserver{}, NewStats()) + } + if err != nil { + t.Fatal(err) + } + if keyed { + err = w.observeKeyed(cfg.SeriesKeyFor(future), future, cfg, newFakeFactory(), &fakeObserver{}, NewStats()) + } else { + err = w.observe(future, cfg, newFakeFactory(), &fakeObserver{}, NewStats()) + } + if !errors.Is(err, ErrFutureData) { + t.Fatalf("keyed=%v: want ErrFutureData, got %v", keyed, err) + } + closed, _ := w.rotate(10_000, cfg) + if len(closed) != 1 || closed[0].Count != 1 { + t.Fatalf("keyed=%v: future sample entered old window: %+v", keyed, closed) + } + } +} + func TestWindow_MaxSeriesDropsNew(t *testing.T) { t.Parallel() cfg := &PrecomputeConfig{ From fbc3937f46af85681b717e18bb84f891da393e25 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 4 Sep 2026 05:54:12 -0600 Subject: [PATCH 2/3] test(asapedge): use distinct physical windows --- .../processor/asapedgeprocessor/delta_test.go | 2 +- .../processor/asapedgeprocessor/emit_heap_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/delta_test.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/delta_test.go index 95af4ad1..ce66e980 100644 --- a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/delta_test.go +++ b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/delta_test.go @@ -64,7 +64,7 @@ func TestDeltaTransmissionEmitsDeltaEncoding(t *testing.T) { // Window 2 -> PROTO_DELTA (against the cached window-1 snapshot). for i := 0; i < 10; i++ { - sa.observe(am, float64(i), base+1000+uint64(i), false, 0, 0) + sa.observe(am, float64(i), base+uint64(time.Hour/time.Millisecond)+uint64(i), false, 0, 0) } envs2 := sa.pc.Drain() if !hasEncoding(envs2, precompute.EncodingProtoDelta) { diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/emit_heap_test.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/emit_heap_test.go index 0d7dabca..14bb1f71 100644 --- a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/emit_heap_test.go +++ b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/emit_heap_test.go @@ -225,7 +225,7 @@ func TestEmitHeap_DeltaFrameAfterFirstWindow(t *testing.T) { } // Window 2: MSGPACK_DELTA frame (sparse matrix delta + full heap). - feed(base + 1_000_000) + feed(base + uint64(time.Hour/time.Millisecond)) env2 := drainOneCountSketch(t, sa.pc.Drain()) if env2.Encoding != precompute.EncodingMsgpackDelta { t.Fatalf("window 2 must be MSGPACK_DELTA, got %v", env2.Encoding) From 954f90b1c715dfb030042c5891ca6eb7479ffd9e Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 4 Sep 2026 05:59:53 -0600 Subject: [PATCH 3/3] fix(precompute): rotate and retry future samples --- asap-precompute-go/precompute.go | 44 ++++++++++++++++++++++++++++--- asap-precompute-go/window_test.go | 20 ++++++++++++++ 2 files changed, 60 insertions(+), 4 deletions(-) diff --git a/asap-precompute-go/precompute.go b/asap-precompute-go/precompute.go index 910c7935..ff4c229a 100644 --- a/asap-precompute-go/precompute.go +++ b/asap-precompute-go/precompute.go @@ -294,6 +294,8 @@ type precompute struct { sketchSink atomic.Pointer[SketchSink] frameReceiver frameReceiver envelopeMu sync.Mutex + pendingMu sync.Mutex + pendingOutput []*SketchEnvelope // 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 @@ -365,7 +367,12 @@ func (p *precompute) Observe(obs *Observation) error { return errors.New("precompute: sketch observer not configured") } - if err := p.window.observe(obs, cfg, p.sketchFactory, p.observer, p.stats); err != nil { + err := p.window.observe(obs, cfg, p.sketchFactory, p.observer, p.stats) + if errors.Is(err, ErrFutureData) { + p.rotateForFuture(obs.TimestampMs, cfg) + err = p.window.observe(obs, cfg, p.sketchFactory, p.observer, p.stats) + } + if err != nil { switch { case errors.Is(err, ErrSeriesCapExceeded): p.stats.DroppedOverflow.Add(1) @@ -407,7 +414,12 @@ func (p *precompute) ObserveKeyed(key string, obs *Observation) error { if p.observer == nil { return errors.New("precompute: sketch observer not configured") } - if err := p.window.observeKeyed(key, obs, cfg, p.sketchFactory, p.observer, p.stats); err != nil { + err := p.window.observeKeyed(key, obs, cfg, p.sketchFactory, p.observer, p.stats) + if errors.Is(err, ErrFutureData) { + p.rotateForFuture(obs.TimestampMs, cfg) + err = p.window.observeKeyed(key, obs, cfg, p.sketchFactory, p.observer, p.stats) + } + if err != nil { switch { case errors.Is(err, ErrSeriesCapExceeded): p.stats.DroppedOverflow.Add(1) @@ -482,7 +494,7 @@ func (p *precompute) Tick(nowMs uint64) []*SketchEnvelope { return nil } closed, rng := p.window.rotate(nowMs, cfg) - return p.finishRotate(closed, rng, nowMs) + return p.takePendingOutput(p.finishRotate(closed, rng, nowMs)) } // Drain implements Precompute.Drain. Unconditionally rotates the @@ -497,7 +509,31 @@ func (p *precompute) Drain() []*SketchEnvelope { return nil } closed, rng := p.window.drain(cfg) - return p.finishRotate(closed, rng, rng[1]) + return p.takePendingOutput(p.finishRotate(closed, rng, rng[1])) +} + +func (p *precompute) rotateForFuture(timestampMs uint64, cfg *PrecomputeConfig) { + closed, rng := p.window.rotate(timestampMs, cfg) + envelopes := p.finishRotate(closed, rng, timestampMs) + if len(envelopes) == 0 { + return + } + p.pendingMu.Lock() + p.pendingOutput = append(p.pendingOutput, envelopes...) + p.pendingMu.Unlock() +} + +func (p *precompute) takePendingOutput(current []*SketchEnvelope) []*SketchEnvelope { + p.pendingMu.Lock() + defer p.pendingMu.Unlock() + if len(p.pendingOutput) == 0 { + return current + } + result := make([]*SketchEnvelope, 0, len(p.pendingOutput)+len(current)) + result = append(result, p.pendingOutput...) + result = append(result, current...) + p.pendingOutput = nil + return result } // EmitSubWindow implements Precompute.EmitSubWindow: serialize an incremental diff --git a/asap-precompute-go/window_test.go b/asap-precompute-go/window_test.go index 106f12a6..a021d450 100644 --- a/asap-precompute-go/window_test.go +++ b/asap-precompute-go/window_test.go @@ -173,6 +173,26 @@ func TestWindow_FutureTimestampNeverEntersCurrentWindow(t *testing.T) { } } +func TestPrecompute_FutureTimestampRotatesQueuesAndRetries(t *testing.T) { + cfg := &PrecomputeConfig{AggID: 1, SketchType: SketchTypeDDSketch, Mode: Tumbling, + Window: WindowSpec{Size: 10 * time.Second}} + p := New(cfg, newFakeFactory(), &fakeObserver{}).(*precompute) + if err := p.Observe(&Observation{TimestampMs: 1_000, Labels: []KeyValue{{Key: "k", Value: "a"}}, Value: FloatValue(1)}); err != nil { + t.Fatal(err) + } + if err := p.Observe(&Observation{TimestampMs: 11_000, Labels: []KeyValue{{Key: "k", Value: "a"}}, Value: FloatValue(2)}); err != nil { + t.Fatal(err) + } + first := p.Tick(11_000) + if len(first) != 1 || first[0].WindowStartMs != 0 || first[0].WindowEndMs != 10_000 { + t.Fatalf("queued old window: %+v", first) + } + second := p.Drain() + if len(second) != 1 || second[0].WindowStartMs != 10_000 || second[0].WindowEndMs != 20_000 { + t.Fatalf("retried new window: %+v", second) + } +} + func TestWindow_MaxSeriesDropsNew(t *testing.T) { t.Parallel() cfg := &PrecomputeConfig{