fix(precompute): enforce active window timestamps

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
@@ -291,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
@@ -362,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)
@@ -404,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)
@@ -479,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
@@ -494,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

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

asap-precompute-go/window_test.go

@@ -128,6 +128,71 @@ 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 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{

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) {

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)