diff --git a/index/scorch/forced_purge_fence_test.go b/index/scorch/forced_purge_fence_test.go new file mode 100644 index 000000000..cf7c18645 --- /dev/null +++ b/index/scorch/forced_purge_fence_test.go @@ -0,0 +1,40 @@ +// Copyright (c) 2026 Couchbase, Inc. +// +// Licensed 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 scorch + +import ( + "testing" + "time" +) + +// This file is the compile fence for the forced-purge knob: it references +// ForcedPurgeInterval directly, so it only builds on a scorch that has it, +// and it wires the shared hook the behavioral test uses (that test file +// deliberately compiles without the knob). +func init() { + shortenForcedPurgeInterval = func(d time.Duration) func() { + orig := ForcedPurgeInterval + ForcedPurgeInterval = d + return func() { ForcedPurgeInterval = orig } + } +} + +// TestForcedPurgeIntervalDefault pins that the forced purge ships enabled; +// a zero default silently reverts to idle-only cleanup. +func TestForcedPurgeIntervalDefault(t *testing.T) { + if ForcedPurgeInterval <= 0 { + t.Fatalf("ForcedPurgeInterval default must be > 0, got %v", ForcedPurgeInterval) + } +} diff --git a/index/scorch/forced_purge_test.go b/index/scorch/forced_purge_test.go new file mode 100644 index 000000000..a475ac467 --- /dev/null +++ b/index/scorch/forced_purge_test.go @@ -0,0 +1,136 @@ +// Copyright (c) 2026 Couchbase, Inc. +// +// Licensed 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 scorch + +import ( + "fmt" + "os" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/blevesearch/bleve/v2/document" + index "github.com/blevesearch/bleve_index_api" +) + +// shortenForcedPurgeInterval shrinks the forced-purge interval for the test +// window and returns a restore func. It is assigned by an init() in +// forced_purge_fence_test.go so that this file compiles against a scorch +// without the forced-purge knob; such builds leave it nil and exhibit +// exactly the starvation this test pins. +var shortenForcedPurgeInterval func(d time.Duration) (restore func()) + +// TestForcedPurgeUnderSustainedMutation pins the forced purge: obsolete +// segment files must be reclaimed WHILE the index is under continuous +// mutation, before any quiescence. The batches are unsafe (no wait for +// persist) and issued back-to-back from a writer goroutine, so the persister +// stays behind the root epoch and never reaches its idle-wait purge; the +// .zap census is sampled mid-churn, not after the writer stops. Without the +// forced purge nothing prunes old bolt snapshots either, so every persisted +// epoch pins its segment files forever and the mid-churn census climbs +// without bound (idle-only cleanup was the pre-patch behavior and fails +// this test). +func TestForcedPurgeUnderSustainedMutation(t *testing.T) { + if shortenForcedPurgeInterval != nil { + restore := shortenForcedPurgeInterval(20 * time.Millisecond) + defer restore() + } + + cfg := CreateConfig("TestForcedPurgeUnderSustainedMutation") + cfg["unsafe_batch"] = true // writer must outrun the persister or it idles + if err := InitTest(cfg); err != nil { + t.Fatal(err) + } + defer func() { + if err := DestroyTest(cfg); err != nil { + t.Log(err) + } + }() + analysisQueue := index.NewAnalysisQueue(1) + idx, err := NewScorch(Name, cfg, analysisQueue) + if err != nil { + t.Fatal(err) + } + if err := idx.Open(); err != nil { + t.Fatal(err) + } + defer func() { + if err := idx.Close(); err != nil { + t.Fatal(err) + } + }() + + countZap := func() int { + entries, err := os.ReadDir(cfg["path"].(string)) + if err != nil { + return 0 // dir mid-rename; sample again next tick + } + n := 0 + for _, e := range entries { + if strings.HasSuffix(e.Name(), ".zap") { + n++ + } + } + return n + } + + // Writer: update the same small doc set as fast as batches are accepted, + // so every batch obsoletes prior segments and the root epoch stays ahead + // of the persister for the whole window. + var stop atomic.Bool + var writerErr atomic.Value + done := make(chan struct{}) + go func() { + defer close(done) + for round := 0; !stop.Load(); round++ { + batch := index.NewBatch() + for d := 0; d < 10; d++ { + doc := document.NewDocument(fmt.Sprintf("doc-%d", d)) + doc.AddField(document.NewTextField("body", []uint64{}, + []byte(fmt.Sprintf("round %d body of doc %d", round, d)))) + batch.Update(doc) + } + if err := idx.Batch(batch); err != nil { + writerErr.Store(err) + return + } + } + }() + + // Sample the census mid-churn for ~2s. The assertion is on the maximum + // observed while the writer is still running. + maxZap := 0 + for i := 0; i < 80; i++ { + time.Sleep(25 * time.Millisecond) + if n := countZap(); n > maxZap { + maxZap = n + } + } + stop.Store(true) + <-done + if err, _ := writerErr.Load().(error); err != nil { + t.Fatal(err) + } + + // With the 20ms forced purge, stale bolt snapshots are pruned and their + // files swept continuously, so the census stays near the live segment + // set. Idle-only cleanup (ForcedPurgeInterval = 0) lets every persisted + // epoch pin its files and the mid-churn maximum climbs far past this. + t.Logf("mid-churn max .zap census: %d", maxZap) + if maxZap > 150 { + t.Fatalf("forced purge did not reclaim obsolete segments mid-churn: max %d .zap files on disk", maxZap) + } +} diff --git a/index/scorch/merge_abort_orphan_test.go b/index/scorch/merge_abort_orphan_test.go new file mode 100644 index 000000000..746456272 --- /dev/null +++ b/index/scorch/merge_abort_orphan_test.go @@ -0,0 +1,232 @@ +// Copyright (c) 2026 Couchbase, Inc. +// +// Licensed 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 scorch + +import ( + "fmt" + "os" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/RoaringBitmap/roaring/v2" + "github.com/blevesearch/bleve/v2/document" + index "github.com/blevesearch/bleve_index_api" + segment "github.com/blevesearch/scorch_segment_api/v2" +) + +// abortMergePlugin wraps the default segment plugin. Once armed, the first +// file merge (all inputs persisted) succeeds and every later one fails, +// so a multi-task merge plan aborts after its first task has already +// written its output. +type abortMergePlugin struct { + delegate SegmentPlugin + armed atomic.Bool + calls atomic.Int32 + + mu sync.Mutex + firstPath string +} + +func (p *abortMergePlugin) Type() string { return "abortmerge" } +func (p *abortMergePlugin) Version() uint32 { return 1 } + +func (p *abortMergePlugin) New(results []index.Document) (segment.Segment, uint64, error) { + return p.delegate.New(results) +} + +func (p *abortMergePlugin) NewUsing(results []index.Document, config map[string]interface{}) (segment.Segment, uint64, error) { + return p.delegate.NewUsing(results, config) +} + +func (p *abortMergePlugin) Open(path string) (segment.Segment, error) { + return p.delegate.Open(path) +} + +func (p *abortMergePlugin) OpenUsing(path string, config map[string]interface{}) (segment.Segment, error) { + return p.delegate.OpenUsing(path, config) +} + +// gate returns nil to let a merge proceed. Only armed merges whose inputs +// are all persisted count: the persister's in-memory flush merges must +// pass through untouched. +func (p *abortMergePlugin) gate(segments []segment.Segment, path string) error { + if !p.armed.Load() { + return nil + } + for _, s := range segments { + if _, ok := s.(segment.PersistedSegment); !ok { + return nil + } + } + if p.calls.Add(1) == 1 { + p.mu.Lock() + p.firstPath = path + p.mu.Unlock() + return nil + } + return fmt.Errorf("injected merge failure") +} + +func (p *abortMergePlugin) firstOutputPath() string { + p.mu.Lock() + defer p.mu.Unlock() + return p.firstPath +} + +func (p *abortMergePlugin) Merge(segments []segment.Segment, drops []*roaring.Bitmap, path string, + closeCh chan struct{}, s segment.StatsReporter) ([][]uint64, uint64, error) { + if err := p.gate(segments, path); err != nil { + return nil, 0, err + } + return p.delegate.Merge(segments, drops, path, closeCh, s) +} + +func (p *abortMergePlugin) MergeUsing(segments []segment.Segment, drops []*roaring.Bitmap, path string, + closeCh chan struct{}, s segment.StatsReporter, config map[string]interface{}) ([][]uint64, uint64, error) { + if err := p.gate(segments, path); err != nil { + return nil, 0, err + } + return p.delegate.MergeUsing(segments, drops, path, closeCh, s, config) +} + +var ( + abortMergeSetupOnce sync.Once + abortMergeTestPlug = &abortMergePlugin{} + abortMergeAllow atomic.Bool +) + +// registered lazily so package init has already built the plugin registry +func abortMergeSetup() { + abortMergeTestPlug.delegate = defaultSegmentPlugin + RegisterSegmentPlugin(abortMergeTestPlug, false) + RegistryEventCallbacks["mergeAbortOrphanTest"] = func(e Event) bool { + if e.Kind == EventKindPreMergeCheck { + return abortMergeAllow.Load() + } + return true + } +} + +// TestMergeAbortOrphanSweptWhilePersisterWaits pins the forced purge inside +// the persister's introducer-wait: a merge plan whose later task fails after +// an earlier task succeeded aborts before any introduction, leaving the +// succeeded task's output on disk, unmarked and unreferenced. No +// introduction ever arrives to wake the persister, so only a purge pass run +// while it waits can reclaim the file. Pre-patch the persister had no such +// pass and the orphan survived indefinitely (accumulating once per replan). +func TestMergeAbortOrphanSweptWhilePersisterWaits(t *testing.T) { + abortMergeSetupOnce.Do(abortMergeSetup) + abortMergeTestPlug.armed.Store(false) + abortMergeTestPlug.calls.Store(0) + abortMergeAllow.Store(false) + + origInterval := ForcedPurgeInterval + ForcedPurgeInterval = 100 * time.Millisecond + defer func() { ForcedPurgeInterval = origInterval }() + + cfg := CreateConfig("TestMergeAbortOrphanSweptWhilePersisterWaits") + cfg["forceSegmentType"] = "abortmerge" + cfg["forceSegmentVersion"] = 1 + cfg["eventCallbackName"] = "mergeAbortOrphanTest" + // small tasks so one plan holds several + cfg["scorchMergePlanOptions"] = map[string]interface{}{ + "maxSegmentsPerTier": 2, + "segmentsPerMergeTask": 2, + } + if err := InitTest(cfg); err != nil { + t.Fatal(err) + } + defer func() { + if err := DestroyTest(cfg); err != nil { + t.Log(err) + } + }() + analysisQueue := index.NewAnalysisQueue(1) + idx, err := NewScorch(Name, cfg, analysisQueue) + if err != nil { + t.Fatal(err) + } + if err := idx.Open(); err != nil { + t.Fatal(err) + } + defer func() { + if err := idx.Close(); err != nil { + t.Fatal(err) + } + }() + s := idx.(*Scorch) + + // build up persisted, unmerged segments while file merges are vetoed + for round := 0; round < 8; round++ { + batch := index.NewBatch() + for d := 0; d < 20; d++ { + doc := document.NewDocument(fmt.Sprintf("b%d-d%d", round, d)) + doc.AddField(document.NewTextField("body", []uint64{}, + []byte(fmt.Sprintf("segment %d body of doc %d", round, d)))) + batch.Update(doc) + } + if err := idx.Batch(batch); err != nil { + t.Fatal(err) + } + } + + // wait for the persister to park in its introducer-wait, so the only + // thing that can reclaim anything afterwards is a purge run from there + parked := atomic.LoadUint64(&s.stats.TotPersistLoopWait) + deadline := time.Now().Add(10 * time.Second) + for atomic.LoadUint64(&s.stats.TotPersistLoopWait) == parked { + if time.Now().After(deadline) { + t.Fatal("persister never parked after final batch") + } + time.Sleep(5 * time.Millisecond) + } + + // arm and release the merger: task 1 merges fine, task 2 fails, the + // plan aborts with task 1's output orphaned on disk + abortMergeTestPlug.armed.Store(true) + abortMergeAllow.Store(true) + + deadline = time.Now().Add(10 * time.Second) + for abortMergeTestPlug.calls.Load() < 2 { + if time.Now().After(deadline) { + t.Fatalf("merge plan produced %d file-merge task(s), need >= 2; tune plan options", + abortMergeTestPlug.calls.Load()) + } + time.Sleep(5 * time.Millisecond) + } + orphan := abortMergeTestPlug.firstOutputPath() + if orphan == "" { + t.Fatal("no successful merge output recorded") + } + + // the orphan must disappear within a bounded number of purge intervals, + // with no successful merge introduction to do it as a side effect + sweepDeadline := time.Now().Add(20 * ForcedPurgeInterval) + for time.Now().Before(sweepDeadline) { + if _, err := os.Stat(orphan); os.IsNotExist(err) { + break + } + time.Sleep(10 * time.Millisecond) + } + if _, err := os.Stat(orphan); !os.IsNotExist(err) { + t.Fatalf("aborted merge plan's output %s still on disk after %v", + orphan, 20*ForcedPurgeInterval) + } + if n := atomic.LoadUint64(&s.stats.TotFileMergeIntroductions); n != 0 { + t.Fatalf("expected zero merge introductions, got %d", n) + } +} diff --git a/index/scorch/persister.go b/index/scorch/persister.go index 977097097..78b6c0dc7 100644 --- a/index/scorch/persister.go +++ b/index/scorch/persister.go @@ -110,6 +110,19 @@ func (s *Scorch) persisterLoop() { var persistWatchers []*epochWatcher var lastPersistedEpoch, lastMergedEpoch uint64 var ew *epochWatcher + var lastForcedPurge time.Time + + // One ticker serves every parked select below. The loop-head pass only + // runs while the loop iterates; a persister parked waiting on the + // introducer never reaches it — and a merge-abort storm (a plan whose + // later task fails after an earlier task succeeded) orphans unmarked + // outputs at replan rate with nothing sweeping them. + var forcedPurgeCh <-chan time.Time + if ForcedPurgeInterval > 0 { + forcedPurgeTicker := time.NewTicker(ForcedPurgeInterval) + defer forcedPurgeTicker.Stop() + forcedPurgeCh = forcedPurgeTicker.C + } var unpersistedCallbacks []index.BatchCallback @@ -127,6 +140,18 @@ OUTER: for { atomic.AddUint64(&s.stats.TotPersistLoopBeg, 1) + // Cleanup is otherwise only attempted while the persister waits for + // changes (below) or inside the merger-catch-up pause - both are + // unreachable on an index under sustained mutation, so obsolete files + // accumulate at churn rate until the process restarts. Force a + // periodic pass regardless of load. + if ForcedPurgeInterval > 0 && time.Since(lastForcedPurge) >= ForcedPurgeInterval { + if ok := s.fireEvent(EventKindPurgerCheck, 0); ok { + s.removeOldData() + } + lastForcedPurge = time.Now() + } + select { case <-s.closeCh: break OUTER @@ -255,6 +280,12 @@ OUTER: // if the watchers are already caught up then let them wait, // else let them continue to do the catch up persistWatchers = append(persistWatchers, ew) + case <-forcedPurgeCh: + // nothing introduced, nothing persisted — but aborted merge + // plans may have orphaned already-unmarked outputs + if ok := s.fireEvent(EventKindPurgerCheck, 0); ok { + s.removeOldData() + } } atomic.AddUint64(&s.stats.TotPersistLoopEnd, 1) @@ -318,6 +349,20 @@ func (s *Scorch) pausePersisterForMergerCatchUp(lastPersistedEpoch uint64, // Persister pause until the merger catches up to reduce the segment // file count under the threshold. // But if there is memory pressure, then skip this sleep maneuvers. + // The nap below blocks on merger progress. If the merger is itself + // starved (e.g. a large FST merge on a constrained CPU), that signal may + // not come for hours - and with the persister napping, nothing ever + // deletes obsolete files, so the file count that keeps this loop alive + // only grows: a livelock. Run the purger on a ticker inside the nap; a + // successful sweep drops numFilesOnDisk below the threshold and exits + // the loop without requiring merger progress. + var napPurgeCh <-chan time.Time + if ForcedPurgeInterval > 0 { + napPurgeTicker := time.NewTicker(ForcedPurgeInterval) + defer napPurgeTicker.Stop() + napPurgeCh = napPurgeTicker.C + } + OUTER: for po.PersisterNapUnderNumFiles > 0 && numFilesOnDisk >= uint64(po.PersisterNapUnderNumFiles) && @@ -330,6 +375,10 @@ OUTER: case ew := <-s.persisterNotifier: persistWatchers = append(persistWatchers, ew) lastMergedEpoch = ew.epoch + case <-napPurgeCh: + if ok := s.fireEvent(EventKindPurgerCheck, 0); ok { + s.removeOldData() + } } atomic.AddUint64(&s.stats.TotPersisterSlowMergerResume, 1) @@ -1094,6 +1143,12 @@ func (s *Scorch) removeOldData() { // rollback'ability. var NumSnapshotsToKeep = 1 +// ForcedPurgeInterval forces a removeOldData pass from the persister loop at +// least this often even when the index never goes idle. Without it, cleanup +// runs only when the persister catches up and waits, which never happens +// under sustained mutation. 0 restores the idle-only behavior. +var ForcedPurgeInterval = time.Minute + // RollbackSamplingInterval controls how far back we are looking // in the history to get the rollback points. // For example, a value of 10 minutes ensures that the