From c0a17624a776a1ea0904d85a9e3d62d6bd74da90 Mon Sep 17 00:00:00 2001 From: Andrew Batz Date: Tue, 21 Jul 2026 11:59:32 -0400 Subject: [PATCH 1/6] fix(scorch): force periodic removeOldData even when the persister never idles removeOldData() - the only path that deletes obsolete segment files and prunes old bolt snapshots - runs from just two call sites: the persister's idle-wait block and the merger-catch-up pause. An index under sustained mutation (e.g. a continuous enrichment backfill feeding batches) never reaches either: the persist loop takes 'continue OUTER' on every pass, so obsolete .zap files accumulate at churn rate, unbounded, and survive process restarts (observed in production: root.bolt referencing 3 snapshots while the store directory held 15,703 segment files / 28GB, growing 17GB/h). Add ForcedPurgeInterval (default 1m, same package-var style as NumSnapshotsToKeep; 0 disables): the persister loop head now fires a purger check and removeOldData pass at least that often, regardless of load. Cleanup becomes a scheduled duty instead of an idle-time courtesy. Same goroutine as the existing call sites - no new concurrency. Suite: index/scorch green except pre-existing TestIndexReader vector assertion failure, identical on the unpatched v2.5.8-antfly002 tag. --- index/scorch/persister.go | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/index/scorch/persister.go b/index/scorch/persister.go index 977097097..792f52563 100644 --- a/index/scorch/persister.go +++ b/index/scorch/persister.go @@ -110,6 +110,7 @@ func (s *Scorch) persisterLoop() { var persistWatchers []*epochWatcher var lastPersistedEpoch, lastMergedEpoch uint64 var ew *epochWatcher + var lastForcedPurge time.Time var unpersistedCallbacks []index.BatchCallback @@ -127,6 +128,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 @@ -1094,6 +1107,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 From ee80552234e00440992c6410e9a36b71e5776ad0 Mon Sep 17 00:00:00 2001 From: Andrew Batz Date: Tue, 21 Jul 2026 12:41:43 -0400 Subject: [PATCH 2/6] fix(scorch): run the purger inside the merger-catch-up nap The nap loop blocks on persisterNotifier waiting for merger progress. If the merger is starved (observed: merge workers grinding a vellum FST merge of a 17k-tiny-segment pile for hours under a 1-vCPU container limit), that signal never comes - and since the napping persister is the only goroutine that runs removeOldData, nothing deletes obsolete files while the file count that keeps the nap alive only grows: a livelock. Observed in production twice in one day, including once at 4 vCPUs. Add a ForcedPurgeInterval ticker case to the nap select: cleanup no longer requires merger permission, and a successful sweep drops numFilesOnDisk below the nap threshold, exiting the loop on its own. Complements the loop-head forced pass (previous commit), which covers the sub-threshold idle-starvation regime. Suite: green except pre-existing TestIndexReader (identical on unpatched v2.5.8-antfly002). --- index/scorch/persister.go | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/index/scorch/persister.go b/index/scorch/persister.go index 792f52563..e5e7e11ea 100644 --- a/index/scorch/persister.go +++ b/index/scorch/persister.go @@ -331,6 +331,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) && @@ -343,6 +357,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) From de520d72b41ae9344256af311a5f9e7bff5ef9e1 Mon Sep 17 00:00:00 2001 From: Andrew Batz Date: Tue, 21 Jul 2026 14:18:30 -0400 Subject: [PATCH 3/6] test(scorch): pin ForcedPurgeInterval - mid-churn segment reclaim under sustained mutation Two guarantees: (1) compile fence - the test references ForcedPurgeInterval, so any merge that drops the forced-purge patch breaks the build rather than silently reverting GC to idle-only; (2) behavioral pin - unsafe batches from a writer goroutine keep the persister behind the root epoch (never idle) and the .zap census is sampled mid-churn: with the 20ms purge the observed max stays at the live set (17); with the interval disabled (pre-patch idle-only semantics) nothing prunes old bolt snapshots either, every persisted epoch pins its files, and the census reaches 868 within 2s (red-run verified). An earlier version asserted after a terminal sleep, which let the idle-wait purge run during quiescence - it passed even with the patch disabled. The assertion must happen while the writer is still running; slowdown only shrinks the census, so the green arm is timing-safe in CI. --- index/scorch/forced_purge_test.go | 128 ++++++++++++++++++++++++++++++ 1 file changed, 128 insertions(+) create mode 100644 index/scorch/forced_purge_test.go diff --git a/index/scorch/forced_purge_test.go b/index/scorch/forced_purge_test.go new file mode 100644 index 000000000..777a32ea1 --- /dev/null +++ b/index/scorch/forced_purge_test.go @@ -0,0 +1,128 @@ +// 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" +) + +// TestForcedPurgeUnderSustainedMutation pins ForcedPurgeInterval: 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; disabling the +// interval reproduces it and fails this test). +func TestForcedPurgeUnderSustainedMutation(t *testing.T) { + origInterval := ForcedPurgeInterval + ForcedPurgeInterval = 20 * time.Millisecond + defer func() { ForcedPurgeInterval = origInterval }() + + 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) + } +} From 8afd5022673268b0606fcc0b4b84637980415bc6 Mon Sep 17 00:00:00 2001 From: Andrew Batz Date: Wed, 22 Jul 2026 02:42:49 -0400 Subject: [PATCH 4/6] fix(scorch): run forced purge while persister waits on the introducer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The persister parks in three places. The loop head purges every iteration and the merger-catch-up nap got a purge ticker previously — but the introducer-wait select had neither, on the assumption that introductions always arrive. A merge plan whose later task fails after an earlier task succeeded aborts before any introduction, so under a persistent merge failure nothing ever wakes the persister: the aborted plans' unmarked outputs (one per replan, observed at ~9/s) accumulate with no sweeper. Give that select the same forced-purge ticker arm, bounding orphaned outputs to one ForcedPurgeInterval's worth. --- index/scorch/persister.go | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/index/scorch/persister.go b/index/scorch/persister.go index e5e7e11ea..78b6c0dc7 100644 --- a/index/scorch/persister.go +++ b/index/scorch/persister.go @@ -112,6 +112,18 @@ func (s *Scorch) persisterLoop() { 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 po, err := s.parsePersisterOptions() @@ -268,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) From ca083a86a73c3e4f4fc22d2da8a35bf48ed79810 Mon Sep 17 00:00:00 2001 From: Andrew Batz Date: Wed, 22 Jul 2026 14:36:03 -0400 Subject: [PATCH 5/6] test(scorch): make forced-purge test behavioral, split compile fence The behavioral test no longer references ForcedPurgeInterval, so it compiles against a scorch without the knob and fails there by accumulation instead of by compile error (853 mid-churn .zap files vs the 150 bound on the pre-patch persister). The interval shortening moves behind a hook wired by forced_purge_fence_test.go, which keeps the direct symbol reference and pins the non-zero default. --- index/scorch/forced_purge_fence_test.go | 40 +++++++++++++++++++++++++ index/scorch/forced_purge_test.go | 20 +++++++++---- 2 files changed, 54 insertions(+), 6 deletions(-) create mode 100644 index/scorch/forced_purge_fence_test.go 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 index 777a32ea1..a475ac467 100644 --- a/index/scorch/forced_purge_test.go +++ b/index/scorch/forced_purge_test.go @@ -26,7 +26,14 @@ import ( index "github.com/blevesearch/bleve_index_api" ) -// TestForcedPurgeUnderSustainedMutation pins ForcedPurgeInterval: obsolete +// 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 @@ -34,12 +41,13 @@ import ( // .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; disabling the -// interval reproduces it and fails this test). +// without bound (idle-only cleanup was the pre-patch behavior and fails +// this test). func TestForcedPurgeUnderSustainedMutation(t *testing.T) { - origInterval := ForcedPurgeInterval - ForcedPurgeInterval = 20 * time.Millisecond - defer func() { ForcedPurgeInterval = origInterval }() + 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 From 3a3b1b4f4d8b439a6d77639aa1aada41a01d12bc Mon Sep 17 00:00:00 2001 From: Andrew Batz Date: Wed, 22 Jul 2026 14:36:03 -0400 Subject: [PATCH 6/6] test(scorch): pin introducer-wait purge of aborted merge plan outputs A wrapper segment plugin lets the first file merge of a plan succeed and fails the rest, so a multi-task plan aborts before introduction with the first task's output orphaned on disk. With the persister parked in its introducer-wait, the orphan must disappear within a bounded number of forced-purge intervals and without any merge introduction. Fails on the pre-patch persister (orphan still on disk after 20 intervals). --- index/scorch/merge_abort_orphan_test.go | 232 ++++++++++++++++++++++++ 1 file changed, 232 insertions(+) create mode 100644 index/scorch/merge_abort_orphan_test.go 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) + } +}