fix(asap-precompute-go): add Drain() and remove 6 pseudo-timestamp workarounds

[zzylol]

The bug

Precompute.Tick(nowMs) only rotates the active window when
nowMs >= activeEndMs. Callers that wanted to drain pending data
on shutdown naturally passed time.Now() — but if shutdown happens
mid-window, nowMs < activeEndMs and Tick returns nil. Mid-window
observations were silently dropped on the lifecycle exit.

All 6 callers worked around this with one of three hacky
pseudo-timestamp patterns:

• HLL OTel shim: nextFlushTick() → 1<<50 base + 1<<32 stride per call
• DDSketch / CountSketch / CountMin OTel shims: forceTickMs = 1<<62 - 1
• Telegraf allsketches plugin: time.Now() + 2*window
• KLL OTel shim: time.Now() (silently buggy on shutdown — no
  workaround, just the original bug)

Same fix written 6 different ways. Solve it once at the API level.

The fix

Add Precompute.Drain() — a sibling of Tick(nowMs) that
unconditionally rotates the active window regardless of wall-clock
time and returns the resulting envelopes. After Drain, the next
active window starts at the same boundary Tick would have used at
natural rotation (activeStartMs := old activeEndMs; activeEndMs += windowSize).

Internals: windowState.rotate and the new windowState.drain
both delegate to a shared rotateLocked helper. Drain skips the
nowMs < activeEndMs gate, and is a no-op on an already-empty
active window so callers can invoke it multiple times on shutdown
without fast-forwarding through empty buckets.

Affected callers (all on main)

Caller	Was	Now
opentelemetry-collector-contrib-patch/processor/hllprocessor/processor.go	nextFlushTick() (1<<50 + 1<<32 stride, plus flushSeq atomic.Uint64 field)	Drain()
opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/shim_helpers.go	Tick(1<<62 - 1)	Drain()
opentelemetry-collector-contrib-patch/processor/countsketchprocessor/shim_helpers.go	Tick(1<<62 - 1)	Drain()
opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/{processor,shim_helpers}.go	Tick(1<<62 - 1) (batch) + Tick(time.Now()) (window)	Drain() (both paths)
opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go	Tick(time.Now()) (silently buggy on shutdown)	Drain()
telegraf-patch/processors/allsketches/allsketches.go	Tick(time.Now() + 2*window) on Stop	Drain() on Stop

PR #237 (Telegraf allsketches plugin) was merged shortly before
this PR was opened, so the Telegraf fix lands here on main rather
than as a follow-up commit on #237's branch.

Validation

cd asap-precompute-go && go test -race ./...
# ok  github.com/ProjectASAP/asap-precompute-go               1.018s
# ok  github.com/ProjectASAP/asap-precompute-go/controlchannel 1.070s
# ok  github.com/ProjectASAP/asap-precompute-go/otel           1.019s
# ok  github.com/ProjectASAP/asap-precompute-go/sketches       1.027s
# ok  github.com/ProjectASAP/asap-precompute-go/telegraf       1.019s

for p in ddsketch kll hll countsketch countminsketch; do
  cd opentelemetry-collector-contrib-patch/processor/${p}processor
  go test -race ./...
done
# all 5 shims green

cd telegraf-patch/processors/allsketches && go test -race ./...
# ok

cd integration/parity && go test -v ./...
# TestParity_AllSketches/TestParity_DDSketch        — 6 envelopes byte-identical
# TestParity_AllSketches/TestParity_KLL             — 3 envelopes byte-identical
# TestParity_AllSketches/TestParity_HLL             — 3 envelopes byte-identical
# TestParity_AllSketches/TestParity_CountSketch     — 1 envelope byte-identical
# TestParity_AllSketches/TestParity_CountMinSketch  — 602 envelopes byte-identical

Parity harness: all-PASS byte-identical, all 5 sketches.

New unit tests

• TestPrecompute_DrainFlushesMidWindowObservations — exercises
  the bug directly: observe 3 values early in window 0, confirm
  Tick(2_000) (mid-window) returns 0 envelopes, confirm
  Drain() returns 1 envelope with Count=3 and the natural
  [0, 10_000) window range. Also confirms the next active
  window starts at the natural rotation boundary so a follow-up
  Tick(20_000) flushes window 1 cleanly.
• TestPrecompute_DrainEqualsTickAtBoundary — pins the contract
  that Drain emits the same envelope payload Tick would emit if
  called precisely at activeEndMs (same range, same count, same
  bytes).

LoC removed across the 6 caller sites

~47 LoC of workaround code (constants, helpers, pseudo-timestamp
arithmetic, and their explanatory comments) removed across the 6
caller sites. Replaced by direct Drain() calls.

🤖 Generated with Claude Code