This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 6fb3a1ba8 feat(trace): finalization sampling (PIPELINE_EVENT_FINALIZE)
(#1210)
6fb3a1ba8 is described below
commit 6fb3a1ba8663aad71856d31ad90ab4b1704def98
Author: Gao Hongtao <[email protected]>
AuthorDate: Wed Jul 8 16:51:10 2026 +0800
feat(trace): finalization sampling (PIPELINE_EVENT_FINALIZE) (#1210)
---
CHANGES.md | 1 +
banyand/internal/storage/segment.go | 39 +++
banyand/internal/storage/storage.go | 17 ++
banyand/internal/storage/tsdb.go | 7 +
banyand/trace/finalize_accounting_test.go | 75 ++++++
banyand/trace/finalize_scanner.go | 224 +++++++++++++++++
banyand/trace/finalize_scanner_test.go | 259 ++++++++++++++++++++
banyand/trace/finalize_state.go | 96 ++++++++
banyand/trace/finalize_state_test.go | 104 ++++++++
banyand/trace/finalizer.go | 182 ++++++++++++++
banyand/trace/finalizer_test.go | 394 ++++++++++++++++++++++++++++++
banyand/trace/flusher.go | 2 +-
banyand/trace/introducer.go | 26 ++
banyand/trace/merger.go | 127 ++++++++--
banyand/trace/merger_durability_test.go | 2 +-
banyand/trace/merger_test.go | 6 +-
banyand/trace/metadata.go | 43 +++-
banyand/trace/part_metadata.go | 6 +
banyand/trace/pipeline_finalize_test.go | 189 ++++++++++++++
banyand/trace/pipeline_registry.go | 144 +++++++++++
banyand/trace/pipeline_watch_test.go | 23 +-
banyand/trace/svc_standalone.go | 23 ++
banyand/trace/trace.go | 1 +
banyand/trace/tstable.go | 44 +++-
docs/design/post-trace-pipeline.md | 49 ++++
implementation-fn-note.md | 117 +++++++++
26 files changed, 2149 insertions(+), 51 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 4957fd89f..fdc916874 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -5,6 +5,7 @@ Release Notes.
## 0.11.0
### Features
+- Add trace **finalization sampling** (`PIPELINE_EVENT_FINALIZE`), a
best-effort background backstop for the in-merge trace retention filter. A
single node-wide, concurrency-1 scanner periodically sweeps cooled segments
(`segEnd < now − finalize_grace`, reopening idle-closed segments via
`SelectSegments(reopenClosed=true)`) and force-merges each shard's un-finalized
parts through the group's registered sampler chain, reusing the hot merge path
(`mergeParts` + per-sidx `Merge` + the shard [...]
- Add two catalogs to the queue batch-write metrics so traffic is comparable
on both ends: a per-batch-stream **batch catalog**
(`total_batch_started`/`total_batch_finished`/`total_batch_latency`, buckets to
~300s) on `queue_pub`/`queue_sub` and the `lifecycle_migration` mirror, and a
per-message **message catalog**
(`total_message_started`/`total_message_finished`) on `queue_sub` (the
publisher's existing `total_*` already counts per message). All existing
`total_*` series are unchanged [...]
- Redesign the queue (`queue_pub`/`queue_sub`) metrics around a uniform model:
keep only `total_started`, `total_finished`, `total_latency` (now a histogram)
and `total_err`, plus file-sync-only `sent_bytes` (pub) / `received_bytes`
(sub). Replace the `topic` label with `operation`
(`batch-write`/`file-sync`/`query`/`control`) and `group`, add an `error_type`
label on `total_err`, and add remote-endpoint labels
(`remote_node`/`remote_role`/`remote_tier`) so the liaison↔data (hot/warm/col
[...]
- Stamp the lifecycle's tier-migration publisher's identity onto the wire so
the receiving data node records a non-empty
`remote_node`/`remote_role`/`remote_tier` on its
`banyandb_queue_sub_total_finished` series. The lifecycle's `parseGroup`
resolves the lifecycle's self identity by matching the lifecycle pod's hostname
(POD_NAME via the K8s downward API, falling back to `os.Hostname()` — same
precedence as `nativeNodeContext` at `banyand/backup/lifecycle/service.go`)
against the data-n [...]
diff --git a/banyand/internal/storage/segment.go
b/banyand/internal/storage/segment.go
index 727e41540..965a6d470 100644
--- a/banyand/internal/storage/segment.go
+++ b/banyand/internal/storage/segment.go
@@ -617,6 +617,12 @@ func (sc *segmentController[T, O])
selectSegments(timeRange timestamp.TimeRange,
if reopenClosed {
// Real read: reopen if closed and mark as
accessed.
if err = s.incRef(ctx); err != nil {
+ // Release the segments already pinned
in earlier iterations so a
+ // mid-loop incRef failure does not
leak refs (which would block
+ // idle-close and retention-delete for
them indefinitely).
+ for _, pinned := range tt {
+ pinned.DecRef()
+ }
return nil, err
}
s.lastAccessed.Store(now)
@@ -638,6 +644,39 @@ func (sc *segmentController[T, O])
selectSegments(timeRange timestamp.TimeRange,
return tt, nil
}
+// peekSegments returns lightweight descriptors of the segments overlapping
timeRange
+// WITHOUT opening (incRef-ing) any of them. It snapshots the matching
segments' time
+// ranges and locations under the read lock, then lists each segment's on-disk
shard
+// directories outside the lock (a cheap readdir, no index reopen or loop
spawn).
+func (sc *segmentController[T, O]) peekSegments(timeRange timestamp.TimeRange)
[]SegmentPeek {
+ type loc struct {
+ start time.Time
+ end time.Time
+ location string
+ }
+ sc.RLock()
+ var locs []loc
+ for _, s := range sc.lst {
+ if s.Overlapping(timeRange) {
+ locs = append(locs, loc{start: s.Start, end: s.End,
location: s.location})
+ }
+ }
+ sc.RUnlock()
+
+ out := make([]SegmentPeek, 0, len(locs))
+ for _, l := range locs {
+ peek := SegmentPeek{Start: l.start, End: l.end}
+ _ = walkDir(l.location, shardPathPrefix, func(suffix string)
error {
+ if id, convErr := strconv.Atoi(suffix); convErr == nil {
+ peek.ShardPaths = append(peek.ShardPaths,
filepath.Join(l.location, fmt.Sprintf(shardTemplate, id)))
+ }
+ return nil
+ })
+ out = append(out, peek)
+ }
+ return out
+}
+
func (sc *segmentController[T, O]) createSegment(ts time.Time) (*segment[T,
O], error) {
s, err := sc.create(context.Background(), ts)
if err != nil {
diff --git a/banyand/internal/storage/storage.go
b/banyand/internal/storage/storage.go
index cb5c8d495..74e72f5c0 100644
--- a/banyand/internal/storage/storage.go
+++ b/banyand/internal/storage/storage.go
@@ -116,6 +116,13 @@ type TSDB[T TSTable, O any] interface {
// stats) returns them without reopening or refreshing their idle
timer. The
// caller must DecRef every returned segment (a no-op for a closed one).
SelectSegments(timeRange timestamp.TimeRange, reopenClosed bool)
([]Segment[T, O], error)
+ // PeekSegments returns lightweight descriptors of the segments
overlapping
+ // timeRange WITHOUT opening (incRef-ing) any of them. It lets
background
+ // housekeeping inspect per-shard on-disk state cheaply and decide
whether a
+ // segment is worth reopening for real work, avoiding a reopen storm
over cold
+ // segments. The returned ShardPaths are on-disk directories; the
caller reads
+ // them read-only.
+ PeekSegments(timeRange timestamp.TimeRange) []SegmentPeek
// SegmentInterval returns the current segment interval rule.
SegmentInterval() IntervalRule
Tick(ts int64)
@@ -133,6 +140,16 @@ type TSDB[T TSTable, O any] interface {
Drop() error
}
+// SegmentPeek describes a segment discoverable WITHOUT opening (incRef-ing)
it — the
+// segment's time range plus the on-disk paths of its existing shard
directories.
+// Background housekeeping (e.g. the trace finalize scanner) uses it to inspect
+// per-shard on-disk state before deciding whether to reopen the segment.
+type SegmentPeek struct {
+ Start time.Time
+ End time.Time
+ ShardPaths []string
+}
+
// Segment is a time range of data.
type Segment[T TSTable, O any] interface {
DecRef()
diff --git a/banyand/internal/storage/tsdb.go b/banyand/internal/storage/tsdb.go
index 4c4dc83ac..282ee5e99 100644
--- a/banyand/internal/storage/tsdb.go
+++ b/banyand/internal/storage/tsdb.go
@@ -299,6 +299,13 @@ func (d *database[T, O]) SelectSegments(timeRange
timestamp.TimeRange, reopenClo
return kept, nil
}
+func (d *database[T, O]) PeekSegments(timeRange timestamp.TimeRange)
[]SegmentPeek {
+ if d.closed.Load() {
+ return nil
+ }
+ return d.segmentController.peekSegments(timeRange)
+}
+
func (d *database[T, O]) SegmentInterval() IntervalRule {
return d.segmentController.getSegmentInterval()
}
diff --git a/banyand/trace/finalize_accounting_test.go
b/banyand/trace/finalize_accounting_test.go
new file mode 100644
index 000000000..0b08ef449
--- /dev/null
+++ b/banyand/trace/finalize_accounting_test.go
@@ -0,0 +1,75 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+// TestAccountUnsampledFlushed verifies the O(1) hot-path accounting: a no-op
until the
+// shard has been finalized (gen>0), then accumulates uncompressed span bytes.
The
+// method takes no filesystem and reads only in-memory part metadata +
atomics, so by
+// construction it performs zero metadata I/O on the flush path (Phase 3
acceptance).
+func TestAccountUnsampledFlushed(t *testing.T) {
+ tst := &tsTable{}
+ flushed := map[uint64]*partWrapper{
+ 1: {p: &part{partMetadata:
partMetadata{UncompressedSpanSizeBytes: 100}}},
+ 2: {p: &part{partMetadata:
partMetadata{UncompressedSpanSizeBytes: 50}}},
+ }
+
+ // gen == 0 (never finalized): no accounting.
+ tst.accountUnsampledFlushed(flushed)
+ assert.Zero(t, tst.unsampledBytes.Load(), "gen 0 must not accumulate")
+
+ // gen > 0: accumulate uncompressed span bytes.
+ tst.finalizeGenCached.Store(1)
+ tst.accountUnsampledFlushed(flushed)
+ assert.Equal(t, int64(150), tst.unsampledBytes.Load())
+
+ // Subsequent flushes keep accumulating.
+ tst.accountUnsampledFlushed(map[uint64]*partWrapper{
+ 3: {p: &part{partMetadata:
partMetadata{UncompressedSpanSizeBytes: 25}}},
+ })
+ assert.Equal(t, int64(175), tst.unsampledBytes.Load())
+
+ // An empty flush set is a harmless no-op.
+ tst.accountUnsampledFlushed(nil)
+ assert.Equal(t, int64(175), tst.unsampledBytes.Load())
+}
+
+// TestInitTSTable_SeedsFinalizeCache verifies the cached finalize fields are
seeded
+// from the shard's persisted finalizeState at open (restart recovery).
+func TestInitTSTable_SeedsFinalizeCache(t *testing.T) {
+ fileSystem := fs.NewLocalFileSystem()
+ dir := t.TempDir()
+ require.NoError(t, writeFinalizeState(fileSystem, dir,
finalizeState{FinalizeGeneration: 4, UnsampledBytes: 999}))
+
+ tst, _ := initTSTable(fileSystem, dir, common.Position{},
logger.GetLogger("finalize-seed-test"),
+ option{protector: protector.Nop{}}, nil)
+
+ assert.Equal(t, uint64(4), tst.finalizeGenCached.Load(),
"finalizeGenCached must be seeded from disk")
+ assert.Equal(t, int64(999), tst.unsampledBytes.Load(), "unsampledBytes
must be seeded from disk")
+}
diff --git a/banyand/trace/finalize_scanner.go
b/banyand/trace/finalize_scanner.go
new file mode 100644
index 000000000..13df8cbd8
--- /dev/null
+++ b/banyand/trace/finalize_scanner.go
@@ -0,0 +1,224 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "time"
+
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+// defaultFinalizeScanInterval is the cadence of the background finalize
scanner. It
+// mirrors the storage idle-check cadence: a backstop does not need tight
latency.
+const defaultFinalizeScanInterval = 10 * time.Minute
+
+// finalizeScanLoop is the trace-owned periodic finalization-sampling scanner.
It runs
+// as a single node-wide goroutine, so finalize compute is serialized
(concurrency-1)
+// and can never fan out to starve the hot merge lanes. On each tick it sweeps
every
+// finalize-enabled group's cooled segments and runs bounded finalize rounds.
+func (sr *schemaRepo) finalizeScanLoop(closer *run.Closer, interval
time.Duration) {
+ defer closer.Done()
+ ticker := time.NewTicker(interval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-closer.CloseNotify():
+ return
+ case <-ticker.C:
+ sr.runFinalizeScan(closer.CloseNotify())
+ }
+ }
+}
+
+// runFinalizeScan performs one scan pass over all finalize-enabled groups. It
is safe
+// to call directly (tests do) without the surrounding loop.
+func (sr *schemaRepo) runFinalizeScan(closeCh <-chan struct{}) {
+ // A local filesystem to read per-shard finalize.json during the
pre-filter, without
+ // reopening the segment. finalize.json is always on local disk.
+ lfs := fs.NewLocalFileSystem()
+ for _, group := range listFinalizeGroups() {
+ select {
+ case <-closeCh:
+ return
+ default:
+ }
+ samplers := lookupSamplers(group)
+ if len(samplers) == 0 {
+ continue
+ }
+ graceNs := lookupFinalizeGrace(group)
+ if graceNs <= 0 {
+ graceNs = int64(sr.finalizeGraceDefault)
+ }
+ if graceNs <= 0 {
+ continue
+ }
+ sr.scanGroup(group, samplers, graceNs,
lookupFinalizeConfig(group, graceNs), lfs, closeCh)
+ }
+}
+
+// scanGroup peeks the group's cooled segments WITHOUT reopening them,
pre-filters on
+// per-shard on-disk state so terminal / in-cooldown / max-rounds segments are
never
+// reopened (this is what keeps a backstop from perturbing the hot node with a
reopen
+// storm over cold segments), and only reopens the segments that may warrant a
round —
+// then applies the precise threshold and finalizes warranting shards
sequentially.
+func (sr *schemaRepo) scanGroup(group string, samplers []sdk.Sampler, graceNs
int64, cfg finalizeConfig, lfs fs.FileSystem, closeCh <-chan struct{}) {
+ tsdb, err := sr.loadTSDB(group)
+ if err != nil || tsdb == nil {
+ return
+ }
+ now := time.Now().UnixNano()
+ coolEnd := now - graceNs
+ coolRange := timestamp.TimeRange{Start: time.Unix(0, 0), End:
time.Unix(0, coolEnd)}
+ for _, peek := range tsdb.PeekSegments(coolRange) {
+ select {
+ case <-closeCh:
+ return
+ default:
+ }
+ // Only fully-cooled segments; and skip (without reopening) any
whose shards are
+ // all terminal / in-cooldown / max-rounds per their on-disk
finalize state.
+ if peek.End.UnixNano() > coolEnd || !segmentMayWarrant(lfs,
peek.ShardPaths, cfg) {
+ continue
+ }
+ // This segment may warrant work: reopen just its range and
finalize warranting
+ // shards. The precise byte threshold is applied here
(warrantsFinalize) against
+ // the reopened snapshot.
+ segs, selErr := tsdb.SelectSegments(timestamp.TimeRange{Start:
peek.Start, End: peek.End}, true)
+ if selErr != nil {
+ sr.l.Warn().Err(selErr).Str("group",
group).Msg("finalize scan: reopen segment failed")
+ continue
+ }
+ sr.finalizeReopened(group, segs, samplers, graceNs, cfg,
coolEnd)
+ }
+}
+
+// finalizeReopened runs bounded finalize rounds on the warranting shards of
the reopened
+// segments, then DecRefs every segment on all paths.
+func (sr *schemaRepo) finalizeReopened(group string, segs
[]storage.Segment[*tsTable, option],
+ samplers []sdk.Sampler, graceNs int64, cfg finalizeConfig, coolEnd
int64,
+) {
+ defer func() {
+ for _, seg := range segs {
+ seg.DecRef()
+ }
+ }()
+ for _, seg := range segs {
+ // SelectSegments may return a boundary neighbor; guard on the
cool window again.
+ if seg.GetTimeRange().End.UnixNano() > coolEnd {
+ continue
+ }
+ tables, _ := seg.Tables()
+ for _, tst := range tables {
+ if !warrantsFinalize(tst, cfg) {
+ continue
+ }
+ if _, ferr := tst.runFinalizeRound(samplers, graceNs);
ferr != nil {
+ sr.l.Warn().Err(ferr).Str("group",
group).Msg("finalize round failed (fail-open)")
+ }
+ }
+ }
+}
+
+// segmentMayWarrant reports whether any of a cooled segment's shards is worth
reopening,
+// judged purely from on-disk finalize state (no segment reopen). It is a
conservative
+// superset of warrantsFinalize: it never skips a shard that would actually
warrant.
+func segmentMayWarrant(lfs fs.FileSystem, shardPaths []string, cfg
finalizeConfig) bool {
+ for _, shardPath := range shardPaths {
+ if shardMayWarrant(readFinalizeState(lfs, shardPath), cfg) {
+ return true
+ }
+ }
+ return false
+}
+
+// shardMayWarrant is the reopen pre-filter for one shard. It skips only the
cases that
+// are authoritative from on-disk state regardless of whether the shard is
open: terminal,
+// the hard round cap, and an un-elapsed cooldown. A never-finalized shard
always warrants;
+// a finalized one past its cooldown is reopened so the precise byte threshold
can run
+// against the live snapshot (the on-disk unsampled counter can lag an open
shard).
+func shardMayWarrant(st finalizeState, cfg finalizeConfig) bool {
+ if st.Terminal || st.FinalizeRounds >= cfg.maxRounds {
+ return false
+ }
+ if st.FinalizeGeneration == 0 || st.LastFinalizedAt == "" {
+ return true
+ }
+ last, perr := time.Parse(time.RFC3339Nano, st.LastFinalizedAt)
+ if perr != nil {
+ return true
+ }
+ return time.Since(last) >= time.Duration(cfg.cooldownNs)
+}
+
+// warrantsFinalize decides whether a shard warrants another finalize round.
It reads
+// the per-shard persisted state (scanner path — off the hot path, so metadata
I/O is
+// fine) and applies the threshold: never-finalized always warrants; otherwise
the
+// newly-arrived unsampled bytes must cross max(FLOOR, RATIO*total); a
cooldown caps the
+// frequency; and max_finalize_rounds is a hard terminal cap.
+func warrantsFinalize(tst *tsTable, cfg finalizeConfig) bool {
+ st := readFinalizeState(tst.fileSystem, tst.root)
+ if st.Terminal {
+ return false
+ }
+ if st.FinalizeRounds >= cfg.maxRounds {
+ // Hard lifetime cap reached: mark terminal so the shard is
never scanned again.
+ st.Terminal = true
+ if err := writeFinalizeState(tst.fileSystem, tst.root, st); err
!= nil {
+ tst.l.Warn().Err(err).Str("group",
tst.group).Msg("cannot mark finalize terminal")
+ }
+ return false
+ }
+ if st.LastFinalizedAt != "" {
+ if last, perr := time.Parse(time.RFC3339Nano,
st.LastFinalizedAt); perr == nil {
+ if time.Since(last) < time.Duration(cfg.cooldownNs) {
+ return false
+ }
+ }
+ }
+ if st.FinalizeGeneration == 0 {
+ return true
+ }
+ total := shardTotalUncompressed(tst)
+ threshold := int64(cfg.floorBytes)
+ if r := int64(cfg.ratio * float64(total)); r > threshold {
+ threshold = r
+ }
+ return tst.unsampledBytes.Load() >= threshold
+}
+
+// shardTotalUncompressed sums the uncompressed span bytes of the shard's
current file
+// parts, the denominator for the RATIO threshold.
+func shardTotalUncompressed(tst *tsTable) int64 {
+ s := tst.currentSnapshot()
+ if s == nil {
+ return 0
+ }
+ defer s.decRef()
+ var total int64
+ for _, pw := range s.parts {
+ if pw.mp == nil {
+ total +=
int64(pw.p.partMetadata.UncompressedSpanSizeBytes)
+ }
+ }
+ return total
+}
diff --git a/banyand/trace/finalize_scanner_test.go
b/banyand/trace/finalize_scanner_test.go
new file mode 100644
index 000000000..0bafcc573
--- /dev/null
+++ b/banyand/trace/finalize_scanner_test.go
@@ -0,0 +1,259 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+// TestWarrantsFinalize exercises the threshold decision: never-finalized
always
+// warrants; a terminal or max-rounds shard never does (and max marks
terminal); an
+// un-elapsed cooldown blocks; and once finalized, the unsampled-bytes counter
must
+// cross the floor.
+func TestWarrantsFinalize(t *testing.T) {
+ fileSystem := fs.NewLocalFileSystem()
+ newTst := func(dir string) *tsTable {
+ return &tsTable{fileSystem: fileSystem, root: dir, l:
logger.GetLogger("warrants-test")}
+ }
+ // ratio 0 so the threshold is just the absolute floor (no snapshot
needed).
+ cfg := finalizeConfig{floorBytes: 100, ratio: 0, cooldownNs:
int64(time.Minute), maxRounds: 3}
+
+ // Never finalized (no state file) => always warrants.
+ assert.True(t, warrantsFinalize(newTst(t.TempDir()), cfg), "gen 0 must
always warrant")
+
+ // Terminal => never.
+ dTerm := t.TempDir()
+ require.NoError(t, writeFinalizeState(fileSystem, dTerm,
finalizeState{Terminal: true}))
+ assert.False(t, warrantsFinalize(newTst(dTerm), cfg))
+
+ // Rounds at the cap => false AND the shard is marked terminal.
+ dMax := t.TempDir()
+ require.NoError(t, writeFinalizeState(fileSystem, dMax,
finalizeState{FinalizeGeneration: 1, FinalizeRounds: 3}))
+ assert.False(t, warrantsFinalize(newTst(dMax), cfg))
+ assert.True(t, readFinalizeState(fileSystem, dMax).Terminal, "reaching
max rounds must mark terminal")
+
+ // Cooldown not elapsed => false.
+ dCool := t.TempDir()
+ require.NoError(t, writeFinalizeState(fileSystem, dCool, finalizeState{
+ FinalizeGeneration: 1, FinalizeRounds: 1,
+ LastFinalizedAt: time.Now().UTC().Format(time.RFC3339Nano),
+ }))
+ assert.False(t, warrantsFinalize(newTst(dCool), cfg))
+
+ // Finalized, cooldown elapsed: warrants iff unsampled bytes reach the
floor.
+ dThresh := t.TempDir()
+ require.NoError(t, writeFinalizeState(fileSystem, dThresh,
finalizeState{
+ FinalizeGeneration: 1, FinalizeRounds: 1,
+ LastFinalizedAt:
time.Now().Add(-time.Hour).UTC().Format(time.RFC3339Nano),
+ }))
+ tst := newTst(dThresh)
+ tst.unsampledBytes.Store(200)
+ assert.True(t, warrantsFinalize(tst, cfg), "unsampled above floor must
warrant")
+ tst.unsampledBytes.Store(50)
+ assert.False(t, warrantsFinalize(tst, cfg), "unsampled below floor must
not warrant")
+}
+
+// TestWarrantsFinalize_RatioBranch exercises the RATIO*total threshold (the
branch
+// TestWarrantsFinalize's ratio=0 never reaches): for a large finalized shard,
a re-round
+// is warranted only when new unsampled bytes reach RATIO of the segment total.
+func TestWarrantsFinalize_RatioBranch(t *testing.T) {
+ fileSystem := fs.NewLocalFileSystem()
+ dir := t.TempDir()
+ // Finalized (gen>0) with an elapsed cooldown, so warrants gates purely
on the bytes.
+ require.NoError(t, writeFinalizeState(fileSystem, dir, finalizeState{
+ FinalizeGeneration: 1, FinalizeRounds: 1,
+ LastFinalizedAt:
time.Now().Add(-time.Hour).UTC().Format(time.RFC3339Nano),
+ }))
+ tst := &tsTable{fileSystem: fileSystem, root: dir, l:
logger.GetLogger("warrants-ratio-test")}
+ // Snapshot totals 1000 uncompressed bytes so RATIO dominates the
absolute floor.
+ tst.snapshot = &snapshot{ref: 1, parts: []*partWrapper{
+ {p: &part{partMetadata: partMetadata{UncompressedSpanSizeBytes:
1000, TotalCount: 1}}},
+ }}
+ cfg := finalizeConfig{floorBytes: 100, ratio: 0.5, cooldownNs:
int64(time.Minute), maxRounds: 8}
+
+ tst.unsampledBytes.Store(600)
+ assert.True(t, warrantsFinalize(tst, cfg), "600 >= max(floor 100,
0.5*1000=500) must warrant")
+ tst.unsampledBytes.Store(400)
+ assert.False(t, warrantsFinalize(tst, cfg), "400 < 500 (ratio-dominated
threshold) must not warrant")
+}
+
+// TestShardMayWarrant verifies the reopen pre-filter: it skips (without
reopening) only
+// the authoritative-from-disk cases — terminal, max-rounds, and un-elapsed
cooldown —
+// and conservatively reopens everything else (never-finalized, past-cooldown).
+func TestShardMayWarrant(t *testing.T) {
+ cfg := finalizeConfig{floorBytes: 100, ratio: 0.1, cooldownNs:
int64(time.Minute), maxRounds: 8}
+
+ assert.True(t, shardMayWarrant(finalizeState{}, cfg), "never-finalized
must be reopened")
+ assert.False(t, shardMayWarrant(finalizeState{Terminal: true}, cfg),
"terminal must be skipped")
+ assert.False(t, shardMayWarrant(finalizeState{FinalizeGeneration: 1,
FinalizeRounds: 8}, cfg), "max rounds must be skipped")
+ assert.False(t, shardMayWarrant(finalizeState{
+ FinalizeGeneration: 1, FinalizeRounds: 1,
+ LastFinalizedAt: time.Now().UTC().Format(time.RFC3339Nano),
+ }, cfg), "un-elapsed cooldown must be skipped without reopening")
+ assert.True(t, shardMayWarrant(finalizeState{
+ FinalizeGeneration: 1, FinalizeRounds: 1,
+ LastFinalizedAt:
time.Now().Add(-time.Hour).UTC().Format(time.RFC3339Nano),
+ }, cfg), "past-cooldown must be reopened for the precise threshold")
+ assert.True(t, shardMayWarrant(finalizeState{FinalizeGeneration: 1,
FinalizeRounds: 1}, cfg),
+ "finalized with no cooldown clock recorded is reopened
conservatively")
+}
+
+// TestRunFinalizeScan_Gating verifies the scan pass is a safe no-op when
there are no
+// finalize-enabled groups, and skips a finalize group that has no samplers
(never
+// reaching loadTSDB).
+func TestRunFinalizeScan_Gating(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ sr := makeDataSchemaRepo("/some/dir")
+
+ // No finalize groups: no-op, no panic.
+ require.NotPanics(t, func() { sr.runFinalizeScan(nil) })
+
+ // A finalize-configured group with NO samplers is skipped before any
TSDB load.
+ setFinalizeConfigForGroup("g-no-samplers", &finalizeConfig{})
+ setFinalizeGraceForGroup("g-no-samplers", int64(5*time.Minute))
+ require.NotPanics(t, func() { sr.runFinalizeScan(nil) })
+}
+
+// TestFinalizeScan_SelectsCooledSegmentAndFinalizes closes the scanner→round
seam: it
+// builds a real TSDB, writes into a cooled (past) segment, then drives the
exact
+// storage calls scanGroup makes — SelectSegments(coolRange,
reopenClosed=true) ->
+// GetTimeRange cool re-check -> Tables() -> runFinalizeRound — and asserts
the dropped
+// trace is gone, the shard generation advanced, and segments are DecRef'd
back to
+// dormant (no open-segment leak).
+func TestFinalizeScan_SelectsCooledSegmentAndFinalizes(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+
+ // Hourly segments + long TTL so a 2h-old segment is cooled but NOT
retention-expired
+ // (SelectSegments excludes expired segments).
+ hourly := storage.IntervalRule{Unit: storage.HOUR, Num: 1}
+ opts := storage.TSDBOpts[*tsTable, option]{
+ ShardNum: 1,
+ Location: filepath.Join(tmpPath, "tab"),
+ TSTableCreator: newTSTable,
+ SegmentInterval: hourly,
+ TTL: storage.IntervalRule{Unit: storage.DAY, Num:
30},
+ Option: option{flushTimeout: 0, protector:
protector.Nop{}, mergePolicy: newDefaultMergePolicyForTesting(), decideTimeout:
time.Second},
+ }
+ require.NoError(t, os.MkdirAll(opts.Location, storage.DirPerm))
+ ctx := common.SetPosition(
+ context.WithValue(context.Background(), logger.ContextKey,
logger.GetLogger("finalize-scan-test")),
+ func(p common.Position) common.Position { p.Database = "test";
return p },
+ )
+ db, err := storage.OpenTSDB[*tsTable, option](ctx, opts, nil,
"test-group")
+ require.NoError(t, err)
+ defer db.Close()
+
+ // Create a cooled segment ~2h in the past and write traces dated
within it.
+ past := time.Now().Add(-2 * time.Hour)
+ seg, err := db.CreateSegmentIfNotExist(past)
+ require.NoError(t, err)
+ // A brand-new segment has no shard tables until one is created (they
are lazy).
+ tst, err := seg.CreateTSTableIfNotExist(common.ShardID(0))
+ require.NoError(t, err)
+
+ pastTraces := &traces{
+ traceIDs: []string{"keep1", "drop2", "keep3"},
+ timestamps: []int64{past.UnixNano(), past.UnixNano(),
past.UnixNano()},
+ tags: [][]*tagValue{
+ {{tag: "t", valueType: pbv1.ValueTypeStr, value:
[]byte("a")}},
+ {{tag: "t", valueType: pbv1.ValueTypeStr, value:
[]byte("b")}},
+ {{tag: "t", valueType: pbv1.ValueTypeStr, value:
[]byte("c")}},
+ },
+ spans: [][]byte{[]byte("s1"), []byte("s2"), []byte("s3")},
+ spanIDs: []string{"s1", "s2", "s3"},
+ }
+ tst.mustAddTraces(pastTraces, nil)
+ require.Eventually(t, func() bool {
+ s := tst.currentSnapshot()
+ if s == nil {
+ return false
+ }
+ defer s.decRef()
+ for _, pw := range s.parts {
+ if pw.mp == nil && pw.p.partMetadata.TotalCount > 0 {
+ return true
+ }
+ }
+ return false
+ }, 10*time.Second, 20*time.Millisecond)
+ seg.DecRef()
+
+ // Drive scanGroup's storage calls directly (loadTSDB is a trivial
registry lookup).
+ graceNs := int64(time.Millisecond)
+ now := time.Now().UnixNano()
+ coolRange := timestamp.TimeRange{Start: time.Unix(0, 0), End:
time.Unix(0, now-graceNs)}
+
+ // PeekSegments must surface the cooled segment's shard path WITHOUT
reopening, and
+ // the gen-0 shard must pass the reopen pre-filter.
+ peeks := db.PeekSegments(coolRange)
+ require.NotEmpty(t, peeks, "PeekSegments must return the cooled
segment")
+ sawShard := false
+ for _, pk := range peeks {
+ if pk.End.UnixNano() <= now-graceNs && len(pk.ShardPaths) > 0 {
+ sawShard = true
+ assert.True(t,
segmentMayWarrant(fs.NewLocalFileSystem(), pk.ShardPaths,
finalizeConfig{floorBytes: 1, ratio: 0.1, maxRounds: 8}),
+ "a never-finalized cooled segment must pass the
reopen pre-filter")
+ }
+ }
+ require.True(t, sawShard, "PeekSegments must expose the cooled
segment's shard path")
+
+ segs, err := db.SelectSegments(coolRange, true)
+ require.NoError(t, err)
+ require.NotEmpty(t, segs, "the cooled segment must be selected")
+
+ var ranRound bool
+ for _, s := range segs {
+ if s.GetTimeRange().End.UnixNano() > now-graceNs {
+ continue
+ }
+ tt, _ := s.Tables()
+ for _, table := range tt {
+ finalized, ferr :=
table.runFinalizeRound([]sdk.Sampler{dropOneSampler{dropID: "drop2"}}, graceNs)
+ require.NoError(t, ferr)
+ if finalized {
+ ranRound = true
+ assert.Equal(t, uint64(2),
snapshotTotalCount(table), "drop2 must be removed, leaving 2")
+ assert.Equal(t, uint64(1),
table.finalizeGenCached.Load())
+ }
+ }
+ }
+ for _, s := range segs {
+ s.DecRef()
+ }
+ assert.True(t, ranRound, "a finalize round must have run on the cooled
segment")
+}
diff --git a/banyand/trace/finalize_state.go b/banyand/trace/finalize_state.go
new file mode 100644
index 000000000..db00b989f
--- /dev/null
+++ b/banyand/trace/finalize_state.go
@@ -0,0 +1,96 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "encoding/json"
+ "path/filepath"
+
+ "github.com/pkg/errors"
+
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+// finalizeStateFilename is the per-shard finalization-sampling state document,
+// stored in the shard directory alongside snapshots.
+const finalizeStateFilename = "finalize.json"
+
+// finalizeState is the per-shard finalization-sampling state, persisted
atomically
+// in the shard directory (Rev 8: per-shard, NOT in the per-segment metadata).
It is
+// the sole authority for a shard's finalize generation, cooldown clock, round
count,
+// and arrival-based unsampled-bytes counter. The finalize worker is the only
writer
+// (serialized node-wide); readers seed the cached tsTable fields at open. A
missing
+// or corrupt file decodes to the zero value ("never finalized"), so
pre-finalize
+// shards and partial writes both fail open.
+type finalizeState struct {
+ // LastFinalizedAt is the RFC3339Nano wall-clock of the last successful
round;
+ // empty means never finalized. It is the cooldown clock.
+ LastFinalizedAt string `json:"lastFinalizedAt,omitempty"`
+ // FinalizeGeneration is the shard's monotonic current generation; 0 =
never
+ // finalized. A round targets generation gNext = FinalizeGeneration + 1
and selects
+ // every part with FinalizeGen < gNext — i.e. FinalizeGen <=
FinalizeGeneration — so
+ // each round re-samples the whole cooled set (survivors stamped at the
current
+ // generation are included). The stamp exists for crash-safety: after a
committed
+ // round both the parts and this field read gNext, so a replay (which
recomputes
+ // gNext) excludes them.
+ FinalizeGeneration uint64 `json:"finalizeGeneration,omitempty"`
+ // UnsampledBytes accumulates uncompressed span bytes of parts that
arrived into
+ // this shard after it cooled/was-finalized; it gates the re-round
threshold and
+ // resets to 0 after a successful round.
+ UnsampledBytes int64 `json:"unsampledBytes,omitempty"`
+ // FinalizeRounds counts rounds run over this shard's lifetime; the
hard cap.
+ FinalizeRounds int `json:"finalizeRounds,omitempty"`
+ // Terminal is set once FinalizeRounds hits max_finalize_rounds; the
shard is
+ // then never re-scanned (all future late data is an accepted miss).
+ Terminal bool `json:"terminal,omitempty"`
+}
+
+// readFinalizeState reads the shard's finalize state. A missing, empty, or
corrupt
+// file yields the zero value (never finalized): finalize is best-effort and
must
+// never panic or block on its own state, so any read/parse error fails open.
+func readFinalizeState(fileSystem fs.FileSystem, shardRoot string)
finalizeState {
+ data, err := fileSystem.Read(filepath.Join(shardRoot,
finalizeStateFilename))
+ if err != nil || len(data) == 0 {
+ return finalizeState{}
+ }
+ var st finalizeState
+ if unmarshalErr := json.Unmarshal(data, &st); unmarshalErr != nil {
+ return finalizeState{}
+ }
+ return st
+}
+
+// writeFinalizeState atomically persists st to the shard directory via
temp+rename
+// (fileSystem.WriteAtomic fsyncs the file then the parent dir), so a crash
leaves
+// either the previous valid state or none — never a torn document.
+func writeFinalizeState(fileSystem fs.FileSystem, shardRoot string, st
finalizeState) error {
+ data, err := json.Marshal(st)
+ if err != nil {
+ return errors.WithMessage(err, "cannot marshal finalize state")
+ }
+ statePath := filepath.Join(shardRoot, finalizeStateFilename)
+ n, err := fileSystem.WriteAtomic(data, statePath, storage.FilePerm)
+ if err != nil {
+ return errors.WithMessage(err, "cannot write finalize state")
+ }
+ if n != len(data) {
+ return errors.Errorf("short write to %s: got %d want %d",
statePath, n, len(data))
+ }
+ return nil
+}
diff --git a/banyand/trace/finalize_state_test.go
b/banyand/trace/finalize_state_test.go
new file mode 100644
index 000000000..193f23f68
--- /dev/null
+++ b/banyand/trace/finalize_state_test.go
@@ -0,0 +1,104 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "encoding/json"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+// TestFinalizeState_RoundTrip verifies write then read returns the same state.
+func TestFinalizeState_RoundTrip(t *testing.T) {
+ fileSystem := fs.NewLocalFileSystem()
+ dir := t.TempDir()
+
+ want := finalizeState{
+ LastFinalizedAt: time.Unix(0,
1_700_000_000_000_000_000).UTC().Format(time.RFC3339Nano),
+ FinalizeGeneration: 5,
+ UnsampledBytes: 12345,
+ FinalizeRounds: 3,
+ Terminal: true,
+ }
+ require.NoError(t, writeFinalizeState(fileSystem, dir, want))
+
+ got := readFinalizeState(fileSystem, dir)
+ assert.Equal(t, want, got)
+}
+
+// TestFinalizeState_MissingFileIsZero verifies a shard with no finalize file
loads
+// as never-finalized (the pre-finalize / fresh-shard common case).
+func TestFinalizeState_MissingFileIsZero(t *testing.T) {
+ fileSystem := fs.NewLocalFileSystem()
+ dir := t.TempDir()
+
+ got := readFinalizeState(fileSystem, dir)
+ assert.Equal(t, finalizeState{}, got, "missing finalize file must
decode to zero state")
+ assert.Zero(t, got.FinalizeGeneration)
+}
+
+// TestFinalizeState_CorruptIsZero verifies a corrupt finalize file fails open
to the
+// zero value rather than panicking (finalize must never crash the node).
+func TestFinalizeState_CorruptIsZero(t *testing.T) {
+ fileSystem := fs.NewLocalFileSystem()
+ dir := t.TempDir()
+
+ _, err := fileSystem.Write([]byte("{not valid json"),
filepath.Join(dir, finalizeStateFilename), storage.FilePerm)
+ require.NoError(t, err)
+
+ assert.Equal(t, finalizeState{}, readFinalizeState(fileSystem, dir),
+ "corrupt finalize file must fail open to zero state")
+}
+
+// TestPartMetadata_FinalizeGenRoundTrip verifies the FinalizeGen field
survives a
+// JSON round-trip and that old-format metadata (no finalizeGen key) loads as
0.
+func TestPartMetadata_FinalizeGenRoundTrip(t *testing.T) {
+ pm := partMetadata{
+ CompressedSizeBytes: 100,
+ UncompressedSpanSizeBytes: 200,
+ TotalCount: 3,
+ BlocksCount: 1,
+ MinTimestamp: 10,
+ MaxTimestamp: 20,
+ FinalizeGen: 7,
+ }
+ data, err := json.Marshal(&pm)
+ require.NoError(t, err)
+
+ var got partMetadata
+ require.NoError(t, json.Unmarshal(data, &got))
+ assert.Equal(t, uint64(7), got.FinalizeGen, "FinalizeGen must survive
round-trip")
+
+ // Old-format metadata (no finalizeGen key) loads as unstamped (0).
+ var old partMetadata
+ require.NoError(t,
json.Unmarshal([]byte(`{"compressedSizeBytes":1,"totalCount":1,"minTimestamp":1,"maxTimestamp":2}`),
&old))
+ assert.Zero(t, old.FinalizeGen, "old-format part metadata must load as
unstamped")
+
+ // omitempty: a zero FinalizeGen is not serialized.
+ zeroPM := partMetadata{TotalCount: 1, MinTimestamp: 1, MaxTimestamp: 2}
+ zeroData, err := json.Marshal(&zeroPM)
+ require.NoError(t, err)
+ assert.NotContains(t, string(zeroData), "finalizeGen", "zero
FinalizeGen must be omitted")
+}
diff --git a/banyand/trace/finalizer.go b/banyand/trace/finalizer.go
new file mode 100644
index 000000000..dad5b7f72
--- /dev/null
+++ b/banyand/trace/finalizer.go
@@ -0,0 +1,182 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "time"
+
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+)
+
+// runFinalizeRound force-merges this shard's cooled, un-finalized parts
through the
+// sampler chain (the finalization-sampling backstop). It reuses the hot merge
path
+// (mergePartsThenSendIntroduction → mergeParts + per-sidx Merge + the
serialized
+// introducer loop, OQ6) with a pre-built filter and a finalize-generation
override,
+// so the core AND sidx follow the exact same code the hot merge uses. It NEVER
+// acquires mergeMaxConcurrencyCh (that lives in the merge-lane worker, not
here), so
+// finalize compute cannot starve the hot merge lanes.
+//
+// Selection is all not-hot (by finalize_grace), not-in-flight parts whose
FinalizeGen
+// is below the round's target generation Gnext=G+1 (by the invariant that
nothing is
+// stamped above the stored generation, this is every cooled part). The round
is a full
+// re-sample of the cooled set; re-sampling already-kept survivors through a
+// deterministic sampler is idempotent (DD11), and rounds are hard-capped by
the
+// scanner (max_finalize_rounds), so the total rewrite volume is bounded.
+//
+// It returns (true, nil) when a round committed, (false, nil) when there was
nothing
+// to do or it yielded under pressure, and (false, err) on a merge error
(fail-open:
+// the segment is left intact and the scanner retries on a later tick). It is
invoked
+// only by the finalize scanner while the owning segment is incRef-held, so the
+// tsTable's introducer loop is alive to apply the introduction.
+func (tst *tsTable) runFinalizeRound(samplers []sdk.Sampler, graceNs int64)
(bool, error) {
+ if len(samplers) == 0 {
+ return false, nil
+ }
+ // Resource gate: never run finalize compute under memory pressure
(constraint 1).
+ if tst.pm != nil && tst.pm.State() == protector.StateHigh {
+ return false, nil
+ }
+ closeCh := tst.loopCloser.CloseNotify()
+ select {
+ case <-closeCh:
+ return false, nil
+ default:
+ }
+
+ cur := tst.currentSnapshot()
+ if cur == nil {
+ return false, nil
+ }
+ defer cur.decRef()
+
+ gNext := tst.finalizeGenCached.Load() + 1
+ now := time.Now().UnixNano()
+ // Snapshot the counter at round start. On commit we subtract exactly
this much
+ // rather than storing 0, so bytes that a concurrent flush accounts
DURING the round
+ // (late arrivals not part of this round's merge) are preserved for the
next round.
+ // unsampledBytes is only ever added to elsewhere and reset only here
(serialized),
+ // so the value never drops below startBytes before the subtract.
+ startBytes := tst.unsampledBytes.Load()
+
+ // Select AND pin candidate parts atomically under a single write lock.
A concurrent
+ // hot-merge dispatch (dispatchAllMerges) is a second selector of the
same parts; if
+ // selection and pinning were split (RLock check, then Lock add) both
could pick the
+ // same part and merge it twice, and since snapshot.remove tolerates an
absent id the
+ // two outputs would both survive — duplicating traces. Doing
check+pin+incRef in one
+ // Lock (and a matching re-check on the dispatcher side) makes the pin
exclusive.
+ // Selection: cooled (non-hot by finalize_grace), not already
in-flight, and below the
+ // round generation (excludes a crashed round's stamped outputs — DD6).
+ var parts []*partWrapper
+ var needBytes uint64
+ tst.inFlightMu.Lock()
+ if tst.inFlight == nil {
+ tst.inFlight = make(map[uint64]struct{})
+ }
+ for _, pw := range cur.parts {
+ if pw.mp != nil || pw.p.partMetadata.TotalCount < 1 {
+ continue
+ }
+ if _, inFlight := tst.inFlight[pw.ID()]; inFlight {
+ continue
+ }
+ if pw.p.partMetadata.FinalizeGen >= gNext {
+ continue
+ }
+ if pw.p.partMetadata.MaxTimestamp > now-graceNs {
+ continue
+ }
+ pw.incRef()
+ tst.inFlight[pw.ID()] = struct{}{}
+ parts = append(parts, pw)
+ needBytes += pw.p.partMetadata.CompressedSizeBytes
+ }
+ tst.inFlightMu.Unlock()
+
+ if len(parts) == 0 {
+ // Nothing eligible this round (all parts still hot, or already
finalized).
+ // Do NOT advance the generation or reset the counter.
+ return false, nil
+ }
+ // Unpin + release on EVERY return path below (disk gate, merge error,
success).
+ defer func() {
+ tst.inFlightMu.Lock()
+ for _, pw := range parts {
+ delete(tst.inFlight, pw.ID())
+ }
+ tst.inFlightMu.Unlock()
+ for _, pw := range parts {
+ pw.decRef()
+ }
+ }()
+
+ // Disk-headroom gate: a full-shard rewrite transiently needs ~the
selected parts'
+ // compressed size in extra space. Skip the round (an accepted miss)
rather than risk
+ // pushing the hot write path into disk-full — finalize must never
starve the core.
+ if tst.freeDiskSpace(tst.root) < needBytes {
+ return false, nil
+ }
+
+ // Build the finalize filter. Unlike the hot path there is no
isMergeHot gate here
+ // (eligibility was already decided above with finalize_grace); the
chain fails open
+ // on any Decide error, retaining the whole batch.
+ chain := newMergeChain(tst.group, "", samplers,
tst.option.decideTimeoutCircuitBreak)
+ filter := &mergeFilter{
+ chain: chain,
+ timeout: tst.option.decideTimeout,
+ stageBudget: resolveStageBudget(tst.option),
+ forceSlow: len(chain.projection.Tags) > 0,
+ }
+
+ merged := make(map[uint64]struct{}, len(parts))
+ for _, pw := range parts {
+ merged[pw.ID()] = struct{}{}
+ }
+
+ gen := gNext
+ if _, err := tst.mergePartsThenSendIntroduction(
+ snapshotCreatorMerger, parts, merged, tst.mergeCh, closeCh,
+ mergeTypeFinalize, mergeLaneFinalize, &mergeOverrides{filter:
filter, finalizeGen: &gen},
+ ); err != nil {
+ // Fail-open: leave the segment intact; the scanner retries on
a later tick.
+ return false, err
+ }
+
+ // Commit the per-shard finalize state AFTER the introduction (DD6
ordering). The
+ // in-memory cache and counter are mutated ONLY after the state file is
durably
+ // written: if persist failed but we still bumped finalizeGenCached,
the next round
+ // would use gNext+1 and re-include (re-sample) the parts this round
just stamped
+ // gNext. On failure we leave cache == disk (still G) and return the
error; the
+ // on-disk parts are stamped gNext, so the retry's predicate
(finalizeGen < G+1)
+ // excludes them — no double-sample, only the round bookkeeping is
retried.
+ // The counter subtracts exactly the bytes counted before the round
started, so a
+ // concurrent flush during the round keeps its contribution for the
next round.
+ remaining := max(tst.unsampledBytes.Load()-startBytes, 0)
+ st := readFinalizeState(tst.fileSystem, tst.root)
+ st.FinalizeGeneration = gNext
+ st.LastFinalizedAt = time.Now().UTC().Format(time.RFC3339Nano)
+ st.FinalizeRounds++
+ st.UnsampledBytes = remaining
+ if err := writeFinalizeState(tst.fileSystem, tst.root, st); err != nil {
+ tst.l.Warn().Err(err).Str("group", tst.group).Msg("cannot
persist finalize state; leaving generation unchanged")
+ return false, err
+ }
+ tst.finalizeGenCached.Store(gNext)
+ tst.unsampledBytes.Add(-startBytes)
+ return true, nil
+}
diff --git a/banyand/trace/finalizer_test.go b/banyand/trace/finalizer_test.go
new file mode 100644
index 000000000..16fa3b5ac
--- /dev/null
+++ b/banyand/trace/finalizer_test.go
@@ -0,0 +1,394 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+// dropOneSampler is a minimal deterministic sampler that drops exactly one
trace id.
+type dropOneSampler struct{ dropID string }
+
+func (dropOneSampler) Kind() sdk.Kind { return sdk.KindSampler }
+func (dropOneSampler) Project() sdk.Projection { return sdk.Projection{} }
+func (dropOneSampler) Close() error { return nil }
+func (s dropOneSampler) Decide(b *sdk.TraceBatch) (sdk.Verdict, error) {
+ keep := make([]bool, len(b.Traces))
+ for i := range b.Traces {
+ keep[i] = b.Traces[i].TraceID != s.dropID
+ }
+ return sdk.Verdict{Keep: keep}, nil
+}
+
+// dropPrefixSampler drops every trace whose id starts with the prefix. Used
to prove
+// MERGE and FINALIZE compose: the same registered sampler runs in both events.
+type dropPrefixSampler struct{ prefix string }
+
+func (dropPrefixSampler) Kind() sdk.Kind { return sdk.KindSampler }
+func (dropPrefixSampler) Project() sdk.Projection { return sdk.Projection{} }
+func (dropPrefixSampler) Close() error { return nil }
+func (s dropPrefixSampler) Decide(b *sdk.TraceBatch) (sdk.Verdict, error) {
+ keep := make([]bool, len(b.Traces))
+ for i := range b.Traces {
+ keep[i] = !strings.HasPrefix(b.Traces[i].TraceID, s.prefix)
+ }
+ return sdk.Verdict{Keep: keep}, nil
+}
+
+// tracesWithIDs builds a minimal traces fixture (one ancient-timestamp span
per id).
+func tracesWithIDs(ids ...string) *traces {
+ ts := &traces{}
+ for _, id := range ids {
+ ts.traceIDs = append(ts.traceIDs, id)
+ ts.timestamps = append(ts.timestamps, int64(1))
+ ts.tags = append(ts.tags, []*tagValue{{tag: "t", valueType:
pbv1.ValueTypeStr, value: []byte("v")}})
+ ts.spans = append(ts.spans, []byte("span-"+id))
+ ts.spanIDs = append(ts.spanIDs, "span-"+id)
+ }
+ return ts
+}
+
+func snapshotTotalCount(tst *tsTable) uint64 {
+ s := tst.currentSnapshot()
+ if s == nil {
+ return 0
+ }
+ defer s.decRef()
+ var total uint64
+ for _, pw := range s.parts {
+ if pw.mp == nil {
+ total += pw.p.partMetadata.TotalCount
+ }
+ }
+ return total
+}
+
+func oneTraceFixture(id string, ts int64) *traces {
+ return &traces{
+ traceIDs: []string{id},
+ timestamps: []int64{ts},
+ tags: [][]*tagValue{
+ {{tag: "tag1", valueType: pbv1.ValueTypeStr, value:
[]byte("v"), valueArr: nil}},
+ },
+ spans: [][]byte{[]byte("span-" + id)},
+ spanIDs: []string{"span-" + id},
+ }
+}
+
+// TestMergeParts_FinalizeGenPropagation covers the four DD1.C1 cases for the
finalize
+// generation stamp threaded through mergeParts. Each output is re-read from
disk
+// (mustOpenFilePart -> mustReadMetadata) to assert the ON-DISK value
survives, i.e.
+// the stamp is written before pm.mustWriteMetadata (crash-safe across a
restart).
+func TestMergeParts_FinalizeGenPropagation(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+ fileSystem := fs.NewLocalFileSystem()
+ tst := &tsTable{pm: protector.Nop{}, fileSystem: fileSystem, root:
tmpPath}
+
+ var nextID uint64
+ // flushPart writes a single-trace part and returns its wrapper with
FinalizeGen set
+ // in-memory to gen (mergeParts min-propagates from the in-memory input
value).
+ flushPart := func(id string, gen uint64) *partWrapper {
+ nextID++
+ pid := nextID
+ mp := generateMemPart()
+ mp.mustInitFromTraces(oneTraceFixture(id, int64(pid)))
+ mp.mustFlush(fileSystem, partPath(tmpPath, pid))
+ releaseMemPart(mp)
+ p := mustOpenFilePart(pid, tmpPath, fileSystem)
+ p.partMetadata.ID = pid
+ p.partMetadata.FinalizeGen = gen
+ return newPartWrapper(nil, p)
+ }
+ // mergeAndReadGen merges parts with the given override, then re-opens
the output
+ // part from disk and returns its persisted FinalizeGen.
+ mergeAndReadGen := func(parts []*partWrapper, override *uint64) uint64 {
+ nextID++
+ outID := nextID
+ closeCh := make(chan struct{})
+ out, _, err := tst.mergeParts(fileSystem, closeCh, parts,
outID, tmpPath, nil, override)
+ close(closeCh)
+ require.NoError(t, err)
+ require.NotNil(t, out)
+ gotID := out.ID()
+ out.decRef()
+ for _, pw := range parts {
+ pw.decRef()
+ }
+ reopened := mustOpenFilePart(gotID, tmpPath, fileSystem)
+ return reopened.partMetadata.FinalizeGen
+ }
+
+ // (a) hot merge (nil override) of two gen-G parts => output gen G
(min), so a hot
+ // merge inside a finalized segment does NOT un-finalize.
+ assert.Equal(t, uint64(3),
mergeAndReadGen([]*partWrapper{flushPart("a1", 3), flushPart("a2", 3)}, nil),
+ "two gen-3 parts must merge to gen 3 (min)")
+
+ // (b) hot merge of a gen-G part + an unstamped (0) late part => output
0 (min), so
+ // the merged part is selectable by the next finalize round.
+ assert.Equal(t, uint64(0),
mergeAndReadGen([]*partWrapper{flushPart("b1", 5), flushPart("b2", 0)}, nil),
+ "gen-5 + unstamped must merge to gen 0 (min)")
+
+ // (c) a finalize round (override=&Gnew) stamps the output at Gnew on
disk; the
+ // re-open proves the value is persisted (crash-safe), and Gnew <
FinalizeGeneration
+ // would be false on replay so it is not re-selected.
+ gnew := uint64(7)
+ assert.Equal(t, uint64(7),
mergeAndReadGen([]*partWrapper{flushPart("c1", 0), flushPart("c2", 0)}, &gnew),
+ "finalize override must stamp Gnew on disk")
+
+ // (d) a lone unstamped part stays 0 under nil override.
+ assert.Equal(t, uint64(0),
mergeAndReadGen([]*partWrapper{flushPart("d1", 0)}, nil),
+ "a lone unstamped part must stay gen 0")
+}
+
+// TestRunFinalizeRound_DropsAndStamps drives a full finalize round on a
running
+// tsTable: writes traces, waits for the flush, then finalizes through a
drop-one
+// sampler and asserts the dropped trace is gone, the generation advanced, and
the
+// per-shard finalize state is persisted.
+func TestRunFinalizeRound_DropsAndStamps(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+ fileSystem := fs.NewLocalFileSystem()
+
+ tst, err := newTSTable(
+ fileSystem, tmpPath,
+ common.Position{Database: "fg"},
+ logger.GetLogger("finalize-round-test"),
+ timestamp.TimeRange{},
+ option{
+ flushTimeout: 0,
+ mergePolicy: newDefaultMergePolicyForTesting(),
+ protector: protector.Nop{},
+ decideTimeout: time.Second,
+ },
+ nil,
+ )
+ require.NoError(t, err)
+ defer tst.Close()
+
+ tst.mustAddTraces(tsTS1, nil)
+ // Wait for the mem part to flush to a file part (mp == nil).
+ require.Eventually(t, func() bool {
+ s := tst.currentSnapshot()
+ if s == nil {
+ return false
+ }
+ defer s.decRef()
+ for _, pw := range s.parts {
+ if pw.mp == nil && pw.p.partMetadata.TotalCount > 0 {
+ return true
+ }
+ }
+ return false
+ }, 10*time.Second, 20*time.Millisecond, "flushed file part must appear")
+
+ require.Equal(t, uint64(3), snapshotTotalCount(tst), "3 traces before
finalize")
+
+ // finalize_grace tiny so the cooled parts (timestamp 1) are non-hot.
+ finalized, roundErr :=
tst.runFinalizeRound([]sdk.Sampler{dropOneSampler{dropID: "trace2"}},
int64(time.Millisecond))
+ require.NoError(t, roundErr)
+ require.True(t, finalized, "a round must have committed")
+
+ assert.Equal(t, uint64(2), snapshotTotalCount(tst), "trace2 must be
dropped, leaving 2")
+ assert.Equal(t, uint64(1), tst.finalizeGenCached.Load(), "generation
must advance to 1")
+ assert.Zero(t, tst.unsampledBytes.Load(), "counter must reset after a
round")
+
+ st := readFinalizeState(fileSystem, tmpPath)
+ assert.Equal(t, uint64(1), st.FinalizeGeneration, "persisted generation
must be 1")
+ assert.Equal(t, 1, st.FinalizeRounds, "persisted round count must be 1")
+ assert.NotEmpty(t, st.LastFinalizedAt, "lastFinalizedAt must be set")
+}
+
+// TestRunFinalizeRound_ReplayExcludesStampedParts reproduces the DD6 crash
window: a
+// part is stamped at the round generation on disk, but the per-shard state
(and cache)
+// still reads the OLD generation (as if the process died after introduce,
before the
+// state persist). On the next round the predicate finalizeGen >= gNext must
EXCLUDE the
+// stamped part, so no round runs and nothing is re-sampled.
+func TestRunFinalizeRound_ReplayExcludesStampedParts(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+ fileSystem := fs.NewLocalFileSystem()
+
+ tst, err := newTSTable(
+ fileSystem, tmpPath, common.Position{Database: "rp"},
+ logger.GetLogger("finalize-replay-test"), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy:
newDefaultMergePolicyForTesting(), protector: protector.Nop{}, decideTimeout:
time.Second},
+ nil,
+ )
+ require.NoError(t, err)
+ defer tst.Close()
+
+ tst.mustAddTraces(tsTS1, nil)
+ require.Eventually(t, func() bool {
+ s := tst.currentSnapshot()
+ if s == nil {
+ return false
+ }
+ defer s.decRef()
+ for _, pw := range s.parts {
+ if pw.mp == nil && pw.p.partMetadata.TotalCount > 0 {
+ return true
+ }
+ }
+ return false
+ }, 10*time.Second, 20*time.Millisecond)
+
+ // Round 1 stamps the output at gen 1 (on disk) and persists state gen
1.
+ finalized, roundErr :=
tst.runFinalizeRound([]sdk.Sampler{dropOneSampler{dropID: "trace2"}},
int64(time.Millisecond))
+ require.NoError(t, roundErr)
+ require.True(t, finalized)
+ countAfterFirst := snapshotTotalCount(tst)
+
+ // Simulate the crash window: the parts stay stamped gen 1 on disk, but
the state +
+ // cache revert to gen 0 (the persist that would have recorded gen 1
never happened).
+ tst.finalizeGenCached.Store(0)
+ require.NoError(t, writeFinalizeState(fileSystem, tmpPath,
finalizeState{}))
+
+ // Replay: the gen-1 output part must be excluded (finalizeGen 1 >=
gNext 1), so no
+ // round runs — no double-sample of already-committed data.
+ finalized2, roundErr2 :=
tst.runFinalizeRound([]sdk.Sampler{dropOneSampler{dropID: "trace2"}},
int64(time.Millisecond))
+ require.NoError(t, roundErr2)
+ assert.False(t, finalized2, "a part already stamped at the round
generation must be excluded on replay")
+ assert.Equal(t, uint64(0), tst.finalizeGenCached.Load(), "excluded
replay must not advance the generation")
+ assert.Equal(t, countAfterFirst, snapshotTotalCount(tst), "replay must
not re-sample committed data")
+}
+
+// TestFinalizeAndMerge_Compose proves the in-merge filter
(PIPELINE_EVENT_MERGE) and
+// finalization sampling (PIPELINE_EVENT_FINALIZE) compose on ONE running
shard, using
+// the same registered sampler (DD11): a real hot merge drops its targets
during the
+// merge, a never-merged part is left for finalize, and a finalize round drops
that
+// part's target while re-keeping the merge survivors (no double-drop). The
auto-merge
+// loop is disabled (maxFanOutSize=0) so the merge below is the only one and
the test
+// is deterministic; the manual merge bypasses the policy, exercising the
exact hot
+// merge path (mergePartsThenSendIntroduction with nil override).
+func TestFinalizeAndMerge_Compose(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "cg"
+ deregister := registerSampler(group, dropPrefixSampler{prefix: "drop"})
+ defer deregister()
+
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+ fileSystem := fs.NewLocalFileSystem()
+
+ tst, err := newTSTable(
+ fileSystem, tmpPath,
+ common.Position{Database: group},
+ logger.GetLogger("finalize-merge-compose"),
+ timestamp.TimeRange{},
+ option{
+ flushTimeout: 0,
+ mergePolicy: newMergePolicy(3, 1,
run.Bytes(0)), // maxFanOutSize=0 disables auto-merge
+ protector: protector.Nop{},
+ decideTimeout: time.Second,
+ decideTimeoutCircuitBreak: 3,
+ mergeGraceDefault: time.Millisecond, // ancient
parts are never hot
+ nativePipelineEnabled: true, //
activates the in-merge MERGE filter
+ },
+ nil,
+ )
+ require.NoError(t, err)
+ defer tst.Close()
+
+ // waitForFileParts blocks until the snapshot holds exactly n file
parts.
+ waitForFileParts := func(n int) {
+ require.Eventually(t, func() bool {
+ s := tst.currentSnapshot()
+ if s == nil {
+ return false
+ }
+ defer s.decRef()
+ cnt := 0
+ for _, pw := range s.parts {
+ if pw.mp == nil && pw.p.partMetadata.TotalCount
> 0 {
+ cnt++
+ }
+ }
+ return cnt == n
+ }, 10*time.Second, 20*time.Millisecond)
+ }
+ collectFileParts := func() []*partWrapper {
+ s := tst.currentSnapshot()
+ require.NotNil(t, s)
+ defer s.decRef()
+ var pws []*partWrapper
+ for _, pw := range s.parts {
+ if pw.mp == nil && pw.p.partMetadata.TotalCount > 0 {
+ pw.incRef()
+ pws = append(pws, pw)
+ }
+ }
+ return pws
+ }
+
+ // Two separately-flushed parts, each with one drop-target and one
keeper.
+ tst.mustAddTraces(tracesWithIDs("dropA", "keepB"), nil)
+ waitForFileParts(1)
+ tst.mustAddTraces(tracesWithIDs("dropC", "keepD"), nil)
+ waitForFileParts(2)
+ require.Equal(t, uint64(4), snapshotTotalCount(tst), "4 traces across 2
parts before any sampling")
+
+ // A real hot merge (MERGE event): the in-merge sampler drops dropA +
dropC.
+ parts := collectFileParts()
+ require.Len(t, parts, 2)
+ merged := make(map[uint64]struct{}, len(parts))
+ for _, pw := range parts {
+ merged[pw.ID()] = struct{}{}
+ }
+ closeCh := make(chan struct{})
+ _, mErr := tst.mergePartsThenSendIntroduction(
+ snapshotCreatorMerger, parts, merged, tst.mergeCh, closeCh,
mergeTypeFile, mergeLaneFast, nil,
+ )
+ close(closeCh)
+ require.NoError(t, mErr)
+ for _, pw := range parts {
+ pw.decRef()
+ }
+ assert.Equal(t, uint64(2), snapshotTotalCount(tst), "MERGE must drop
dropA + dropC, leaving keepB + keepD")
+ assert.Zero(t, tst.finalizeGenCached.Load(), "a MERGE must not advance
the finalize generation")
+
+ // A third part that never merges — the case finalize exists to
backstop.
+ tst.mustAddTraces(tracesWithIDs("dropE", "keepF"), nil)
+ waitForFileParts(2) // merged part + part3
+ require.Equal(t, uint64(4), snapshotTotalCount(tst), "keepB + keepD +
dropE + keepF")
+
+ // FINALIZE event: drops dropE from the never-merged part and re-keeps
the MERGE
+ // survivors (deterministic sampler => no double-drop). Advances the
generation.
+ finalized, fErr :=
tst.runFinalizeRound([]sdk.Sampler{dropPrefixSampler{prefix: "drop"}},
int64(time.Millisecond))
+ require.NoError(t, fErr)
+ require.True(t, finalized, "finalize must have run")
+ assert.Equal(t, uint64(3), snapshotTotalCount(tst), "keepB + keepD +
keepF survive; dropE dropped, no keeper double-dropped")
+ assert.Equal(t, uint64(1), tst.finalizeGenCached.Load(), "finalize must
advance the generation to 1")
+}
diff --git a/banyand/trace/flusher.go b/banyand/trace/flusher.go
index 2ae71f54c..dde0c99a2 100644
--- a/banyand/trace/flusher.go
+++ b/banyand/trace/flusher.go
@@ -149,7 +149,7 @@ func (tst *tsTable) mergeMemParts(snp *snapshot, mergeCh
chan *mergerIntroductio
// merge memory must not be closed by the tsTable.close
closeCh := make(chan struct{})
newPart, err :=
tst.mergePartsThenSendIntroduction(snapshotCreatorMergedFlusher, memParts,
- currentMergedIDs, mergeCh, closeCh, mergeTypeMem, "")
+ currentMergedIDs, mergeCh, closeCh, mergeTypeMem, "",
nil)
close(closeCh)
if err != nil {
if errors.Is(err, errClosed) {
diff --git a/banyand/trace/introducer.go b/banyand/trace/introducer.go
index b72047fd0..78509ee41 100644
--- a/banyand/trace/introducer.go
+++ b/banyand/trace/introducer.go
@@ -324,7 +324,29 @@ func (tst *tsTable) introducePart(nextIntroduction
*introduction, epoch uint64)
}
}
+// accountUnsampledFlushed is the arrival-based unsampled-bytes accounting for
+// finalization sampling. Once this shard has been finalized at least once
(cached
+// generation > 0), a newly-flushed part is new unsampled data arriving into a
cooled
+// segment; accumulate its uncompressed span bytes so the finalize scanner can
decide
+// whether another round is warranted. It performs O(1) atomic loads/adds over
already
+// in-memory part metadata and takes no filesystem — there is NO metadata I/O
on the
+// hot flush path (Phase 3 acceptance).
+func (tst *tsTable) accountUnsampledFlushed(flushed map[uint64]*partWrapper) {
+ if tst.finalizeGenCached.Load() == 0 {
+ return
+ }
+ var newBytes int64
+ for _, pw := range flushed {
+ newBytes += int64(pw.p.partMetadata.UncompressedSpanSizeBytes)
+ }
+ if newBytes > 0 {
+ tst.unsampledBytes.Add(newBytes)
+ }
+}
+
func (tst *tsTable) introduceFlushed(nextIntroduction *flusherIntroduction,
epoch uint64) {
+ tst.accountUnsampledFlushed(nextIntroduction.flushed)
+
// Create generic transaction
txn := snapshotpkg.NewTransaction()
defer txn.Release()
@@ -375,6 +397,10 @@ func (tst *tsTable) introduceFlushed(nextIntroduction
*flusherIntroduction, epoc
// The snapshots are updated atomically so the syncer can always find
// the corresponding index once a flushed trace part becomes visible.
func (tst *tsTable) introduceFlushedForSync(nextIntroduction
*flusherIntroduction, epoch uint64) {
+ // No unsampled-bytes accounting here: on the sync path these parts are
handed off
+ // to other nodes rather than retained locally for finalization, so
counting them
+ // would inflate the counter and trigger no-op finalize rounds (finding
#9).
+
// Create generic transaction
txn := snapshotpkg.NewTransaction()
defer txn.Release()
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index d9c4e2b9c..ac58bab03 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -41,10 +41,12 @@ import (
var mergeMaxConcurrencyCh = make(chan struct{}, cgroups.CPUs())
const (
- mergeTypeMem = "mem"
- mergeTypeFile = "file"
- mergeLaneFast = "fast"
- mergeLaneSlow = "slow"
+ mergeTypeMem = "mem"
+ mergeTypeFile = "file"
+ mergeTypeFinalize = "finalize"
+ mergeLaneFast = "fast"
+ mergeLaneSlow = "slow"
+ mergeLaneFinalize = "finalize"
)
const defaultSmallMergeThreshold = 32 << 20 // 32MB fallback
@@ -225,6 +227,27 @@ func (tst *tsTable) dispatchAllMerges(threshold uint64,
fastCh, slowCh chan *mer
if tst.inFlight == nil {
tst.inFlight = make(map[uint64]struct{})
}
+ // Re-check membership under the lock before pinning.
getPartsToMerge already
+ // skips in-flight parts, but the finalize scanner is a second,
concurrent part
+ // selector: it may have pinned one of these parts in the
window between
+ // getPartsToMerge's RLock and this Lock. Pinning it here
anyway would let both
+ // actors merge the same part, and since snapshot.remove
tolerates an absent id
+ // both merge outputs would survive — duplicating traces. On
conflict, abandon
+ // this dispatch cycle (the next flush trigger re-dispatches).
+ conflict := false
+ for _, pw := range dst {
+ if _, inFlight := tst.inFlight[pw.ID()]; inFlight {
+ conflict = true
+ break
+ }
+ }
+ if conflict {
+ tst.inFlightMu.Unlock()
+ for _, pw := range dst {
+ pw.decRef()
+ }
+ return false
+ }
for _, pw := range dst {
tst.inFlight[pw.ID()] = struct{}{}
}
@@ -288,7 +311,7 @@ func (tst *tsTable) mergeLaneWorker(ch chan
*mergeDispatchRequest, merges chan *
tst.incTotalMergeLoopStarted(1)
_, mergeErr := tst.mergePartsThenSendIntroduction(
snapshotCreatorMerger, req.parts, req.toBeMerged,
merges,
- tst.loopCloser.CloseNotify(), req.typ, req.lane,
+ tst.loopCloser.CloseNotify(), req.typ, req.lane, nil,
)
tst.incTotalMergeLoopFinished(1)
<-mergeMaxConcurrencyCh
@@ -315,32 +338,69 @@ func (tst *tsTable) releaseDispatchRequest(req
*mergeDispatchRequest) {
}
}
+// mergeOverrides lets the finalize round drive mergePartsThenSendIntroduction
with a
+// pre-built sampler filter (bypassing the hot-path isMergeHot/merge_grace
build) and a
+// finalize-generation stamp for the output part. A nil *mergeOverrides means
the hot /
+// flusher path: build the filter from merge-time rules and min-propagate
finalizeGen.
+type mergeOverrides struct {
+ filter *mergeFilter
+ finalizeGen *uint64
+}
+
+// buildHotMergeFilter builds the in-merge retention filter for the hot /
flusher merge
+// path from the merge-time rules (registered samplers + merge_grace
maturity). It
+// returns nil when the native pipeline is off, no samplers are registered, or
the merge
+// is still hot (parts younger than merge_grace), in which case the merge runs
unfiltered.
+func (tst *tsTable) buildHotMergeFilter(parts []*partWrapper) *mergeFilter {
+ if !tst.option.nativePipelineEnabled {
+ return nil
+ }
+ samplers := lookupSamplers(tst.group)
+ if len(samplers) == 0 {
+ return nil
+ }
+ // The sampler set is shared with FINALIZE (DD11); only filter hot
merges when the
+ // MERGE event is actually enabled for this group, so a FINALIZE-only
config does
+ // not silently filter hot merges.
+ if !mergeEventEnabledForGroup(tst.group) {
+ return nil
+ }
+ graceNs := lookupMergeGrace(tst.group)
+ if graceNs <= 0 {
+ graceNs = int64(tst.option.mergeGraceDefault)
+ }
+ if isMergeHot(parts, graceNs, time.Now().UnixNano()) {
+ return nil
+ }
+ chain := newMergeChain(tst.group, "", samplers,
tst.option.decideTimeoutCircuitBreak)
+ return &mergeFilter{
+ chain: chain,
+ timeout: tst.option.decideTimeout,
+ stageBudget: resolveStageBudget(tst.option),
+ forceSlow: len(chain.projection.Tags) > 0,
+ }
+}
+
func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator,
parts []*partWrapper, merged map[uint64]struct{}, merges chan
*mergerIntroduction,
- closeCh <-chan struct{}, typ string, lane string,
+ closeCh <-chan struct{}, typ string, lane string, ov *mergeOverrides,
) (*partWrapper, error) {
reservedSpace := tst.reserveSpace(parts)
defer releaseDiskSpace(reservedSpace)
start := time.Now()
newPartID := atomic.AddUint64(&tst.curPartID, 1)
var filter *mergeFilter
- if tst.option.nativePipelineEnabled {
- if samplers := lookupSamplers(tst.group); len(samplers) > 0 {
- graceNs := lookupMergeGrace(tst.group)
- if graceNs <= 0 {
- graceNs = int64(tst.option.mergeGraceDefault)
- }
- if !isMergeHot(parts, graceNs, time.Now().UnixNano()) {
- chain := newMergeChain(tst.group, "", samplers,
tst.option.decideTimeoutCircuitBreak)
- filter = &mergeFilter{
- chain: chain,
- timeout: tst.option.decideTimeout,
- stageBudget:
resolveStageBudget(tst.option),
- forceSlow: len(chain.projection.Tags)
> 0,
- }
- }
- }
- }
- newPart, dropped, err := tst.mergeParts(tst.fileSystem, closeCh, parts,
newPartID, tst.root, filter)
+ var finalizeGenOverride *uint64
+ if ov != nil {
+ filter = ov.filter
+ finalizeGenOverride = ov.finalizeGen
+ }
+ // Hot / flusher path (no pre-built filter): build the in-merge
retention filter from
+ // the merge-time rules. The finalize path supplies its own filter via
ov, so this
+ // hot build is skipped there.
+ if filter == nil {
+ filter = tst.buildHotMergeFilter(parts)
+ }
+ newPart, dropped, err := tst.mergeParts(tst.fileSystem, closeCh, parts,
newPartID, tst.root, filter, finalizeGenOverride)
if err != nil {
return nil, err
}
@@ -547,7 +607,7 @@ func (tst *tsTable) removeSidxPartOnFailure(sidxName
string, partID uint64) {
}
func (tst *tsTable) mergeParts(fileSystem fs.FileSystem, closeCh <-chan
struct{}, parts []*partWrapper, partID uint64, root string,
- filter *mergeFilter,
+ filter *mergeFilter, finalizeGenOverride *uint64,
) (*partWrapper, map[string]struct{}, error) {
if len(parts) == 0 {
return nil, nil, errNoPartToMerge
@@ -600,6 +660,23 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem,
closeCh <-chan struct{}
tt.mustWriteTagType(fileSystem, dstPath)
pm.MinTimestamp = minTimestamp
pm.MaxTimestamp = maxTimestamp
+ // Finalization-sampling generation stamp, applied BEFORE the on-disk
metadata write
+ // (and the subsequent re-open below) so it survives restart.
finalizeGenOverride set
+ // => this is a finalize round: stamp the output at the round's
generation. nil => any
+ // other merge (hot/flusher): min-propagate from inputs so a merge is
"finalized" only
+ // as much as its least-finalized input — merging two G-stamped parts
stays G (never
+ // un-finalizes), merging a G part with an unstamped late part yields 0
(selectable).
+ if finalizeGenOverride != nil {
+ pm.FinalizeGen = *finalizeGenOverride
+ } else {
+ minGen := parts[0].p.partMetadata.FinalizeGen
+ for _, pw := range parts[1:] {
+ if g := pw.p.partMetadata.FinalizeGen; g < minGen {
+ minGen = g
+ }
+ }
+ pm.FinalizeGen = minGen
+ }
pm.mustWriteMetadata(fileSystem, dstPath)
// No SyncPath here: each mustWrite* helper goes through
fileSystem.WriteAtomic
// which already fsyncs the parent directory after rename. The last
atomic
diff --git a/banyand/trace/merger_durability_test.go
b/banyand/trace/merger_durability_test.go
index 27dcd9dd2..9f13fc044 100644
--- a/banyand/trace/merger_durability_test.go
+++ b/banyand/trace/merger_durability_test.go
@@ -173,7 +173,7 @@ func Test_merger_tornSpansBin_stillPanics(t *testing.T) {
mergedPart, _, mergeErr := tst.mergeParts(fileSystem, closeCh,
[]*partWrapper{
newPartWrapper(nil, p1),
newPartWrapper(nil, p2),
- }, 99, tmpPath, nil)
+ }, 99, tmpPath, nil, nil)
_, _ = mergedPart, mergeErr
t.Fatal("mergeParts returned without panicking — torn-part fail-fast
contract is broken")
diff --git a/banyand/trace/merger_test.go b/banyand/trace/merger_test.go
index af236146f..0d50b2c81 100644
--- a/banyand/trace/merger_test.go
+++ b/banyand/trace/merger_test.go
@@ -966,7 +966,7 @@ func Test_multipleRoundMerges(t *testing.T) {
// Merge all parts for this round
closeCh := make(chan struct{})
- mergedPart, _, err :=
tst.mergeParts(fileSystem, closeCh, partsToMerge, partID, tmpPath, nil)
+ mergedPart, _, err :=
tst.mergeParts(fileSystem, closeCh, partsToMerge, partID, tmpPath, nil, nil)
close(closeCh)
require.NoError(t, err, "Round %d merge
failed", roundIdx+1)
require.NotNil(t, mergedPart, "Round %d
produced nil merged part", roundIdx+1)
@@ -1239,7 +1239,7 @@ func Test_mergeParts(t *testing.T) {
closeCh := make(chan struct{})
defer close(closeCh)
tst := &tsTable{pm: protector.Nop{}}
- p, _, err := tst.mergeParts(fileSystem,
closeCh, pp, partID, root, nil)
+ p, _, err := tst.mergeParts(fileSystem,
closeCh, pp, partID, root, nil, nil)
if tt.wantErr != nil {
if !errors.Is(err, tt.wantErr) {
t.Fatalf("Unexpected error: got
%v, want %v", err, tt.wantErr)
@@ -1392,7 +1392,7 @@ func
Test_mergePartsThenSendIntroduction_cleansUpOnSidxMergeError(t *testing.T)
merges := make(chan *mergerIntroduction, 1)
closeCh := make(chan struct{})
- _, err := tst.mergePartsThenSendIntroduction(snapshotCreatorMerger,
parts, merged, merges, closeCh, mergeTypeFile, mergeLaneFast)
+ _, err := tst.mergePartsThenSendIntroduction(snapshotCreatorMerger,
parts, merged, merges, closeCh, mergeTypeFile, mergeLaneFast, nil)
require.Error(t, err)
require.Contains(t, err.Error(), "sidx merge failed")
diff --git a/banyand/trace/metadata.go b/banyand/trace/metadata.go
index 8a5fa1a20..efe68da17 100644
--- a/banyand/trace/metadata.go
+++ b/banyand/trace/metadata.go
@@ -74,6 +74,7 @@ type schemaRepo struct {
path string
nodeID string
trustedPluginDir string
+ finalizeGraceDefault time.Duration
role databasev1.Role
nativePipelineEnabled bool
}
@@ -88,6 +89,7 @@ func newSchemaRepo(path string, svc *standalone, nodeLabels
map[string]string, n
role: databasev1.Role_ROLE_DATA,
nativePipelineEnabled: svc.option.nativePipelineEnabled,
trustedPluginDir: svc.option.trustedPluginDir,
+ finalizeGraceDefault: svc.option.finalizeGraceDefault,
samplerMeter: newSamplerMetrics(pipelineFactory),
pluginTelemetryFactory: pipelineFactory,
Repository: resourceSchema.NewRepository(
@@ -253,6 +255,9 @@ func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
removeSamplersForGroup(g.Metadata.Name)
teardownGroupTelemetry(g.Metadata.Name)
setMergeGraceForGroup(g.Metadata.Name, 0)
+ setMergeEventForGroup(g.Metadata.Name, false)
+ setFinalizeGraceForGroup(g.Metadata.Name, 0)
+ setFinalizeConfigForGroup(g.Metadata.Name, nil)
sr.samplerMeter.setActiveCount(g.Metadata.Name, 0)
sr.samplerMeter.incRemoveTotal(g.Metadata.Name)
}
@@ -298,6 +303,14 @@ func mergeEventEnabled(cfg *commonv1.TracePipelineConfig)
bool {
return slices.Contains(events,
commonv1.PipelineEvent_PIPELINE_EVENT_MERGE)
}
+// finalizeEventEnabled reports whether cfg applies to the scheduled
finalization
+// pass. Unlike MERGE, an empty enabled_events list does NOT default finalize
on:
+// finalization is an extra background procedure and must be opted into
explicitly
+// (empty list keeps only the backward-compatible MERGE default).
+func finalizeEventEnabled(cfg *commonv1.TracePipelineConfig) bool {
+ return slices.Contains(cfg.GetEnabledEvents(),
commonv1.PipelineEvent_PIPELINE_EVENT_FINALIZE)
+}
+
// samplerLoadFailReason maps a plugin load error to a small, stable set of
reason
// codes for the sampler_load_failed metric label, keeping label cardinality
bounded
// (the full error is logged separately). Unrecognized errors fall back to
"load_error".
@@ -334,14 +347,19 @@ func samplerLoadFailReason(err error) string {
// merge_grace. If any plugin fails to load, the previous good set is kept
intact
// (fail-open) and an ERROR log is emitted.
func (sr *schemaRepo) reconcilePipeline(group string, cfg
*commonv1.TracePipelineConfig) {
- // A nil/disabled config, or one that does not enable the merge event,
clears the
- // group's samplers (retain all). enabled_events is honored here so
merge filtering
- // can be disabled without removing the config.
- if cfg == nil || !cfg.GetEnabled() || !mergeEventEnabled(cfg) {
+ // A nil/disabled config, or one that enables NEITHER the merge nor the
finalize
+ // event, clears the group's samplers (retain all). The sampler set is
shared by
+ // both events (DD11: finalize reuses the group's registered samplers),
so it is
+ // cleared only when both are off. enabled_events is honored here so
filtering can
+ // be disabled without removing the config.
+ if cfg == nil || !cfg.GetEnabled() || (!mergeEventEnabled(cfg) &&
!finalizeEventEnabled(cfg)) {
hadSamplers := len(lookupSamplers(group)) > 0
removeSamplersForGroup(group)
teardownGroupTelemetry(group)
setMergeGraceForGroup(group, 0)
+ setMergeEventForGroup(group, false)
+ setFinalizeGraceForGroup(group, 0)
+ setFinalizeConfigForGroup(group, nil)
sr.samplerMeter.setActiveCount(group, 0)
if hadSamplers {
sr.samplerMeter.incRemoveTotal(group)
@@ -449,6 +467,23 @@ func (sr *schemaRepo) reconcilePipeline(group string, cfg
*commonv1.TracePipelin
graceNs = gd.AsDuration().Nanoseconds()
}
setMergeGraceForGroup(group, graceNs)
+ setMergeEventForGroup(group, mergeEventEnabled(cfg))
+ // Store finalize_grace + threshold config ONLY when the FINALIZE event
is enabled;
+ // a finalize config entry is what marks a group for the background
finalize scanner
+ // (a merge-only group must not be scanned). The proto carries no
threshold-override
+ // fields in v1, so the config registry holds defaults (filled by
lookupFinalizeConfig).
+ // finalize_grace falls back to option.finalizeGraceDefault at lookup
time when unset.
+ if finalizeEventEnabled(cfg) {
+ var finalizeGraceNs int64
+ if fg := cfg.GetFinalizeGrace(); fg != nil {
+ finalizeGraceNs = fg.AsDuration().Nanoseconds()
+ }
+ setFinalizeGraceForGroup(group, finalizeGraceNs)
+ setFinalizeConfigForGroup(group, &finalizeConfig{})
+ } else {
+ setFinalizeGraceForGroup(group, 0)
+ setFinalizeConfigForGroup(group, nil)
+ }
sr.samplerMeter.setActiveCount(group, len(newSet))
if isUpdate {
sr.samplerMeter.incUpdateTotal(group, "success")
diff --git a/banyand/trace/part_metadata.go b/banyand/trace/part_metadata.go
index 62468b444..97ec09bdc 100644
--- a/banyand/trace/part_metadata.go
+++ b/banyand/trace/part_metadata.go
@@ -42,6 +42,11 @@ type partMetadata struct {
MinTimestamp int64 `json:"minTimestamp"`
MaxTimestamp int64 `json:"maxTimestamp"`
ID uint64 `json:"-"`
+ // FinalizeGen is the finalization-sampling generation stamp: 0/absent
means the
+ // part was never touched by a finalize round. On merge it is
min-propagated from
+ // the inputs; a finalize round overrides it to the round's generation.
Unlike ID
+ // it carries a real JSON tag so it survives on disk (see
mergeParts/finalizeGenOverride).
+ FinalizeGen uint64 `json:"finalizeGen,omitempty"`
}
func (pm *partMetadata) reset() {
@@ -52,6 +57,7 @@ func (pm *partMetadata) reset() {
pm.MinTimestamp = 0
pm.MaxTimestamp = 0
pm.ID = 0
+ pm.FinalizeGen = 0
}
func (pm *partMetadata) fillFromSyncContext(ctx *queue.ChunkedSyncPartContext)
{
diff --git a/banyand/trace/pipeline_finalize_test.go
b/banyand/trace/pipeline_finalize_test.go
new file mode 100644
index 000000000..b55a33e7a
--- /dev/null
+++ b/banyand/trace/pipeline_finalize_test.go
@@ -0,0 +1,189 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "google.golang.org/protobuf/types/known/durationpb"
+
+ commonv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+)
+
+// TestBuildHotMergeFilter_MergeEventGate verifies that the hot merge filter
is applied
+// only when the MERGE event is enabled for the group — even though the
sampler set is
+// shared with FINALIZE (DD11). A FINALIZE-only group (samplers registered,
MERGE event
+// off) must NOT filter hot merges.
+func TestBuildHotMergeFilter_MergeEventGate(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "merge-gate-group"
+ replaceSamplersForGroup(group, []namedSampler{{name: "s", sampler:
&dummySampler{}}})
+ tst := &tsTable{group: group, option: option{nativePipelineEnabled:
true, mergeGraceDefault: time.Second}}
+ // An ancient (non-hot) part so hotness never masks the gate.
+ parts := []*partWrapper{{p: &part{partMetadata:
partMetadata{MaxTimestamp: 1, TotalCount: 1}}}}
+
+ // FINALIZE-only: samplers registered, MERGE event disabled → no
hot-merge filter.
+ setMergeEventForGroup(group, false)
+ assert.Nil(t, tst.buildHotMergeFilter(parts), "FINALIZE-only group must
not filter hot merges")
+
+ // MERGE enabled → the filter is built.
+ setMergeEventForGroup(group, true)
+ assert.NotNil(t, tst.buildHotMergeFilter(parts), "MERGE-enabled group
must filter hot merges")
+}
+
+// TestFinalizeEventEnabled_Matrix verifies the finalize event gate. Unlike
MERGE,
+// an empty enabled_events list must NOT default finalize on.
+func TestFinalizeEventEnabled_Matrix(t *testing.T) {
+ tests := []struct {
+ name string
+ events []commonv1.PipelineEvent
+ want bool
+ }{
+ {"empty defaults off", nil, false},
+ {"merge only",
[]commonv1.PipelineEvent{commonv1.PipelineEvent_PIPELINE_EVENT_MERGE}, false},
+ {"finalize only",
[]commonv1.PipelineEvent{commonv1.PipelineEvent_PIPELINE_EVENT_FINALIZE}, true},
+ {"both",
[]commonv1.PipelineEvent{commonv1.PipelineEvent_PIPELINE_EVENT_MERGE,
commonv1.PipelineEvent_PIPELINE_EVENT_FINALIZE}, true},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ cfg := &commonv1.TracePipelineConfig{EnabledEvents:
tt.events}
+ assert.Equal(t, tt.want, finalizeEventEnabled(cfg))
+ })
+ }
+ // mergeEventEnabled semantics unchanged: empty defaults on.
+ assert.True(t, mergeEventEnabled(&commonv1.TracePipelineConfig{}),
"empty events must default MERGE on")
+}
+
+// TestSetFinalizeGrace_RoundTripAndDeleteOnZero mirrors the merge_grace
registry
+// contract: a positive value round-trips, zero deletes.
+func TestSetFinalizeGrace_RoundTripAndDeleteOnZero(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "fg-group"
+ assert.Zero(t, lookupFinalizeGrace(group), "unset group must return 0")
+
+ setFinalizeGraceForGroup(group, int64(7*time.Minute))
+ assert.Equal(t, int64(7*time.Minute), lookupFinalizeGrace(group))
+
+ setFinalizeGraceForGroup(group, 0)
+ assert.Zero(t, lookupFinalizeGrace(group), "zero must delete the entry")
+}
+
+// TestFinalizeConfig_Defaults verifies lookupFinalizeConfig fills unset
fields with
+// package defaults and that the cooldown falls back to the supplied
finalize_grace.
+func TestFinalizeConfig_Defaults(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "fc-group"
+ const graceNs = int64(5 * time.Minute)
+
+ // Unset: all defaults; cooldown == grace.
+ got := lookupFinalizeConfig(group, graceNs)
+ assert.Equal(t, finalizeFloorBytesDefault, got.floorBytes)
+ assert.InDelta(t, finalizeRatioDefault, got.ratio, 1e-9)
+ assert.Equal(t, graceNs, got.cooldownNs, "unset cooldown must fall back
to finalize_grace")
+ assert.Equal(t, finalizeMaxRoundsDefault, got.maxRounds)
+
+ // Explicit overrides are respected.
+ setFinalizeConfigForGroup(group, &finalizeConfig{
+ floorBytes: 1 << 20,
+ ratio: 0.25,
+ cooldownNs: int64(90 * time.Second),
+ maxRounds: 3,
+ })
+ got = lookupFinalizeConfig(group, graceNs)
+ assert.Equal(t, uint64(1<<20), got.floorBytes)
+ assert.InDelta(t, 0.25, got.ratio, 1e-9)
+ assert.Equal(t, int64(90*time.Second), got.cooldownNs)
+ assert.Equal(t, 3, got.maxRounds)
+
+ // nil removes the override -> back to defaults.
+ setFinalizeConfigForGroup(group, nil)
+ got = lookupFinalizeConfig(group, graceNs)
+ assert.Equal(t, finalizeFloorBytesDefault, got.floorBytes)
+}
+
+// TestReconcilePipeline_FinalizeGrace verifies reconcilePipeline stores
finalize_grace
+// and a finalize config entry, and clears them on a nil config.
+func TestReconcilePipeline_FinalizeGrace(t *testing.T) {
+ requirePlugin(t)
+
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "finalize-grace-group"
+
+ sp, spErr := makeSamplerPlugin(pluginTestSoName, 100)
+ require.NoError(t, spErr)
+
+ const wantGrace = 8 * time.Minute
+ cfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {Name: "lss", Kind: &commonv1.Plugin_Sampler{Sampler:
sp}},
+ },
+ EnabledEvents:
[]commonv1.PipelineEvent{commonv1.PipelineEvent_PIPELINE_EVENT_FINALIZE},
+ FinalizeGrace: durationpb.New(wantGrace),
+ }
+
+ sr := makeDataSchemaRepo(pluginTestDir)
+ sr.reconcilePipeline(group, cfg)
+
+ assert.Equal(t, wantGrace.Nanoseconds(), lookupFinalizeGrace(group),
+ "lookupFinalizeGrace must return the configured value in
nanoseconds")
+
+ // After nil config, finalize grace must be cleared.
+ sr.reconcilePipeline(group, nil)
+ assert.Zero(t, lookupFinalizeGrace(group), "finalize grace must be 0
after nil config")
+}
+
+// TestReconcilePipeline_FinalizeOnlyRegistersSamplers verifies the DD11
behavior:
+// a FINALIZE-only config (no MERGE) with a valid plugin now REGISTERS the
group's
+// sampler set (finalize reuses the same samplers as merge).
+func TestReconcilePipeline_FinalizeOnlyRegistersSamplers(t *testing.T) {
+ requirePlugin(t)
+
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "finalize-only-registers"
+
+ sp, spErr := makeSamplerPlugin(pluginTestSoName, 100)
+ require.NoError(t, spErr)
+
+ cfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {Name: "lss", Kind: &commonv1.Plugin_Sampler{Sampler:
sp}},
+ },
+ EnabledEvents:
[]commonv1.PipelineEvent{commonv1.PipelineEvent_PIPELINE_EVENT_FINALIZE},
+ }
+
+ sr := makeDataSchemaRepo(pluginTestDir)
+ sr.reconcilePipeline(group, cfg)
+
+ require.NotEmpty(t, lookupSamplers(group),
+ "a FINALIZE-only config must register the shared sampler set
(DD11)")
+}
diff --git a/banyand/trace/pipeline_registry.go
b/banyand/trace/pipeline_registry.go
index eb2beb015..67dbaea49 100644
--- a/banyand/trace/pipeline_registry.go
+++ b/banyand/trace/pipeline_registry.go
@@ -49,6 +49,145 @@ func lookupMergeGrace(group string) int64 {
return localMergeGraceRegistry.m[group]
}
+// localMergeEventRegistry records, per group, whether the MERGE pipeline
event is
+// enabled. The sampler set is shared by MERGE and FINALIZE (DD11), so the
presence of
+// samplers alone does NOT mean hot merges should be filtered: a FINALIZE-only
group
+// registers samplers for the background pass but must not touch the hot merge
path.
+// buildHotMergeFilter consults this flag so enabled_events reliably gates
MERGE-time
+// filtering independently of FINALIZE.
+var localMergeEventRegistry = struct {
+ m map[string]bool
+ mu sync.RWMutex
+}{m: make(map[string]bool)}
+
+// setMergeEventForGroup records whether the MERGE event is enabled for group.
+// enabled == false removes the entry.
+func setMergeEventForGroup(group string, enabled bool) {
+ localMergeEventRegistry.mu.Lock()
+ if enabled {
+ localMergeEventRegistry.m[group] = true
+ } else {
+ delete(localMergeEventRegistry.m, group)
+ }
+ localMergeEventRegistry.mu.Unlock()
+}
+
+// mergeEventEnabledForGroup reports whether the MERGE-time retention filter
should run
+// for group (false when only FINALIZE is enabled, or nothing is configured).
+func mergeEventEnabledForGroup(group string) bool {
+ localMergeEventRegistry.mu.RLock()
+ defer localMergeEventRegistry.mu.RUnlock()
+ return localMergeEventRegistry.m[group]
+}
+
+// Finalization-sampling threshold defaults (see
.omc/plans/trace-finalize-sampling.md
+// DD2). They bound whether a cooled segment warrants another finalize round.
+const (
+ // finalizeFloorBytesDefault is the absolute minimum newly-arrived
uncompressed
+ // span bytes since the last round that justifies re-finalizing a
segment. Below
+ // this a trickle of late writes is left unsampled (an accepted miss).
+ finalizeFloorBytesDefault uint64 = 8 << 20 // 8 MiB
+ // finalizeRatioDefault gates large segments: re-finalize only when new
unsampled
+ // bytes reach this fraction of the segment's total, so a
proportionally-tiny
+ // addition to a big segment does not trigger a full rewrite.
+ finalizeRatioDefault = 0.10
+ // finalizeMaxRoundsDefault is the hard per-segment lifetime cap on
finalize
+ // rounds; after this the segment is marked terminal and never
re-scanned.
+ finalizeMaxRoundsDefault = 8
+)
+
+// localFinalizeGraceRegistry stores per-group finalize_grace in nanoseconds
as set
+// by reconcilePipeline. A zero value means "use option.finalizeGraceDefault".
It
+// mirrors localMergeGraceRegistry.
+var localFinalizeGraceRegistry = struct {
+ m map[string]int64
+ mu sync.RWMutex
+}{m: make(map[string]int64)}
+
+// setFinalizeGraceForGroup stores graceNs for group. graceNs == 0 removes the
entry.
+func setFinalizeGraceForGroup(group string, graceNs int64) {
+ localFinalizeGraceRegistry.mu.Lock()
+ if graceNs > 0 {
+ localFinalizeGraceRegistry.m[group] = graceNs
+ } else {
+ delete(localFinalizeGraceRegistry.m, group)
+ }
+ localFinalizeGraceRegistry.mu.Unlock()
+}
+
+// lookupFinalizeGrace returns the per-group finalize_grace in nanoseconds, or
0 if
+// none is configured (caller falls back to option.finalizeGraceDefault).
+func lookupFinalizeGrace(group string) int64 {
+ localFinalizeGraceRegistry.mu.RLock()
+ defer localFinalizeGraceRegistry.mu.RUnlock()
+ return localFinalizeGraceRegistry.m[group]
+}
+
+// finalizeConfig holds the per-group finalize threshold knobs. A zero value is
+// never stored directly; lookupFinalizeConfig fills unset fields with
defaults.
+type finalizeConfig struct {
+ floorBytes uint64 // absolute floor; 0 => finalizeFloorBytesDefault
+ ratio float64 // fraction of segment total; <=0 =>
finalizeRatioDefault
+ cooldownNs int64 // min ns between rounds; 0 => the group's
finalize_grace
+ maxRounds int // hard lifetime cap; <=0 => finalizeMaxRoundsDefault
+}
+
+// localFinalizeConfigRegistry stores per-group finalize threshold overrides.
The
+// proto carries no override fields today (v1), so reconcilePipeline stores
defaults;
+// tests may inject custom values via setFinalizeConfigForGroup.
+var localFinalizeConfigRegistry = struct {
+ m map[string]finalizeConfig
+ mu sync.RWMutex
+}{m: make(map[string]finalizeConfig)}
+
+// listFinalizeGroups returns the groups that have finalization sampling
enabled (a
+// finalize config entry exists). The background finalize scanner iterates
these.
+func listFinalizeGroups() []string {
+ localFinalizeConfigRegistry.mu.RLock()
+ defer localFinalizeConfigRegistry.mu.RUnlock()
+ if len(localFinalizeConfigRegistry.m) == 0 {
+ return nil
+ }
+ groups := make([]string, 0, len(localFinalizeConfigRegistry.m))
+ for group := range localFinalizeConfigRegistry.m {
+ groups = append(groups, group)
+ }
+ return groups
+}
+
+// setFinalizeConfigForGroup stores cfg for group. A nil cfg removes the entry.
+func setFinalizeConfigForGroup(group string, cfg *finalizeConfig) {
+ localFinalizeConfigRegistry.mu.Lock()
+ if cfg != nil {
+ localFinalizeConfigRegistry.m[group] = *cfg
+ } else {
+ delete(localFinalizeConfigRegistry.m, group)
+ }
+ localFinalizeConfigRegistry.mu.Unlock()
+}
+
+// lookupFinalizeConfig returns the group's finalize threshold config with any
unset
+// field filled from the package defaults. graceNs is the resolved
finalize_grace for
+// the group and is used as the cooldown when no explicit cooldown is
configured.
+func lookupFinalizeConfig(group string, graceNs int64) finalizeConfig {
+ localFinalizeConfigRegistry.mu.RLock()
+ cfg := localFinalizeConfigRegistry.m[group]
+ localFinalizeConfigRegistry.mu.RUnlock()
+ if cfg.floorBytes == 0 {
+ cfg.floorBytes = finalizeFloorBytesDefault
+ }
+ if cfg.ratio <= 0 {
+ cfg.ratio = finalizeRatioDefault
+ }
+ if cfg.cooldownNs <= 0 {
+ cfg.cooldownNs = graceNs
+ }
+ if cfg.maxRounds <= 0 {
+ cfg.maxRounds = finalizeMaxRoundsDefault
+ }
+ return cfg
+}
+
// namedSampler pairs a sampler instance with its stable identity. The
identity is the
// plugin name plus everything that selects which code runs: the config hash
AND the
// plugin reference (path, symbol, ABI version). reconcilePipeline compares
this identity
@@ -100,6 +239,10 @@ func registerSampler(group string, s sdk.Sampler) func() {
localSamplerRegistry.mu.Lock()
localSamplerRegistry.m[group] = append(localSamplerRegistry.m[group],
namedSampler{sampler: s})
localSamplerRegistry.mu.Unlock()
+ // Direct registration models an active MERGE-filter sampler (the
production reconcile
+ // path sets this from mergeEventEnabled); without it
buildHotMergeFilter would treat
+ // the group as MERGE-disabled and skip filtering.
+ setMergeEventForGroup(group, true)
return func() {
localSamplerRegistry.mu.Lock()
defer localSamplerRegistry.mu.Unlock()
@@ -112,6 +255,7 @@ func registerSampler(group string, s sdk.Sampler) func() {
}
if len(localSamplerRegistry.m[group]) == 0 {
delete(localSamplerRegistry.m, group)
+ setMergeEventForGroup(group, false)
}
}
}
diff --git a/banyand/trace/pipeline_watch_test.go
b/banyand/trace/pipeline_watch_test.go
index b6ff5a7b8..6e5910d6f 100644
--- a/banyand/trace/pipeline_watch_test.go
+++ b/banyand/trace/pipeline_watch_test.go
@@ -47,6 +47,18 @@ func resetRegistries() {
localMergeGraceRegistry.m = make(map[string]int64)
localMergeGraceRegistry.mu.Unlock()
+ localMergeEventRegistry.mu.Lock()
+ localMergeEventRegistry.m = make(map[string]bool)
+ localMergeEventRegistry.mu.Unlock()
+
+ localFinalizeGraceRegistry.mu.Lock()
+ localFinalizeGraceRegistry.m = make(map[string]int64)
+ localFinalizeGraceRegistry.mu.Unlock()
+
+ localFinalizeConfigRegistry.mu.Lock()
+ localFinalizeConfigRegistry.m = make(map[string]finalizeConfig)
+ localFinalizeConfigRegistry.mu.Unlock()
+
pluginTelemetryRegistry.mu.Lock()
pluginTelemetryRegistry.m = make(map[string][]*pluginTelemetry)
pluginTelemetryRegistry.mu.Unlock()
@@ -264,10 +276,11 @@ func TestReconcilePipeline_DisabledConfig(t *testing.T) {
assert.Empty(t, lookupSamplers(group), "disabled config must clear the
registry")
}
-// TestReconcilePipeline_MergeEventNotEnabled verifies that an enabled config
whose
-// enabled_events does not include PIPELINE_EVENT_MERGE installs no merge
samplers
-// (v1 only implements the merge-time filter), clearing any previous set.
-func TestReconcilePipeline_MergeEventNotEnabled(t *testing.T) {
+// TestReconcilePipeline_NoEventOrPluginsClears verifies that an enabled
config with
+// no plugins clears any previous sampler set. (A FINALIZE-only config now
shares the
+// group's sampler set with MERGE per DD11, so the previous set is torn down
and, with
+// no plugins to load, the group ends up empty.)
+func TestReconcilePipeline_NoEventOrPluginsClears(t *testing.T) {
resetRegistries()
defer resetRegistries()
@@ -283,7 +296,7 @@ func TestReconcilePipeline_MergeEventNotEnabled(t
*testing.T) {
EnabledEvents:
[]commonv1.PipelineEvent{commonv1.PipelineEvent_PIPELINE_EVENT_FINALIZE},
})
- assert.Empty(t, lookupSamplers(group), "config without MERGE event must
install no merge samplers")
+ assert.Empty(t, lookupSamplers(group), "a config with no plugins must
install no samplers")
}
// TestReconcilePipeline_OnDeleteKindGroup verifies the OnDelete KindGroup
diff --git a/banyand/trace/svc_standalone.go b/banyand/trace/svc_standalone.go
index 73848710f..b76a7f2b3 100644
--- a/banyand/trace/svc_standalone.go
+++ b/banyand/trace/svc_standalone.go
@@ -68,6 +68,7 @@ type standalone struct {
metadata metadata.Repo
l *logger.Logger
diskMonitor *storage.DiskMonitor
+ finalizeCloser *run.Closer
schemaRepo schemaRepo
snapshotDir string
root string
@@ -103,6 +104,8 @@ func (s *standalone) FlagSet() *run.FlagSet {
fs.BoolVar(&s.option.nativePipelineEnabled,
"trace-pipeline-native-plugin-enabled", false, "enable the native plugin
pipeline for in-merge trace retention")
fs.StringVar(&s.option.trustedPluginDir,
"trace-pipeline-trusted-plugin-dir", "", "trusted directory for native trace
pipeline plugins")
fs.DurationVar(&s.option.mergeGraceDefault,
"trace-pipeline-merge-grace-default", 30*time.Second, "default merge_grace for
in-merge trace retention filter")
+ fs.DurationVar(&s.option.finalizeGraceDefault,
"trace-pipeline-finalize-grace-default", 5*time.Minute,
+ "default finalize_grace (per-segment settling window) for
finalization sampling")
fs.DurationVar(&s.option.decideTimeout,
"trace-pipeline-decide-timeout", 5*time.Second, "hard per-batch Decide timeout
for trace pipeline plugins")
fs.IntVar(&s.option.decideTimeoutCircuitBreak,
"trace-pipeline-decide-timeout-circuit-break", 3,
"consecutive-timeout threshold to disable a (group,schema)
pipeline chain")
@@ -218,6 +221,21 @@ func (s *standalone) PreRun(ctx context.Context) error {
if err != nil {
return err
}
+ // Start the background finalization-sampling scanner when the native
pipeline is
+ // enabled. It is a single node-wide goroutine (concurrency-1) that
sweeps cooled
+ // segments of finalize-enabled groups; it is inert until a group
enables the
+ // FINALIZE event, so starting it unconditionally-under-the-flag is
safe.
+ if s.option.nativePipelineEnabled {
+ s.finalizeCloser = run.NewCloser(1)
+ // Launch via run.Go (not a bare `go`) so a panic in the scan
loop is recovered
+ // and reported instead of crashing the process; the loop's
deferred closer.Done
+ // still fires during panic unwinding. It uses the run.Closer
for cancellation,
+ // hence the contextcheck suppression (mirrors the tsTable
background loops).
+ run.Go(context.Background(), "trace.finalize-scanner", s.l,
func(_ context.Context) { //nolint:contextcheck
+ s.schemaRepo.finalizeScanLoop(s.finalizeCloser,
defaultFinalizeScanInterval) //nolint:contextcheck
+ })
+ }
+
s.l.Info().
Str("root", s.root).
Str("dataPath", s.dataPath).
@@ -232,6 +250,11 @@ func (s *standalone) Serve() run.StopNotify {
}
func (s *standalone) GracefulStop() {
+ // Stop the finalize scanner before closing the schema repo so no round
starts
+ // against a segment whose tables are being torn down.
+ if s.finalizeCloser != nil {
+ s.finalizeCloser.CloseThenWait()
+ }
// Stop disk monitor
if s.diskMonitor != nil {
s.diskMonitor.Stop()
diff --git a/banyand/trace/trace.go b/banyand/trace/trace.go
index 4c770dac2..3f9dc4cdc 100644
--- a/banyand/trace/trace.go
+++ b/banyand/trace/trace.go
@@ -61,6 +61,7 @@ type option struct {
syncInterval time.Duration
memWaitTimeout time.Duration
mergeGraceDefault time.Duration
+ finalizeGraceDefault time.Duration
decideTimeout time.Duration
failedPartsMaxTotalSizeBytes uint64
vectorized vtrace.VectorizedConfig
diff --git a/banyand/trace/tstable.go b/banyand/trace/tstable.go
index acd85d86b..29df045c4 100644
--- a/banyand/trace/tstable.go
+++ b/banyand/trace/tstable.go
@@ -49,17 +49,21 @@ const (
)
type tsTable struct {
- fileSystem fs.FileSystem
- pm protector.Memory
- inFlight map[uint64]struct{}
- handoffCtrl *handoffController
- metrics *metrics
- snapshot *snapshot
- loopCloser *run.Closer
- getNodes func() []string
- l *logger.Logger
- sidxMap map[string]sidx.SIDX
- introductions chan *introduction
+ fileSystem fs.FileSystem
+ pm protector.Memory
+ inFlight map[uint64]struct{}
+ handoffCtrl *handoffController
+ metrics *metrics
+ snapshot *snapshot
+ loopCloser *run.Closer
+ getNodes func() []string
+ l *logger.Logger
+ sidxMap map[string]sidx.SIDX
+ introductions chan *introduction
+ // mergeCh is the introducer loop's merged-introduction channel,
retained so the
+ // finalize worker (an external goroutine) can introduce its
force-merged part
+ // through the same serialized introducer loop the hot merges use.
+ mergeCh chan *mergerIntroduction
p common.Position
group string
root string
@@ -67,7 +71,15 @@ type tsTable struct {
option option
curPartID uint64
pendingDataCount atomic.Int64
- inFlightMu sync.RWMutex
+ // finalizeGenCached mirrors the shard's persisted
finalizeState.FinalizeGeneration.
+ // Seeded at open, bumped by each finalize round. Read O(1) on the hot
introduction
+ // path to decide whether a newly-flushed part is new unsampled data.
+ finalizeGenCached atomic.Uint64
+ // unsampledBytes mirrors finalizeState.UnsampledBytes: uncompressed
span bytes of
+ // parts that arrived after this shard was first finalized. Incremented
on the hot
+ // introduction path (atomic add only), reset by a successful finalize
round.
+ unsampledBytes atomic.Int64
+ inFlightMu sync.RWMutex
sync.RWMutex
shardID common.ShardID
isHot bool
@@ -127,6 +139,7 @@ func (tst *tsTable) startLoop(cur uint64) {
tst.introductions = make(chan *introduction)
flushCh := make(chan *flusherIntroduction)
mergeCh := make(chan *mergerIntroduction)
+ tst.mergeCh = mergeCh
introducerWatcher := make(watcher.Channel, 1)
flusherWatcher := make(watcher.Channel, 1)
// Each loop already calls tst.loopCloser.Done via defer, so a panic
@@ -153,6 +166,7 @@ func (tst *tsTable) startLoopWithConditionalMerge(cur
uint64) {
flushCh := make(chan *flusherIntroduction)
syncCh := make(chan *syncIntroduction)
mergeCh := make(chan *mergerIntroduction)
+ tst.mergeCh = mergeCh
introducerWatcher := make(watcher.Channel, 1)
flusherWatcher := make(watcher.Channel, 1)
// See startLoop for the rationale on routing through run.Go and the
@@ -237,6 +251,12 @@ func initTSTable(fileSystem fs.FileSystem, rootPath
string, p common.Position,
if m != nil {
tst.metrics = m.(*metrics)
}
+ // Seed the cached finalize state from the shard's persisted
finalizeState so the
+ // hot introduction path and the finalize scanner see the on-disk
generation and
+ // unsampled-bytes counter after a restart (a missing file yields zero).
+ fst := readFinalizeState(fileSystem, rootPath)
+ tst.finalizeGenCached.Store(fst.FinalizeGeneration)
+ tst.unsampledBytes.Store(fst.UnsampledBytes)
tst.gc.init(&tst)
ee := fileSystem.ReadDir(rootPath)
if len(ee) == 0 {
diff --git a/docs/design/post-trace-pipeline.md
b/docs/design/post-trace-pipeline.md
index 999d1735b..7b134ef3d 100644
--- a/docs/design/post-trace-pipeline.md
+++ b/docs/design/post-trace-pipeline.md
@@ -657,6 +657,55 @@ flowchart TD
B4 --> C5
```
+## Implementation: finalization sampling (PIPELINE_EVENT_FINALIZE, v1)
+
+Finalization sampling is the background backstop for the in-merge filter. The
in-merge
+sampler (`PIPELINE_EVENT_MERGE`) only drops traces during opportunistic merges
of parts
+older than `merge_grace`; when hot writing rotates to the next segment, the
just-cooled
+segment can be left with parts the in-merge sampler never revisited. The
finalize pass
+sweeps those cooled parts through the **same** registered sampler chain.
+
+**Semantics — best-effort, misses accepted.** Coverage is
best-effort-when-resources-allow,
+**not guaranteed**. The pass yields under memory pressure and never blocks the
hot
+write/merge/query path, so a segment can be born → cool → never-finalized →
TTL-deleted
+un-sampled. That is an accepted outcome (there is no finalize-before-delete
gate).
+
+**Model (per-shard, trace-owned scanner).**
+- A single node-wide scanner goroutine ticks (every
`defaultFinalizeScanInterval`, 10m)
+ and, for each group with the FINALIZE event enabled, selects cooled segments
via
+ `tsdb.SelectSegments(range, reopenClosed=true)` — reopening idle-closed
segments so the
+ common cooled-and-reclaimed case is still covered — then evaluates each
shard.
+- Finalize compute is **serialized (concurrency-1)** and reuses the hot merge
path
+ (`mergePartsThenSendIntroduction` → `mergeParts` + per-sidx `Merge` + the
shard's
+ introducer loop), so core and sidx follow exactly the code a hot merge uses.
It never
+ acquires the hot merge semaphore (`mergeMaxConcurrencyCh`).
+- Each cooled part carries a per-part `finalizeGen` stamp; a round
force-merges the
+ cooled, non-hot parts through the sampler and stamps the output at the
shard's next
+ generation, written to disk **before** the part's metadata is persisted so a
crash
+ cannot cause a double-sample on replay. A round re-samples the cooled set
(like the
+ in-merge filter re-samples on every merge); re-sampling already-kept
survivors through
+ a deterministic sampler is idempotent, and rounds are hard-capped.
+- Per-shard finalize state (`finalize.json`) records the generation, cooldown
clock,
+ round count, terminal flag, and an arrival-based unsampled-bytes counter
(incremented
+ O(1) on the flush path only once the shard has been finalized).
+
+**Knobs** (the proto carries `finalize_grace`; the rest are engine defaults in
v1):
+- `finalize_grace` (default **5m**) — per-segment settling window; a segment
is eligible
+ only when `segEnd < now − finalize_grace`, and parts hotter than this are
skipped.
+- `--trace-pipeline-finalize-grace-default` — node default when a group sets no
+ `finalize_grace`.
+- **FLOOR** (default **8 MiB** uncompressed) — absolute minimum newly-arrived
unsampled
+ bytes to warrant a re-round; smaller trickles are left (accepted miss).
+- **RATIO** (default **0.10**) — for large segments, re-round only when new
unsampled
+ bytes reach this fraction of the segment total.
+- **finalize_cooldown** (default = `finalize_grace`) — at most one round per
window.
+- **max_finalize_rounds** (default **8**) — hard per-shard lifetime cap; after
it the
+ shard is marked terminal and never re-scanned.
+
+**Observability.** Finalize rounds emit the existing merge metrics under
+`type="finalize"`/`lane="finalize"` (`total_merged`, `total_merge_latency`,
+`total_merged_parts`), distinct from the hot-merge labels.
+
## References
- OpenTelemetry Collector — Tail Sampling Processor (`tailsamplingprocessor`):
the keep-all-errors / latency / probabilistic-policy model a typical sampler
plugin implements.
diff --git a/implementation-fn-note.md b/implementation-fn-note.md
new file mode 100644
index 000000000..2d9dd6e69
--- /dev/null
+++ b/implementation-fn-note.md
@@ -0,0 +1,117 @@
+# Finalization Sampling — Implementation Notes (running log)
+
+Plan: `.omc/plans/trace-finalize-sampling.md` (Rev 7). Branch:
`trace-pipeline-finalize-event`.
+This file records decisions NOT in the spec, changes I had to make, tradeoffs,
and anything the user should know. Newest entries at the bottom of each section.
+
+## Ground-truth line numbers (verified against current tree, 2026-07-05)
+- `banyand/internal/storage/version.go`: `SegmentMetadata` struct `:52`;
`segmentMeta` `:74`; `readSegmentMeta` `:79`; `currentVersion = "1.5.0"` `:33`.
NOTE: there is NO create-literal in version.go; the create write lives in
segment.go (to be located).
+- `banyand/trace/part_metadata.go`: `partMetadata` struct `:37`; `reset()`
`:47`; `mustReadMetadata` `:79`; `mustWriteMetadata` `:98` (uses
`fileSystem.WriteAtomic` — already atomic). `fillFromSyncContext` `:57`
(chunked-sync path — must also carry finalizeGen? see decisions).
`ParsePartMetadata` `:262`.
+- `banyand/trace/pipeline_registry.go`: `localMergeGraceRegistry` +
`setMergeGraceForGroup`/`lookupMergeGrace` `:28-50` — mirror for finalize.
`lookupSamplers` `:140`.
+- `banyand/trace/merger.go`: `mergePartsThenSendIntroduction` `:318`; sampler
chain build `:326-342`; `isMergeHot` `:694`; sidx keepFn `:394-404`;
`getAllSidx` loop `:406`; `mergeParts` `:549`; pm min/max ts fill +
`pm.mustWriteMetadata` `:601-603`; re-open `:607`; `mergeBlocks` `:811`;
zero-value `var pm partMetadata` `:980`.
+- `banyand/trace/tstable.go`: `tsTable` struct `:51-74` (has `group`,
`option`, `pm protector.Memory`, `curPartID`, `loopCloser`, `shardID`, `root`,
`fileSystem`; does NOT cache segment EndTime/generation). `startLoop` `:125`
(`run.NewCloser(1 + 3)`, spawns introducer+flusher+merger).
`startLoopWithConditionalMerge` `:150` (`run.NewCloser(1 + 3)`, spawns
introducer-sync+flusher+syncer — **NO mergeLoop**).
+
+## Open implementation questions found while grounding (to resolve during impl)
+- **Q-A (conditional-merge path has no merger):**
`startLoopWithConditionalMerge` spawns introducer-sync/flusher/syncer but NO
`mergeLoop`. The plan (M2) says add the finalizer to BOTH start paths. Need to
confirm whether the sync-mode data node retains cooled segments locally long
enough to finalize, or whether parts are synced away (making finalize a no-op
there). Decision pending — will verify against the sync path before wiring the
finalizer into it. Safe default: still spawn it (it [...]
+- **Q-B (finalizeGen stamp home):** the plan says thread `finalizeGenOverride`
through BOTH `mergeParts` and `mergeBlocks`. But `mergeBlocks` has no access to
the input `partWrapper`s (only the block reader), so it cannot compute
`min(inputs)`. `mergeParts` DOES have `parts []*partWrapper` and already
post-processes `pm` (sets Min/MaxTimestamp) BEFORE `pm.mustWriteMetadata` at
`:603`. Cleaner: thread the override into `mergeParts` only, compute
min-propagation from `parts` there, set `pm [...]
+- **Q-C (fillFromSyncContext / chunked sync):**
`partMetadata.fillFromSyncContext` populates a part's metadata from a sync
context and does NOT set finalizeGen. A synced part would arrive with
finalizeGen=0 (unstamped=selectable), which is correct-by-default. No change
needed unless sync must preserve finalize state across nodes (out of scope —
finalize is per-local-segment). Recording as intentionally unchanged.
+
+## MAJOR ARCHITECTURE DEVIATION (user-approved 2026-07-05) — per-shard state +
trace-owned scanner
+**Why:** The plan put the finalize generation in the per-SEGMENT metadata file
(DD4a/M1) but the lane per-SHARD (`tsTable`, DD7). A segment has N shards
(verified: `segment[T,O].sLst []*shard[T]`, each `shard.table` is a tsTable
rooted at `<segment>/shard-N`, created via `TSTableCreator(..., s.TimeRange,
...)` — `shard.go:81`). So N per-shard lanes would multi-write ONE segment
metadata file, and idle-closed cooled segments (segments get idle-reclaimed)
have NO running tsTable loop, so a [...]
+**Resolution (user picked Option A):**
+- **Per-shard finalize state**, persisted as a small atomic JSON file in the
shard dir (`tst.root`), sole-written by the finalize worker. NOT in
`SegmentMetadata`. => **M1 (both segment metadata structs) is DROPPED**;
`version.go`/`segment.go` create-literal are UNTOUCHED. The segment's cooling
is read from the in-memory `Segment.GetTimeRange().End` (already available), no
persisted finalize field needed at the segment level.
+- **Trace-owned periodic scanner** using the PUBLIC
`tsdb.SelectSegments(coolRange, reopenClosed=true)` (`tsdb.go:277`) +
`segment.Tables()` to enumerate cooled segments' per-shard tsTables.
`reopenClosed=true` REOPENS idle-closed cooled segments (the key fact that
makes this cover the common case), holding an incRef for the finalize duration;
DecRef after. => **no storage/rotation.go hook, no storage→trace seam** (avoids
the layering coupling the user removed for the delete gate). => ** [...]
+- **Concurrency-1 finalize worker**: the scanner processes one shard finalize
task at a time (node-wide serialization), so the finalize compute never fans
out. The tsTable's finalize round reuses the tsTable's existing introducer loop
(a reopened segment has live loops) + `mergeParts(finalizeGenOverride=&G)`. =>
**M2 (4th run.Go loop in BOTH start paths) is DROPPED** — no per-tsTable
finalizer loop; the worker calls `tst.runFinalizeRound(...)` synchronously
while the segment is incRef-he [...]
+- **tsTable gains** `segEndUnixNano` (captured from the `timestamp.TimeRange`
ALREADY passed to `newTSTable` — currently ignored as `_`), plus cached
`finalizeGenCached atomic.Uint64` / `segCooledCached atomic.Bool` /
`unsampledBytes atomic.Int64` seeded from the per-shard finalize file at open.
+**UNCHANGED from plan:** per-part `finalizeGen` stamp + min-propagation via
`finalizeGenOverride` in `mergeParts` before `:603` (DD1.C1/DD6); force
filter-merge reuse + keepFn core+sidx coupling; fail-open; threshold
FLOOR/RATIO/cooldown/max_rounds (Phase 1, per-shard now); protector/disk
gating; sidx rides hot merge path (OQ6). Best-effort/misses-accepted (P2), no
delete gate.
+**PRD updated:** US-002/US-003/US-005 acceptance criteria re-scoped to
per-shard state + trace-owned scanner; M1/M2 references dropped.
+
+## Phase 2 (US-002) — DONE 2026-07-05
+- Per-part `finalizeGen uint64 json:"finalizeGen,omitempty"` added to
`partMetadata` (+ zeroed in `reset()`). Old-format parts load as 0 (verified).
+- New `finalize_state.go`: `finalizeState{LastFinalizedAt, FinalizeGeneration,
UnsampledBytes, FinalizeRounds, Terminal}` +
`readFinalizeState`/`writeFinalizeState` (atomic via `fs.WriteAtomic`
temp+rename). Missing/corrupt file => zero (fail-open, never panics). Stored
per-shard in `tst.root` as `finalize.json`.
+- `SegmentMetadata`/`segmentMeta`/segment create-literal UNTOUCHED (Rev 8). No
version bump.
+- 4 tests pass.
+
+## Phase 3 (US-003) — DONE 2026-07-05
+- `tsTable` gained `finalizeGenCached atomic.Uint64`, `unsampledBytes
atomic.Int64`, `segEndUnixNano int64`. Seeded in `initTSTable` from
`readFinalizeState`; `segEndUnixNano` captured in `newTSTable` from the
(previously-ignored `_`) `timestamp.TimeRange`.
+- `accountUnsampledFlushed(flushed)` helper (takes NO fs → zero metadata I/O
by construction): if `finalizeGenCached>0`, add sum of flushed parts'
`UncompressedSpanSizeBytes` to `unsampledBytes`. Called from `introduceFlushed`
AND `introduceFlushedForSync`. No-op until first finalize (gen 0).
+- 2 tests pass (accounting + seed-from-disk).
+
+## CRITICAL SEMANTICS DECISION — finalize generation is "full-rewrite per
round", NOT minimal-rewrite (plan DD1 inconsistency)
+The plan's DD1 claimed THREE things that are mutually inconsistent: (a)
monotonic per-round generation, (b) selection predicate `finalizeGen <
FinalizeGeneration`, (c) "minimal rewrite volume — only
not-yet-current-generation parts rewritten each round." With a monotonic gen
and a `< G` predicate, previously-finalized outputs (stamped at gen k) are
ALWAYS re-selected once stored advances past k, so true minimal-rewrite is
unachievable with this scheme (worked through several variants; al [...]
+**Chosen (correct + crash-safe):** a round with current stored gen G computes
`Gnext = G+1`; selects all non-hot / not-in-flight parts with `finalizeGen <
Gnext` (which, by the invariant that nothing is stamped above stored, is
effectively ALL cooled parts of the shard); force filter-merges them through
the sampler; stamps outputs `finalizeGen = Gnext` (via `finalizeGenOverride`);
introduces; then persists stored `= Gnext` + resets the unsampled counter. So
**each round re-finalizes the [...]
+- **Why it's safe:** re-sampling already-sampled survivors through a
DETERMINISTIC sampler is idempotent (DD11 soft expectation); a probabilistic
sampler drops a bounded extra amount, never unsafe. Rounds are hard-capped at
`max_finalize_rounds` (8), so ≤8 full rewrites per segment lifetime.
+- **Crash-safety (DD6 preserved):** crash after introduce, before persisting
stored (stored still G, outputs on disk stamped Gnext=G+1): on replay stored=G
→ next round Gnext=G+1 → predicate `finalizeGen < G+1` excludes the
Gnext-stamped outputs (G+1 < G+1 = false); the merged-away inputs are already
gone from the snapshot → no double-sample. Matches DD6 crash-window-2 exactly.
+- **DD1.C1 min-propagation (hot merge inside finalized segment) UNCHANGED and
still needed:** a hot merge with `finalizeGenOverride=nil` sets output `=
min(inputs)`, so merging two gen-Gnext parts stays gen-Gnext (not reset to 0)
and a hot-merge never un-finalizes.
+- **Deviation recorded:** DD1's "minimal rewrite" bullet is relaxed to
"full-rewrite-per-round, bounded + idempotent." If true minimal-rewrite is
later required, it needs a more complex per-part epoch scheme (deferred).
+
+## Phase 4 (US-004) — DONE 2026-07-05
+- `mergeParts` gained `finalizeGenOverride *uint64` (7th param); applied to
`pm.FinalizeGen` (override or min-propagate) BEFORE `pm.mustWriteMetadata`
(:603). **Deviation from plan wording:** the stamp is threaded through
`mergeParts` ONLY (not `mergeBlocks`), because `mergeBlocks` has no access to
the input `partWrapper`s for min-propagation and `mergeParts` already
post-processes `pm` before the on-disk write. Same invariant (on-disk before
re-open).
+- `mergePartsThenSendIntroduction` gained an `ov *mergeOverrides{filter,
finalizeGen}` param so the finalize round reuses the ENTIRE hot tail (sidx
`getAllSidx().Merge` + keepFn coupling + introducer send). Hot callers (merger
lane, flusher) pass `nil`. This is the OQ6 "core+sidx follow hot merge path"
made literal — finalize issues the exact same sidx.Merge calls.
+- `tst.mergeCh` retained in BOTH start paths so the finalize worker (external
goroutine) introduces through the shard's serialized introducer loop. Never
acquires `mergeMaxConcurrencyCh` (that's in `mergeLaneWorker`, which finalize
bypasses).
+- New `finalizer.go` `runFinalizeRound(samplers, graceNs)`: protector gate →
snapshot select (non-hot by finalize_grace, not-in-flight, `finalizeGen <
gNext`) → incRef+in-flight pin → force filter-merge via
`mergePartsThenSendIntroduction(ov={filter,&gNext})` → persist per-shard state
(gen=gNext, rounds++, lastFinalizedAt, counter=0) → refresh cache. Fail-open:
merge error leaves segment intact, scanner retries.
+- Tests: 4 C1 cases (mergeParts propagation, on-disk asserts) + a
running-tsTable round (drop trace2 → 2 remain, gen=1, state persisted). All
pass. 3 existing test callers of mergeParts + 1 of
mergePartsThenSendIntroduction updated for the new param.
+
+## Phase 5 (US-005) — DONE 2026-07-05
+- New `finalize_scanner.go`: `sr.finalizeScanLoop(closer, interval)` (single
node-wide goroutine, concurrency-1) → `runFinalizeScan` iterates
`listFinalizeGroups()` → `scanGroup` does `loadTSDB` +
`SelectSegments(coolRange, reopenClosed=true)` + per-shard `warrantsFinalize` →
`runFinalizeRound`. DecRefs segments.
+- `warrantsFinalize`: reads per-shard `finalize.json` (scanner path, off hot
path); terminal/max-rounds (marks terminal) / cooldown / gen0-always /
`unsampledBytes >= max(FLOOR, RATIO*total)`.
+- reconcile now stores finalize config ONLY when `finalizeEventEnabled` (so
merge-only groups aren't scanned); `listFinalizeGroups()` added;
`schemaRepo.finalizeGraceDefault` added.
+- Wired into `svc_standalone`: `finalizeCloser` started in PreRun (when
`nativePipelineEnabled`), `CloseThenWait` in GracefulStop.
+- **Deviation:** "at most one round per worker wakeup" (PRD literal)
interpreted as concurrency-1 = SERIALIZED (one shard round at a time),
processing all warranting shards per tick sequentially — otherwise 1 shard /
10min is uselessly slow. Recorded.
+- **Coverage gap noted:** full scan→round e2e (SelectSegments→Tables→round on
a real TSDB) is NOT unit-tested (needs a full schemaRepo+TSDB); covered by
`warrantsFinalize` + `runFinalizeRound` unit/integration tests + scan-gating
test. Full e2e deferred to the integration/soak suite (plan E2E section).
**Data-node service (if separate from standalone) needs the same
PreRun/GracefulStop wiring — only `svc_standalone` was wired (Q-A); flag if a
distinct data svc exists.**
+
+## Phase 6 (US-006) — DONE 2026-07-05
+- Finalize compute observable via REUSED merge metrics
`type="finalize"`/`lane="finalize"`
(`total_merged`/`total_merge_latency`/`total_merged_parts`). **Deviation:** did
NOT add dedicated coverage-gauge/rounds/dropped/fail-open finalize counters —
the existing `incPipelineTraces*` counters (evaluated/dropped/…) are DEFINED
BUT UNWIRED (zero call sites, even for the hot path — a pre-existing gap), so
adding a parallel finalize metric set would be inconsistent scope. The reused
lane label [...]
+- Docs: `docs/design/post-trace-pipeline.md` gained an "Implementation:
finalization sampling (v1)" section (best-effort/misses-accepted,
per-shard/scanner model, all knobs). CHANGES.md feature entry added under
0.11.0.
+
+## Verification & review (2026-07-06)
+- **Architect review: APPROVED (0 blockers)**, all 8 correctness claims
verified with file:line evidence (crash-safety ordering, min-propagation,
semaphore bypass, in-flight pin + mergeCh-close safety, O(1) counter, threshold
termination, no-delete-gate, reopen-closed coverage).
+ - It RESOLVED my fn-note Q-A worry: `trace.NewService` (which wires the
finalize scanner) is used by BOTH `pkg/cmdsetup/standalone.go` AND
`pkg/cmdsetup/data.go`, so the data-node role IS covered; the liaison correctly
has no scanner (write-queue path). No separate wiring needed.
+ - It confirmed the select-then-pin TOCTOU in `runFinalizeRound` is
structurally identical to the existing hot dispatcher (`dispatchAllMerges`) and
introduces no new race (the serialized introducer tolerates overlap) — left
as-is intentionally.
+- **Two MINOR findings fixed:**
+ - Counter lost-update: `runFinalizeRound` now snapshots `unsampledBytes` at
round start and `Add(-startBytes)` on commit (persisting the remainder) instead
of `Store(0)`, so late writes arriving DURING a round are preserved for the
next round.
+ - Scanner→round e2e gap: added
`TestFinalizeScan_SelectsCooledSegmentAndFinalizes` — builds a real TSDB
(hourly segments, 30d TTL), writes into a 2h-cooled segment, drives the exact
storage calls `scanGroup` makes (`SelectSegments(reopenClosed=true)` → End
re-check → `Tables()` → `runFinalizeRound`), asserts drop+gen advance, DecRefs.
(Gotcha: the TSDB Option needed `decideTimeout` set — 0 makes the chain time
out → fail-open kept all traces.)
+- **Composition coverage (added 2026-07-06, on user question):**
`TestFinalizeAndMerge_Compose` — one running tsTable with
`nativePipelineEnabled` + a registered sampler, auto-merge disabled
(`maxFanOutSize=0`) for determinism. A real hot merge (MERGE event) drops its
targets via the in-merge filter; a never-merged part is left; a finalize round
(FINALIZE event) drops that part's target and re-keeps the merge survivors (no
double-drop); generation advances 0→1. Proves MERGE+FINALIZE comp [...]
+- **Deslop pass:** no slop found — every finalize symbol is referenced (no
dead code), no duplication (the
`mergePartsThenSendIntroduction`+`mergeOverrides` reuse avoided ~100 lines of
sidx/introduce dup), no needless wrappers, no boundary leaks. Comments document
genuinely complex crash-safety/OQ6 invariants (CLAUDE.md "complex logic"
exception).
+- **Gates:** `make lint` clean; `go build ./...` clean; full
`banyand/trace/...` suite green.
+
+## Commit + UT/IT verification (2026-07-06)
+- Committed as `1c8ad66a` "feat(trace): finalization sampling
(PIPELINE_EVENT_FINALIZE)" on branch `trace-pipeline-finalize-event` (code +
docs + CHANGES; `implementation-fn-note.md` intentionally left untracked,
`.omc/*` gitignored).
+- **UT (green):** `go test ./banyand/trace/... ./banyand/internal/storage/...`
→ trace 27.5s ok, storage 9.3s ok.
+- **IT via `make test-trace-pipeline`** (builds CGO server +
`latencystatussampler.so`, runs standalone + distributed pipeline Ginkgo
suites):
+ - **Standalone in-merge filter suite: 16/16 SUCCESS**, consistently across
re-runs — the trace-package merge changes are sound. (One earlier run had a
single 30s `Eventually` timeout on the first drop spec `t_drop_1` while its
sibling `t_drop_2` (same drop predicate) passed → a merge-completion timing
flake, cleared on re-run.)
+ - **Distributed dynamic-sampler restart spec fails** (`Restart: data node 0
re-applies config from schema store … drop-eligible trace is DROPPED after node
0 restart`): a 30s timeout on "trace schema not yet visible after node 0
restart" (`TraceRegistryService`, a liaison schema service, polled on the
restarted data node). **PROVEN PRE-EXISTING:** checked out `HEAD~1` (f0e0dc21,
before my commit), rebuilt, ran the distributed suite → the SAME spec fails
identically (12 passed / 1 faile [...]
+
+## Sync to latest main + re-check failed distributed test (2026-07-06)
+- Rebased `trace-pipeline-finalize-event` onto latest `origin/main` (apache) —
one new commit `a4fac2f4` "Add Consistently check after Eventually checks
(#1204)" (broad test-hardening: adds `pkg/test/eventually.go` +
`pkg/test/flags/flags.go`; does NOT touch the distributed pipeline restart
test). Rebase clean, no conflicts; my finalize commit is now `f4fe46f1`. `go
build ./...` clean; UT (`banyand/trace` + `banyand/internal/storage`) green.
+- **Re-ran the failed distributed restart spec on latest main: STILL FAILS
identically** — `Restart: data node 0 re-applies config from schema store …
drop-eligible trace is DROPPED after node 0 restart`, "trace schema not yet
visible after node 0 restart" timeout (12 passed / 1 failed). So latest main
does NOT fix it (#1204 didn't touch that test). Combined with the earlier
`HEAD~1` baseline result, this is conclusively a **pre-existing,
environment-specific distributed-restart schema-s [...]
+
+## Fable-model audit + fixes (2026-07-06)
+Second independent audit (code-reviewer, model=fable). Verdict: FIX-FIRST (0
blockers, 2 majors). It caught a real duplication race the earlier architect
pass under-weighted. Fixed:
+- **#1 MAJOR (fixed) — scanner↔dispatcher double-pin → duplicate traces.**
`runFinalizeRound` and the hot `dispatchAllMerges` are TWO concurrent part
selectors. Both did check-under-RLock then pin-under-Lock separately, so both
could select the same cooled part (a cooled segment can still receive late
writes → flusher → dispatcher, exactly while finalize runs). `snapshot.remove`
tolerates an absent id, so both merge outputs survive → the part's rows
duplicated. Fix: (a) `runFinalizeRound [...]
+- **#3 MINOR (fixed) — persist failure could disable the round caps.** The old
code bumped `finalizeGenCached` even when `writeFinalizeState` failed; the next
round's higher gNext would re-include (re-sample) the just-stamped parts. Now:
compute remainder, write state, and ONLY on success mutate the cache + counter;
on failure return the error and leave cache==disk (still G), so replay's
predicate excludes the gNext-stamped parts (no double-sample).
+- **#4 MINOR (fixed) — no disk gate.** Added a free-disk headroom check
(`freeDiskSpace < sum(selected CompressedSizeBytes)` → skip the round as an
accepted miss) so a full-shard rewrite can't push the hot write path into
disk-full.
+- **#5 MINOR (fixed) — dead field.** Removed unused `tsTable.segEndUnixNano`
(+ its assignment; `newTSTable` TimeRange param back to `_`); the scanner uses
`seg.GetTimeRange().End`, not a cached field.
+- **#7 (addressed) — test gaps.** Added
`TestRunFinalizeRound_ReplayExcludesStampedParts` (DD6 crash-window: a
gen-stamped part is excluded on replay, no re-sample) and
`TestWarrantsFinalize_RatioBranch` (the RATIO*total threshold branch, which the
ratio=0 warrants test never hit).
+- **#2 MAJOR (documented, NOT fixed — needs a decision) — scanner reopens
every cooled segment every tick.** `SelectSegments(reopenClosed=true)`
incRef→reopens EVERY cooled segment of every finalize-enabled group each 10-min
tick, including terminal ones, BEFORE the warrants/terminal check. Accurate
severity is DEPLOYMENT-DEPENDENT: with idle-reclaim disabled (default when
idle_timeout<1s) cooled segments stay DORMANT (index!=nil) so the reopen is a
cheap CAS-bump; only with idle-reclaim [...]
+- **#6 MINOR (pre-existing, noted) — `selectSegments` error-path ref leak**
(storage/segment.go:618): on a mid-loop `incRef` error it returns without
DecRef'ing already-incRef'd segments. Pre-existing; the finalize scanner is the
first heavy caller. Not fixed here (storage-layer, low likelihood).
+- **#8/#9 NITs (noted, not fixed):** `tables,_ := seg.Tables()` drops a
(currently-always-nil) error; sync-path accounting can inflate the counter for
parts synced away (cheap no-op rounds).
+Gates after fixes: `go build ./...` clean, `make lint` clean, finalize + merge
unit + integration tests green.
+- **Full-suite `TestTrace` failure investigated → environmental, NOT this
change.** The full `banyand/trace/...` run failed one Ginkgo spec:
`metadata_test.go:395` "changed tag type after merge" (Timed out after 100s
waiting for a background merge to reduce the part count) — the known load-flaky
merge test (memory: flaky under slow-IO/CI, passes in clean env). Ruled my
change in/out decisively: (1) REVERTED the dispatcher conflict-check and the
spec STILL timed out at 100s → my change is [...]
+- **RESOLVED 2026-07-06:** the load was traced to `/tmp/repro3.sh` — an
orphaned 18-day-old script spawning 48 `yes` processes (each ~67% CPU = the
entire load-49), NOT the soak (which was ~0% CPU). Terminated the `yes` burners
+ the vectorized-query soak (banyand + log watchers). Load fell 49.6 → 5.86.
Reran: the merge spec passes in 3.7s (was a 100s timeout), and the FULL
`banyand/trace/...` suite passes in 26.9s (exit 0). All trace tests green. (The
`/data` banyand still holding :1791 [...]
+
+## Audit follow-up round 2 (2026-07-06, user-directed) — #2/#6/#9 fixed, #8 N/A
+User reviewed the unfixed audit items and chose: fix #2 via a storage API, and
fix #6/#8/#9.
+- **#2 MAJOR (fixed) — reopen storm.** Added a storage peek API so the scanner
no longer reopens every cooled segment each tick. New
`storage.SegmentPeek{Start,End,ShardPaths}` + `TSDB.PeekSegments(timeRange)`
(impl `database.PeekSegments` → `segmentController.peekSegments`): snapshots
matching segments' time-range+location under RLock, then lists each segment's
on-disk shard dirs via `walkDir` OUTSIDE the lock — **no `incRef`, no index
reopen, no loop spawn**. The trace scanner (`scanGr [...]
+- **#6 MINOR (fixed) — `selectSegments` error-path ref leak**
(`storage/segment.go`): on a mid-loop `incRef` failure, DecRef the segments
already pinned in earlier iterations before returning.
+- **#8 — N/A (Fable misread).** `Segment.Tables()` returns `([]T, []Cache)`,
NOT an error — the `_` is the unused Cache slice, so there's no dropped error
to log. No change.
+- **#9 MINOR (fixed) — sync-path accounting.** Removed
`accountUnsampledFlushed` from `introduceFlushedForSync`; only
`introduceFlushed` accounts unsampled bytes (parts on the sync path are handed
off, not retained locally, so counting them caused no-op finalize rounds).
+Gates: `go build ./...` clean, `make lint` clean,
`ShardMayWarrant`/`FinalizeScan`/`Warrants` tests pass; full storage + trace
suites re-running.
+
+## Phase 1 (US-001) — DONE 2026-07-05
+Files: `pipeline_registry.go`, `trace.go` (option), `svc_standalone.go`
(flag), `metadata.go` (reconcile), tests `pipeline_finalize_test.go` +
`pipeline_watch_test.go` (resetRegistries + renamed test).
+Decisions / deviations:
+- **Threshold overrides are registry-backed with DEFAULTS only; proto override
fields deferred.** The proto (`TracePipelineConfig`) has `finalize_grace`
(field 4) but NO FLOOR/RATIO/cooldown/max_rounds fields. Rather than invent 4
proto fields (repo-wide pb.go regen + validation), I built
`localFinalizeConfigRegistry` +
`finalizeConfig{floorBytes,ratio,cooldownNs,maxRounds}` with
`lookupFinalizeConfig(group, graceNs)` filling unset fields from package
defaults (8MiB / 0.10 / grace / 8). [...]
+- **finalizeEventEnabled: empty enabled_events does NOT default finalize ON**
(opt-in), unlike `mergeEventEnabled` (empty => MERGE on, for back-compat).
Finalize is an extra procedure, must be explicitly requested.
+- **reconcile now registers the sampler set when MERGE OR FINALIZE is
enabled** (shared set per DD11). The clear branch fires only when NEITHER is
enabled. finalize_grace default (5m) lives in `option.finalizeGraceDefault` via
new flag `--trace-pipeline-finalize-grace-default`, resolved lazily at lane
time (lookupFinalizeGrace==0 => option default), mirroring merge_grace.
+- **Renamed stale test** `TestReconcilePipeline_MergeEventNotEnabled` →
`TestReconcilePipeline_NoEventOrPluginsClears`: its old name/comment ("v1 only
implements merge; finalize-only installs no samplers") became false once
finalize shares the sampler set. The assertion (empty when no plugins) still
holds; the test now documents the correct new semantics. Added
`TestReconcilePipeline_FinalizeOnlyRegistersSamplers` to prove finalize-only
WITH a plugin DOES register.
+- Verified: `go build ./banyand/trace/...` clean; 6 finalize unit tests pass
(reconcile tests ran against the built .so, not skipped).
