From 43e30261f18cb0c0591d61985e0d6c2ab1a649a6 Mon Sep 17 00:00:00 2001 From: Ritesh Kumar Date: Thu, 17 Sep 2026 18:38:53 +0530 Subject: [PATCH 1/6] rosmar changes prior to CBG-5041 - change rosmar's DCP behvaior to remove checkpoints on one shot DCP feed - add test coverage for the same - modify behaviour of other tests --- feeds.go | 21 +++++++ feeds_test.go | 162 ++++++++++++++++++++++++++++++++++++++++++++++++-- queue.go | 7 +++ 3 files changed, 185 insertions(+), 5 deletions(-) diff --git a/feeds.go b/feeds.go index b4e6db3..ccf10aa 100644 --- a/feeds.go +++ b/feeds.go @@ -324,6 +324,15 @@ func (feed *dcpFeed) run() { } debug("%s stopping", feed) + // A dump feed that reached its end-of-feed marker streamed everything, so its checkpoint is spent. + // A terminated feed keeps its checkpoint so the next run can resume. + if feed.args.Dump && !feed.events.closed() { + if err := feed.deleteCheckpoint(); err != nil { + logError("Error deleting %s checkpoint: %v", feed, err) + } + return + } + if feed.lastCasChanged { if err := feed.writeCheckpoint(); err != nil { logError("Error saving %s checkpoint: %v", feed, err) @@ -331,6 +340,18 @@ func (feed *dcpFeed) run() { } } +// Deletes the feed's checkpoint document, if there is one. +func (feed *dcpFeed) deleteCheckpoint() error { + if feed.args.CheckpointPrefix == "" { + return nil + } + err := feed.metadataStore.Delete(feed.ctx, feed.checkpointKey()) + if _, ok := err.(sgbucket.MissingError); ok { + return nil + } + return err +} + func (feed *dcpFeed) close() { feed.events.close() } diff --git a/feeds_test.go b/feeds_test.go index 2dd0453..6f62d26 100644 --- a/feeds_test.go +++ b/feeds_test.go @@ -22,6 +22,8 @@ import ( "github.com/stretchr/testify/require" ) +// TestBackfill verifies that a dump feed replays the documents already in the collection, bracketed +// by BeginBackfill and EndBackfill, and then closes DoneChan. func TestBackfill(t *testing.T) { ensureNoLeakedFeeds(t) bucket := makeTestBucket(t) @@ -49,6 +51,8 @@ func TestBackfill(t *testing.T) { assert.False(t, ok) } +// TestMutations verifies that a feed with no backfill delivers only mutations made after it starts, +// including deletions, and closes DoneChan when the bucket is closed. func TestMutations(t *testing.T) { ctx := t.Context() ensureNoLeakedFeeds(t) @@ -83,6 +87,8 @@ func TestMutations(t *testing.T) { assert.False(t, ok) } +// TestCheckpoint verifies that a feed writes a checkpoint when it is terminated, and that a later +// feed with the same CheckpointPrefix resumes from it rather than replaying earlier documents. func TestCheckpoint(t *testing.T) { ensureNoLeakedFeeds(t) bucket := makeTestBucket(t) @@ -99,9 +105,11 @@ func TestCheckpoint(t *testing.T) { }, ID: "myID", Backfill: sgbucket.FeedResume, - Dump: true, + Dump: false, + Terminator: make(chan bool), CheckpointPrefix: "Checkpoint", } + events, doneChan := startFeedWithArgs(t, bucket, args) event := <-events @@ -110,6 +118,7 @@ func TestCheckpoint(t *testing.T) { event = <-events assert.Equal(t, sgbucket.FeedOpEndBackfill, event.Opcode) + close(args.Terminator) _, ok := <-doneChan assert.False(t, ok) @@ -126,7 +135,8 @@ func TestCheckpoint(t *testing.T) { }, ID: "myID", Backfill: sgbucket.FeedResume, - Dump: true, + Dump: false, + Terminator: make(chan bool), CheckpointPrefix: "Checkpoint", } events, doneChan = startFeedWithArgs(t, bucket, args) @@ -143,10 +153,130 @@ func TestCheckpoint(t *testing.T) { event = <-events assert.Equal(t, sgbucket.FeedOpEndBackfill, event.Opcode) + close(args.Terminator) + _, ok = <-doneChan + assert.False(t, ok) +} + +// TestDumpDeletesCheckpoint verifies that a dump feed deletes its checkpoint once it has streamed +// everything, so a second dump feed with the same CheckpointPrefix replays the collection from the +// start rather than resuming. +func TestDumpDeletesCheckpoint(t *testing.T) { + ensureNoLeakedFeeds(t) + bucket := makeTestBucket(t) + ctx := t.Context() + c := bucket.DefaultDataStore(ctx) + + addToCollection(t, c, "able", 0, "A") + addToCollection(t, c, "baker", 0, "B") + addToCollection(t, c, "charlie", 0, "C") + + const prefix = "DumpCheckpoint" + args := sgbucket.FeedArguments{ + Scopes: map[string][]string{ + "_default": {"_default"}, + }, + ID: "myID", + Backfill: sgbucket.FeedResume, + Dump: true, + CheckpointPrefix: prefix, + } + + // A dump feed ends on its own once it has streamed everything. + events, doneChan := startFeedWithArgs(t, bucket, args) + + event := <-events + assert.Equal(t, sgbucket.FeedOpBeginBackfill, event.Opcode) + readExpectedEventsABC(t, events) + event = <-events + assert.Equal(t, sgbucket.FeedOpEndBackfill, event.Opcode) + + _, ok := <-doneChan + assert.False(t, ok) + + // The checkpoint is spent, so the feed removed it. + var checkpt checkpoint + _, err := c.Get(ctx, prefix, &checkpt) + require.Error(t, err) + require.IsType(t, sgbucket.MissingError{}, err) + + // With no checkpoint to resume from, a second dump feed replays the same documents. + t.Logf("---- Second dump feed, no checkpoint to resume from ---") + events, doneChan = startFeedWithArgs(t, bucket, args) + + event = <-events + assert.Equal(t, sgbucket.FeedOpBeginBackfill, event.Opcode) + readExpectedEventsABC(t, events) + event = <-events + assert.Equal(t, sgbucket.FeedOpEndBackfill, event.Opcode) + _, ok = <-doneChan assert.False(t, ok) } +// TestDumpKeepsCheckpointWhenTerminated verifies that a dump feed stopped before it streams +// everything keeps its checkpoint, so the next feed can resume from where it stopped. +func TestDumpKeepsCheckpointWhenTerminated(t *testing.T) { + ensureNoLeakedFeeds(t) + bucket := makeTestBucket(t) + ctx := t.Context() + c := bucket.DefaultDataStore(ctx) + + // enough documents that the feed is still draining when the terminator closes + const numDocs = 1000 + for i := range numDocs { + addToCollection(t, c, fmt.Sprintf("doc%d", i), 0, "V") + } + + const prefix = "TerminatedDumpCheckpoint" + terminator := make(chan bool) + doneChan := make(chan struct{}) + firstMutation := make(chan struct{}) + release := make(chan struct{}) + + var processed atomic.Int64 + var once sync.Once + callback := func(event sgbucket.FeedEvent) bool { + if event.Opcode != sgbucket.FeedOpMutation { + return true + } + processed.Add(1) + // hold the feed inside its own goroutine while the test terminates it + once.Do(func() { + close(firstMutation) + <-release + }) + return true + } + + args := sgbucket.FeedArguments{ + ID: "myID", + Backfill: sgbucket.FeedResume, + Dump: true, + Terminator: terminator, + DoneChan: doneChan, + CheckpointPrefix: prefix, + MetadataStore: c, + } + require.NoError(t, bucket.StartDCPFeed(ctx, args, callback, nil)) + + <-firstMutation + close(terminator) + close(release) + <-doneChan + + // the feed was cut short, so it never reached its end-of-feed marker + assert.Less(t, processed.Load(), int64(numDocs)) + + // a terminated feed keeps its checkpoint, unlike one that streamed everything + var checkpt checkpoint + _, err := c.Get(ctx, prefix, &checkpt) + require.NoError(t, err) + assert.NotEmpty(t, checkpt.LastCas) +} + +// TestResumeFromCheckpoint verifies that a resumed feed delivers only the documents written while it +// was stopped, and that it sees the checkpoint document itself as a mutation. func TestResumeFromCheckpoint(t *testing.T) { ensureNoLeakedFeeds(t) bucket := makeTestBucket(t) @@ -162,7 +292,8 @@ func TestResumeFromCheckpoint(t *testing.T) { args := sgbucket.FeedArguments{ ID: "myID", Backfill: sgbucket.FeedResume, - Dump: true, + Dump: false, + Terminator: make(chan bool), CheckpointPrefix: prefix, } events, doneChan := startFeedWithArgs(t, bucket, args) @@ -178,6 +309,7 @@ func TestResumeFromCheckpoint(t *testing.T) { event = <-events assert.Equal(t, sgbucket.FeedOpEndBackfill, event.Opcode) + close(args.Terminator) <-doneChan // Add new docs while feed is off @@ -185,6 +317,7 @@ func TestResumeFromCheckpoint(t *testing.T) { addToCollection(t, c, "doc4", 0, "V4") // Resume feed + args.Terminator = make(chan bool) events, doneChan = startFeedWithArgs(t, bucket, args) event = <-events @@ -205,9 +338,12 @@ func TestResumeFromCheckpoint(t *testing.T) { assert.Contains(t, seenDocs, "doc4") assert.Contains(t, seenDocs, prefix) // The checkpoint doc update + close(args.Terminator) <-doneChan } +// TestSharedCheckpoint verifies that feeds on different collections share one checkpoint document, +// storing a last CAS per collection, and that each resumes from its own entry. func TestSharedCheckpoint(t *testing.T) { ensureNoLeakedFeeds(t) bucket := makeTestBucket(t) @@ -225,7 +361,8 @@ func TestSharedCheckpoint(t *testing.T) { args1 := sgbucket.FeedArguments{ ID: "id1", Backfill: sgbucket.FeedResume, - Dump: true, + Dump: false, + Terminator: make(chan bool), CheckpointPrefix: prefix, } events1, done1 := startFeedWithArgs(t, bucket, args1) @@ -234,13 +371,15 @@ func TestSharedCheckpoint(t *testing.T) { break } } + close(args1.Terminator) <-done1 // Run feed for c2 args2 := sgbucket.FeedArguments{ ID: "id2", Backfill: sgbucket.FeedResume, - Dump: true, + Dump: false, + Terminator: make(chan bool), CheckpointPrefix: prefix, Scopes: map[string][]string{"S": {"C"}}, } @@ -250,6 +389,7 @@ func TestSharedCheckpoint(t *testing.T) { break } } + close(args2.Terminator) <-done2 // Verify checkpoint document in DefaultDataStore (c1) @@ -264,6 +404,7 @@ func TestSharedCheckpoint(t *testing.T) { addToCollection(t, c1, "c1-doc2", 0, "V1-2") addToCollection(t, c2, "c2-doc2", 0, "V2-2") + args1.Terminator = make(chan bool) events1, done1 = startFeedWithArgs(t, bucket, args1) foundC1Doc2 := false for e := range events1 { @@ -275,8 +416,10 @@ func TestSharedCheckpoint(t *testing.T) { } } assert.True(t, foundC1Doc2) + close(args1.Terminator) <-done1 + args2.Terminator = make(chan bool) events2, done2 = startFeedWithArgs(t, bucket, args2) foundC2Doc2 := false for e := range events2 { @@ -288,6 +431,7 @@ func TestSharedCheckpoint(t *testing.T) { } } assert.True(t, foundC2Doc2) + close(args2.Terminator) <-done2 } @@ -345,6 +489,8 @@ func readExpectedEventsDEF(t *testing.T, events chan sgbucket.FeedEvent) { assertEventEquals(t, sgbucket.FeedEvent{Opcode: sgbucket.FeedOpMutation, Key: []byte("fahrvergnügen"), Value: []byte(`"F"`), DataType: sgbucket.FeedDataTypeJSON, RevNo: 1}, e) } +// TestCrossBucketEvents verifies that a feed on a second bucket handle opened over the same file +// receives the mutations written through the first. func TestCrossBucketEvents(t *testing.T) { ctx := t.Context() ensureNoLeakedFeeds(t) @@ -387,6 +533,8 @@ func TestCrossBucketEvents(t *testing.T) { assert.False(t, ok) } +// TestCollectionMutations verifies that a feed scoped to two named collections reports each event +// against the right collection ID, and delivers every document from both. func TestCollectionMutations(t *testing.T) { ctx := t.Context() ensureNoLeakedFeeds(t) @@ -468,6 +616,8 @@ func TestCollectionMutations(t *testing.T) { assert.Equal(t, len(c2Keys), numDocs) } +// TestSetRawAutodetectJSON verifies that the raw write methods detect JSON content and set the feed +// event's DataType accordingly. func TestSetRawAutodetectJSON(t *testing.T) { ctx := t.Context() ensureNoLeakedFeeds(t) @@ -700,6 +850,8 @@ func TestFeedStopsOnCorruptEvent(t *testing.T) { } } +// TestFeedContent verifies that each FeedContent option controls what a feed event carries: the body, +// the xattrs, both, or neither. func TestFeedContent(t *testing.T) { tests := []struct { name string diff --git a/queue.go b/queue.go index a68549a..1e16548 100644 --- a/queue.go +++ b/queue.go @@ -56,6 +56,13 @@ func (q *queue[T]) pull() (result T) { return } +// closed returns true if the queue was closed, rather than drained to its end-of-feed marker. +func (q *queue[T]) closed() bool { + q.cond.L.Lock() + defer q.cond.L.Unlock() + return q.list == nil +} + func (q *queue[T]) close() { q.cond.L.Lock() if q.list != nil { From ed4a2df59a9547802736520bd31a06017d5bc0d4 Mon Sep 17 00:00:00 2001 From: Ritesh Kumar Date: Mon, 21 Sep 2026 17:43:06 +0530 Subject: [PATCH 2/6] Delete a dump feed's checkpoint once every collection completes The checkpoint is shared by the collections in a feed, so StartDCPFeed drops it once they have all drained to the end, rather than each feed doing so in run(). A terminated feed keeps it so the next run can resume. Guards against a nil metadata store. Co-Authored-By: Claude Opus 5 --- feeds.go | 45 ++++++++++++------- feeds_test.go | 117 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 146 insertions(+), 16 deletions(-) diff --git a/feeds.go b/feeds.go index ccf10aa..1a817a4 100644 --- a/feeds.go +++ b/feeds.go @@ -25,18 +25,16 @@ var activeFeedCount int32 // for tests func (bucket *Bucket) StartDCPFeed(ctx context.Context, args sgbucket.FeedArguments, callback sgbucket.FeedEventCallbackFunc, dbStats *expvar.Map) error { traceEnter("StartDCPFeed", "bucket=%s, args=%+v", bucket.GetName(), args) - // If no scopes are specified, return feed for the default collection, if it exists + // Validate requested collections exist before starting feeds. No scopes means the default + // collection, if it exists. + requestedCollections := make([]*Collection, 0) if len(args.Scopes) == 0 { collection, err := bucket.getCollection(defaultDataStoreName) if err != nil { return err } - _, err = collection.startDCPFeed(ctx, args, callback, dbStats) - return err + requestedCollections = append(requestedCollections, collection) } - - // Validate requested collections exist before starting feeds - requestedCollections := make([]*Collection, 0) for scopeName, collections := range args.Scopes { for _, collectionName := range collections { collection, err := bucket.getCollection(sgbucket.DataStoreNameImpl{Scope: scopeName, Collection: collectionName}) @@ -80,6 +78,13 @@ func (bucket *Bucket) StartDCPFeed(ctx context.Context, args sgbucket.FeedArgume for _, collection := range requestedCollections { <-doneChans[collection] } + // The checkpoint is shared by every collection in the feed, so drop it only once they have all + // streamed to the end. A single terminated collection means it is still needed to resume. + if args.Dump && allFeedsCompleted(startedFeeds) { + if err := startedFeeds[0].deleteCheckpoint(); err != nil { + logError("Error deleting checkpoint %q: %v", args.CheckpointPrefix, err) + } + } if doneChan != nil { close(doneChan) } @@ -217,6 +222,8 @@ type dcpFeed struct { events eventQueue lastCas CAS lastCasChanged bool + // completed reports that the feed drained to its end-of-feed marker, rather than being terminated. + completed atomic.Bool } func (feed *dcpFeed) String() string { @@ -324,14 +331,9 @@ func (feed *dcpFeed) run() { } debug("%s stopping", feed) - // A dump feed that reached its end-of-feed marker streamed everything, so its checkpoint is spent. - // A terminated feed keeps its checkpoint so the next run can resume. - if feed.args.Dump && !feed.events.closed() { - if err := feed.deleteCheckpoint(); err != nil { - logError("Error deleting %s checkpoint: %v", feed, err) - } - return - } + // Recorded before the deferred close of DoneChan, so a waiter sees a settled value. The checkpoint + // is shared by every collection in the feed, so only StartDCPFeed can decide to delete it. + feed.completed.Store(!feed.events.closed()) if feed.lastCasChanged { if err := feed.writeCheckpoint(); err != nil { @@ -340,9 +342,20 @@ func (feed *dcpFeed) run() { } } -// Deletes the feed's checkpoint document, if there is one. +// allFeedsCompleted reports whether every feed drained to its end-of-feed marker. +func allFeedsCompleted(feeds []*dcpFeed) bool { + for _, feed := range feeds { + if !feed.completed.Load() { + return false + } + } + return len(feeds) > 0 +} + +// Deletes the feed's checkpoint document, if there is one. A feed given no metadata store never wrote +// one, so there is nothing to delete. func (feed *dcpFeed) deleteCheckpoint() error { - if feed.args.CheckpointPrefix == "" { + if feed.args.CheckpointPrefix == "" || feed.metadataStore == nil { return nil } err := feed.metadataStore.Delete(feed.ctx, feed.checkpointKey()) diff --git a/feeds_test.go b/feeds_test.go index 6f62d26..8105104 100644 --- a/feeds_test.go +++ b/feeds_test.go @@ -207,6 +207,12 @@ func TestDumpDeletesCheckpoint(t *testing.T) { event = <-events assert.Equal(t, sgbucket.FeedOpBeginBackfill, event.Opcode) readExpectedEventsABC(t, events) + + // the first feed wrote a checkpoint before the feed deleted it, so its tombstone is in the backfill + e := <-events + assert.Equal(t, sgbucket.FeedOpDeletion, e.Opcode) + assert.Equal(t, prefix, string(e.Key)) + event = <-events assert.Equal(t, sgbucket.FeedOpEndBackfill, event.Opcode) @@ -214,6 +220,117 @@ func TestDumpDeletesCheckpoint(t *testing.T) { assert.False(t, ok) } +// TestDumpWithoutMetadataStore verifies that a dump feed given a checkpoint prefix but no metadata +// store finishes without panicking. Such a feed never wrote a checkpoint, so there is none to delete. +func TestDumpWithoutMetadataStore(t *testing.T) { + ensureNoLeakedFeeds(t) + bucket := makeTestBucket(t) + ctx := t.Context() + + doneChan := make(chan struct{}) + args := sgbucket.FeedArguments{ + ID: "myID", + Backfill: 0, + Dump: true, + CheckpointPrefix: "NoMetadataStoreCheckpoint", + DoneChan: doneChan, + } + require.NoError(t, bucket.StartDCPFeed(ctx, args, func(sgbucket.FeedEvent) bool { return true }, nil)) + <-doneChan +} + +// TestDumpAcrossCollectionsDeletesCheckpointOnlyWhenAllComplete verifies that the checkpoint shared by +// a multi-collection feed survives while any collection is still using it. The feed runs one goroutine +// per collection, so one collection finishing must not drop the document the others depend on. +func TestDumpAcrossCollectionsDeletesCheckpointOnlyWhenAllComplete(t *testing.T) { + const prefix = "MultiCollectionDumpCheckpoint" + scopes := map[string][]string{"scope1": {"collection1", "collection2"}} + + newArgs := func(terminator chan bool) sgbucket.FeedArguments { + return sgbucket.FeedArguments{ + ID: "myID", + Backfill: sgbucket.FeedResume, + Dump: true, + Terminator: terminator, + CheckpointPrefix: prefix, + Scopes: scopes, + } + } + + seed := func(t *testing.T, bucket *Bucket) (*Collection, uint32) { + ctx := t.Context() + c1, err := bucket.NamedDataStore(ctx, dsName("scope1", "collection1")) + require.NoError(t, err) + c2, err := bucket.NamedDataStore(ctx, dsName("scope1", "collection2")) + require.NoError(t, err) + for i := range 500 { + _, err := c1.Add(ctx, fmt.Sprintf("c1-doc%d", i), 0, "V") + require.NoError(t, err) + _, err = c2.Add(ctx, fmt.Sprintf("c2-doc%d", i), 0, "V") + require.NoError(t, err) + } + return bucket.DefaultDataStore(ctx).(*Collection), c2.GetCollectionID() + } + + t.Run("all collections complete", func(t *testing.T) { + ensureNoLeakedFeeds(t) + bucket := makeTestBucket(t) + ctx := t.Context() + metadataStore, _ := seed(t, bucket) + + args := newArgs(nil) + args.MetadataStore = metadataStore + doneChan := make(chan struct{}) + args.DoneChan = doneChan + require.NoError(t, bucket.StartDCPFeed(ctx, args, func(sgbucket.FeedEvent) bool { return true }, nil)) + <-doneChan + + var checkpt checkpoint + _, err := metadataStore.Get(ctx, prefix, &checkpt) + require.Error(t, err) + require.IsType(t, sgbucket.MissingError{}, err) + }) + + t.Run("one collection terminated", func(t *testing.T) { + ensureNoLeakedFeeds(t) + bucket := makeTestBucket(t) + ctx := t.Context() + metadataStore, c2ID := seed(t, bucket) + + terminator := make(chan bool) + doneChan := make(chan struct{}) + args := newArgs(terminator) + args.MetadataStore = metadataStore + args.DoneChan = doneChan + + firstC2 := make(chan struct{}) + release := make(chan struct{}) + var once sync.Once + callback := func(event sgbucket.FeedEvent) bool { + // hold collection2 inside its own goroutine so it is still running when the feed is stopped + if event.CollectionID == c2ID && event.Opcode == sgbucket.FeedOpMutation { + once.Do(func() { + close(firstC2) + <-release + }) + } + return true + } + require.NoError(t, bucket.StartDCPFeed(ctx, args, callback, nil)) + + <-firstC2 + close(terminator) + close(release) + <-doneChan + + // collection2 was cut short, so the shared checkpoint is still needed + var checkpt checkpoint + _, err := metadataStore.Get(ctx, prefix, &checkpt) + require.NoError(t, err) + assert.NotEmpty(t, checkpt.LastCas) + }) +} + // TestDumpKeepsCheckpointWhenTerminated verifies that a dump feed stopped before it streams // everything keeps its checkpoint, so the next feed can resume from where it stopped. func TestDumpKeepsCheckpointWhenTerminated(t *testing.T) { From 60eaa72909764bee0fc8127d691e452577d3a1ba Mon Sep 17 00:00:00 2001 From: Ritesh Kumar Date: Mon, 21 Sep 2026 20:08:39 +0530 Subject: [PATCH 3/6] Require Go 1.27 and use errors.AsType errors.AsType needs Go 1.26 or later, so the go directive and the CI toolchain move up together. Co-Authored-By: Claude Opus 5 --- .github/workflows/ci.yml | 2 +- feeds.go | 3 ++- go.mod | 2 +- 3 files changed, 4 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 865374f..b9f2d55 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -27,7 +27,7 @@ env: ACTIONS_SETUP_GO_VERSION: &actions_setup_go_version actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 ACTIONS_GOLANGCI_LINT_VERSION: &actions_golangci_lint_version golangci/golangci-lint-action@ba0d7d2ec06a0ea1cb5fa41b2e4a3ab91d21278a # v9.3.0 ACTIONS_GO_TEST_ANNOTATIONS_VERSION: &actions_go_test_annotations_version guyarb/golang-test-annotations@96fc379b171c49932041d6c789e73331a7bdeec1 # v0.9.0 - GO_VERSION: &go_version 1.26.6 + GO_VERSION: &go_version 1.27.1 jobs: addlicense: diff --git a/feeds.go b/feeds.go index 1a817a4..31fd21c 100644 --- a/feeds.go +++ b/feeds.go @@ -11,6 +11,7 @@ package rosmar import ( "context" "encoding/json" + "errors" "expvar" "fmt" "sync/atomic" @@ -359,7 +360,7 @@ func (feed *dcpFeed) deleteCheckpoint() error { return nil } err := feed.metadataStore.Delete(feed.ctx, feed.checkpointKey()) - if _, ok := err.(sgbucket.MissingError); ok { + if _, ok := errors.AsType[sgbucket.MissingError](err); ok { return nil } return err diff --git a/go.mod b/go.mod index aeddc3a..49079ea 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/couchbaselabs/rosmar -go 1.25.0 +go 1.27.0 require ( github.com/couchbase/sg-bucket v0.0.0-20260714130240-329df715dd61 From 4471e62174bf8c0e325e2745b6d68af588493a1f Mon Sep 17 00:00:00 2001 From: Ritesh Kumar Date: Tue, 22 Sep 2026 17:40:52 +0530 Subject: [PATCH 4/6] Target Go 1.26 and stop the terminated dump tests racing No released golangci-lint targets Go 1.27, so the directive stays at the lowest version errors.AsType compiles on. The terminated feed tests now step one event at a time, so the feed yields and the terminator always closes the queue first. Co-Authored-By: Claude Opus 5 --- .github/workflows/ci.yml | 2 +- feeds_test.go | 57 ++++++++++++++++++++++------------------ go.mod | 2 +- 3 files changed, 33 insertions(+), 28 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b9f2d55..865374f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -27,7 +27,7 @@ env: ACTIONS_SETUP_GO_VERSION: &actions_setup_go_version actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 ACTIONS_GOLANGCI_LINT_VERSION: &actions_golangci_lint_version golangci/golangci-lint-action@ba0d7d2ec06a0ea1cb5fa41b2e4a3ab91d21278a # v9.3.0 ACTIONS_GO_TEST_ANNOTATIONS_VERSION: &actions_go_test_annotations_version guyarb/golang-test-annotations@96fc379b171c49932041d6c789e73331a7bdeec1 # v0.9.0 - GO_VERSION: &go_version 1.27.1 + GO_VERSION: &go_version 1.26.6 jobs: addlicense: diff --git a/feeds_test.go b/feeds_test.go index 8105104..31c099f 100644 --- a/feeds_test.go +++ b/feeds_test.go @@ -303,25 +303,27 @@ func TestDumpAcrossCollectionsDeletesCheckpointOnlyWhenAllComplete(t *testing.T) args.MetadataStore = metadataStore args.DoneChan = doneChan - firstC2 := make(chan struct{}) - release := make(chan struct{}) - var once sync.Once + // step collection2 one event at a time, so it is still running when the feed is stopped + step := make(chan struct{}) callback := func(event sgbucket.FeedEvent) bool { - // hold collection2 inside its own goroutine so it is still running when the feed is stopped if event.CollectionID == c2ID && event.Opcode == sgbucket.FeedOpMutation { - once.Do(func() { - close(firstC2) - <-release - }) + <-step } return true } require.NoError(t, bucket.StartDCPFeed(ctx, args, callback, nil)) - <-firstC2 + for range 3 { + step <- struct{}{} + } close(terminator) - close(release) - <-doneChan + for done := false; !done; { + select { + case <-doneChan: + done = true + case step <- struct{}{}: + } + } // collection2 was cut short, so the shared checkpoint is still needed var checkpt checkpoint @@ -339,7 +341,6 @@ func TestDumpKeepsCheckpointWhenTerminated(t *testing.T) { ctx := t.Context() c := bucket.DefaultDataStore(ctx) - // enough documents that the feed is still draining when the terminator closes const numDocs = 1000 for i := range numDocs { addToCollection(t, c, fmt.Sprintf("doc%d", i), 0, "V") @@ -348,21 +349,16 @@ func TestDumpKeepsCheckpointWhenTerminated(t *testing.T) { const prefix = "TerminatedDumpCheckpoint" terminator := make(chan bool) doneChan := make(chan struct{}) - firstMutation := make(chan struct{}) - release := make(chan struct{}) + // Advance the feed one event at a time. Blocking between events gives the terminator a scheduling + // point to close the queue, which a free-running feed can otherwise outrun. + step := make(chan struct{}) var processed atomic.Int64 - var once sync.Once callback := func(event sgbucket.FeedEvent) bool { - if event.Opcode != sgbucket.FeedOpMutation { - return true + if event.Opcode == sgbucket.FeedOpMutation { + processed.Add(1) + <-step } - processed.Add(1) - // hold the feed inside its own goroutine while the test terminates it - once.Do(func() { - close(firstMutation) - <-release - }) return true } @@ -377,10 +373,19 @@ func TestDumpKeepsCheckpointWhenTerminated(t *testing.T) { } require.NoError(t, bucket.StartDCPFeed(ctx, args, callback, nil)) - <-firstMutation + // let a few documents through, so the feed has progress worth checkpointing + for range 3 { + step <- struct{}{} + } + close(terminator) - close(release) - <-doneChan + for done := false; !done; { + select { + case <-doneChan: + done = true + case step <- struct{}{}: + } + } // the feed was cut short, so it never reached its end-of-feed marker assert.Less(t, processed.Load(), int64(numDocs)) diff --git a/go.mod b/go.mod index 49079ea..4dbe7eb 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/couchbaselabs/rosmar -go 1.27.0 +go 1.26.0 require ( github.com/couchbase/sg-bucket v0.0.0-20260714130240-329df715dd61 From e30642cccbb0c76a333d87ebb69901ac599f9534 Mon Sep 17 00:00:00 2001 From: Ritesh Kumar Date: Fri, 25 Sep 2026 16:38:01 +0530 Subject: [PATCH 5/6] Remove only a dump's own checkpoint entries once it completes Feeds on different collections can share a checkpoint document. A dump now saves it once, after all of its collections stop. A resumed dump that streamed everything removes only its own entries, and deletes the document when none are left. A dump that did not resume never removes entries. A feed that stops on a corrupt event now counts as cut short. pull() reports a closed queue under the same lock, so a stop after the end-of-feed marker no longer marks a completed feed as terminated. A FeedResume feed with no MetadataStore fails to start, and other feeds with no store skip the checkpoint instead of panicking. Co-Authored-By: Claude Opus 5.5 --- feeds.go | 134 +++++++++++++++++++++--------------- feeds_test.go | 186 ++++++++++++++++++++++++++++++++++++++++++++------ queue.go | 12 +--- queue_test.go | 57 ++++++++++++++++ 4 files changed, 305 insertions(+), 84 deletions(-) create mode 100644 queue_test.go diff --git a/feeds.go b/feeds.go index 31fd21c..17c4af3 100644 --- a/feeds.go +++ b/feeds.go @@ -79,12 +79,9 @@ func (bucket *Bucket) StartDCPFeed(ctx context.Context, args sgbucket.FeedArgume for _, collection := range requestedCollections { <-doneChans[collection] } - // The checkpoint is shared by every collection in the feed, so drop it only once they have all - // streamed to the end. A single terminated collection means it is still needed to resume. - if args.Dump && allFeedsCompleted(startedFeeds) { - if err := startedFeeds[0].deleteCheckpoint(); err != nil { - logError("Error deleting checkpoint %q: %v", args.CheckpointPrefix, err) - } + // The collections share one checkpoint document, so a dump saves it once they have all stopped. + if args.Dump { + _ = saveDumpCheckpoint(startedFeeds) } if doneChan != nil { close(doneChan) @@ -119,6 +116,9 @@ func (c *Collection) startDCPFeed(ctx context.Context, args sgbucket.FeedArgumen if args.CheckpointPrefix == "" { return nil, fmt.Errorf("feed's Backfill is FeedResume but no CheckpointPrefix given") } + if args.MetadataStore == nil { + return nil, fmt.Errorf("feed's Backfill is FeedResume but no MetadataStore given") + } if err := feed.readCheckpoint(); err != nil { return nil, fmt.Errorf("couldn't read DCP feed checkpoint: %w", err) } @@ -235,15 +235,20 @@ func (feed *dcpFeed) checkpointKey() string { return feed.args.CheckpointPrefix } +// usesCheckpoint reports whether the feed has somewhere to keep a checkpoint. +func (feed *dcpFeed) usesCheckpoint() bool { + return feed.args.CheckpointPrefix != "" && feed.metadataStore != nil +} + // Reads the feed's lastCas from the checkpoint document, if there is one. func (feed *dcpFeed) readCheckpoint() (err error) { - if feed.args.CheckpointPrefix == "" { + if !feed.usesCheckpoint() { return } key := feed.checkpointKey() var checkpt checkpoint if _, err = feed.metadataStore.Get(feed.ctx, key, &checkpt); err != nil { - if _, ok := err.(sgbucket.MissingError); ok { + if _, ok := errors.AsType[sgbucket.MissingError](err); ok { err = nil debug("%s checkpoint %q missing", feed, key) } else { @@ -260,14 +265,32 @@ func (feed *dcpFeed) readCheckpoint() (err error) { } // Writes the feed's lastCas to the checkpoint document, if there is one. -func (feed *dcpFeed) writeCheckpoint() (err error) { - if feed.args.CheckpointPrefix == "" || !feed.lastCasChanged { - return +func (feed *dcpFeed) writeCheckpoint() error { + if !feed.lastCasChanged { + return nil } - key := feed.checkpointKey() - colID := feed.collection.GetCollectionID() + return updateCheckpoint([]*dcpFeed{feed}, false) +} - _, err = feed.metadataStore.Update(feed.ctx, key, 0, func(current []byte) (updated []byte, expiry *uint32, delete bool, err error) { +// saveDumpCheckpoint records a dump's progress. A resumed dump whose collections all streamed to the end has +// nothing left to resume, so it removes their entries instead. +func saveDumpCheckpoint(feeds []*dcpFeed) error { + if len(feeds) == 0 { + return nil + } + remove := feeds[0].args.Backfill == sgbucket.FeedResume && allFeedsCompleted(feeds) + return updateCheckpoint(feeds, remove) +} + +// updateCheckpoint writes each feed's lastCas to their shared checkpoint document in one update. With remove, +// it drops the feeds' entries instead, and deletes the document once none are left. +func updateCheckpoint(feeds []*dcpFeed, remove bool) error { + feed := feeds[0] + if !feed.usesCheckpoint() { + return nil + } + key := feed.checkpointKey() + _, err := feed.metadataStore.Update(feed.ctx, key, 0, func(current []byte) (updated []byte, expiry *uint32, del bool, err error) { var checkpt checkpoint if current != nil { if err := json.Unmarshal(current, &checkpt); err != nil { @@ -277,8 +300,24 @@ func (feed *dcpFeed) writeCheckpoint() (err error) { if checkpt.LastCas == nil { checkpt.LastCas = make(map[uint32]uint64) } - checkpt.LastCas[colID] = feed.lastCas - + changed := false + for _, f := range feeds { + colID := f.collection.GetCollectionID() + if remove { + _, found := checkpt.LastCas[colID] + changed = changed || found + delete(checkpt.LastCas, colID) + } else if f.lastCasChanged { + checkpt.LastCas[colID] = f.lastCas + changed = true + } + } + if !changed { + return nil, nil, false, nil // cancels the update + } + if len(checkpt.LastCas) == 0 { + return nil, nil, true, nil + } updated, err = json.Marshal(checkpt) if err != nil { return nil, nil, false, fmt.Errorf("failed to marshal checkpoint: %w", err) @@ -287,9 +326,9 @@ func (feed *dcpFeed) writeCheckpoint() (err error) { }) if err == nil { - debug("%s wrote lastCas 0x%x to %q", feed, feed.lastCas, key) + debug("%s updated checkpoint %q, remove=%v", feed, key, remove) } else { - logError("%s failed to write checkpoint to %q: %v", feed, key, err) + logError("%s failed to update checkpoint %q: %v", feed, key, err) } return err } @@ -313,33 +352,33 @@ func (feed *dcpFeed) run() { collectionID := feed.collection.GetCollectionID() feedContent := feed.args.FeedContent for { - if e := feed.events.pull(); e != nil { - feedEvent, err := e.asFeedEvent(collectionID, feedContent) - if err != nil { - logError("Fatal error converting %s event to feed event: %v", feed, err) - break - } - feed.callback(*feedEvent) - if feedEvent.Cas > feed.lastCas { - feed.lastCas = feedEvent.Cas - feed.lastCasChanged = true - debug("%s lastCas = 0x%x", feed, feed.lastCas) - // TODO: Set a timer to write the checkpoint "soon" - } - } else { + e, ok := feed.events.pull() + if !ok { + break // terminated + } + if e == nil { + // Stored before the deferred close of DoneChan, so a waiter sees a settled value. + feed.completed.Store(true) break } + feedEvent, err := e.asFeedEvent(collectionID, feedContent) + if err != nil { + logError("Fatal error converting %s event to feed event: %v", feed, err) + break + } + feed.callback(*feedEvent) + if feedEvent.Cas > feed.lastCas { + feed.lastCas = feedEvent.Cas + feed.lastCasChanged = true + debug("%s lastCas = 0x%x", feed, feed.lastCas) + // TODO: Set a timer to write the checkpoint "soon" + } } debug("%s stopping", feed) - // Recorded before the deferred close of DoneChan, so a waiter sees a settled value. The checkpoint - // is shared by every collection in the feed, so only StartDCPFeed can decide to delete it. - feed.completed.Store(!feed.events.closed()) - - if feed.lastCasChanged { - if err := feed.writeCheckpoint(); err != nil { - logError("Error saving %s checkpoint: %v", feed, err) - } + // A dump saves its checkpoint in StartDCPFeed, once all of its collections stop. + if !feed.args.Dump { + _ = feed.writeCheckpoint() } } @@ -350,20 +389,7 @@ func allFeedsCompleted(feeds []*dcpFeed) bool { return false } } - return len(feeds) > 0 -} - -// Deletes the feed's checkpoint document, if there is one. A feed given no metadata store never wrote -// one, so there is nothing to delete. -func (feed *dcpFeed) deleteCheckpoint() error { - if feed.args.CheckpointPrefix == "" || feed.metadataStore == nil { - return nil - } - err := feed.metadataStore.Delete(feed.ctx, feed.checkpointKey()) - if _, ok := errors.AsType[sgbucket.MissingError](err); ok { - return nil - } - return err + return true } func (feed *dcpFeed) close() { diff --git a/feeds_test.go b/feeds_test.go index 31c099f..64d1e6d 100644 --- a/feeds_test.go +++ b/feeds_test.go @@ -158,14 +158,16 @@ func TestCheckpoint(t *testing.T) { assert.False(t, ok) } -// TestDumpDeletesCheckpoint verifies that a dump feed deletes its checkpoint once it has streamed -// everything, so a second dump feed with the same CheckpointPrefix replays the collection from the -// start rather than resuming. +// TestDumpDeletesCheckpoint verifies that a resumed dump feed leaves no checkpoint once it has streamed +// everything, so a later dump feed with the same CheckpointPrefix replays the collection from the start. func TestDumpDeletesCheckpoint(t *testing.T) { ensureNoLeakedFeeds(t) bucket := makeTestBucket(t) ctx := t.Context() c := bucket.DefaultDataStore(ctx) + // keep the checkpoint out of the streamed collection, so its writes are not in the events + metadataStore, err := bucket.getOrCreateCollection(sgbucket.DataStoreNameImpl{Scope: "S", Collection: "Meta"}, true) + require.NoError(t, err) addToCollection(t, c, "able", 0, "A") addToCollection(t, c, "baker", 0, "B") @@ -178,11 +180,12 @@ func TestDumpDeletesCheckpoint(t *testing.T) { }, ID: "myID", Backfill: sgbucket.FeedResume, - Dump: true, + Terminator: make(chan bool), CheckpointPrefix: prefix, + MetadataStore: metadataStore, } - // A dump feed ends on its own once it has streamed everything. + // A terminated feed leaves a checkpoint for the dump to resume from. events, doneChan := startFeedWithArgs(t, bucket, args) event := <-events @@ -191,28 +194,45 @@ func TestDumpDeletesCheckpoint(t *testing.T) { event = <-events assert.Equal(t, sgbucket.FeedOpEndBackfill, event.Opcode) + close(args.Terminator) _, ok := <-doneChan assert.False(t, ok) - // The checkpoint is spent, so the feed removed it. var checkpt checkpoint - _, err := c.Get(ctx, prefix, &checkpt) + _, err = metadataStore.Get(ctx, prefix, &checkpt) + require.NoError(t, err) + + addToCollection(t, c, "delta", 0, "D") + + // The dump resumes after charlie and streams to the end, so it has nothing left to resume. + t.Logf("---- Dump feed, resuming from checkpoint ---") + args.Terminator = nil + args.Dump = true + events, doneChan = startFeedWithArgs(t, bucket, args) + + event = <-events + assert.Equal(t, sgbucket.FeedOpBeginBackfill, event.Opcode) + event = <-events + assert.Equal(t, "delta", string(event.Key)) + event = <-events + assert.Equal(t, sgbucket.FeedOpEndBackfill, event.Opcode) + + _, ok = <-doneChan + assert.False(t, ok) + + _, err = metadataStore.Get(ctx, prefix, &checkpt) require.Error(t, err) require.IsType(t, sgbucket.MissingError{}, err) - // With no checkpoint to resume from, a second dump feed replays the same documents. + // With no checkpoint to resume from, a second dump feed replays every document. t.Logf("---- Second dump feed, no checkpoint to resume from ---") events, doneChan = startFeedWithArgs(t, bucket, args) event = <-events assert.Equal(t, sgbucket.FeedOpBeginBackfill, event.Opcode) readExpectedEventsABC(t, events) - - // the first feed wrote a checkpoint before the feed deleted it, so its tombstone is in the backfill - e := <-events - assert.Equal(t, sgbucket.FeedOpDeletion, e.Opcode) - assert.Equal(t, prefix, string(e.Key)) - + event = <-events + assert.Equal(t, "delta", string(event.Key)) event = <-events assert.Equal(t, sgbucket.FeedOpEndBackfill, event.Opcode) @@ -220,12 +240,13 @@ func TestDumpDeletesCheckpoint(t *testing.T) { assert.False(t, ok) } -// TestDumpWithoutMetadataStore verifies that a dump feed given a checkpoint prefix but no metadata -// store finishes without panicking. Such a feed never wrote a checkpoint, so there is none to delete. +// TestDumpWithoutMetadataStore verifies that a dump feed given a checkpoint prefix but no metadata store +// streams its documents without panicking, because it has nowhere to write a checkpoint. func TestDumpWithoutMetadataStore(t *testing.T) { ensureNoLeakedFeeds(t) bucket := makeTestBucket(t) ctx := t.Context() + addToCollection(t, bucket.DefaultDataStore(ctx), "able", 0, "A") doneChan := make(chan struct{}) args := sgbucket.FeedArguments{ @@ -239,9 +260,24 @@ func TestDumpWithoutMetadataStore(t *testing.T) { <-doneChan } -// TestDumpAcrossCollectionsDeletesCheckpointOnlyWhenAllComplete verifies that the checkpoint shared by -// a multi-collection feed survives while any collection is still using it. The feed runs one goroutine -// per collection, so one collection finishing must not drop the document the others depend on. +// TestResumeWithoutMetadataStore verifies that a FeedResume feed with no metadata store fails to start, +// rather than silently replaying from the start. +func TestResumeWithoutMetadataStore(t *testing.T) { + ensureNoLeakedFeeds(t) + bucket := makeTestBucket(t) + + args := sgbucket.FeedArguments{ + ID: "myID", + Backfill: sgbucket.FeedResume, + Dump: true, + CheckpointPrefix: "NoMetadataStoreCheckpoint", + } + err := bucket.StartDCPFeed(t.Context(), args, func(sgbucket.FeedEvent) bool { return true }, nil) + require.ErrorContains(t, err, "no MetadataStore given") +} + +// TestDumpAcrossCollectionsDeletesCheckpointOnlyWhenAllComplete verifies that the checkpoint shared by a +// multi-collection feed survives while any of its collections was cut short. func TestDumpAcrossCollectionsDeletesCheckpointOnlyWhenAllComplete(t *testing.T) { const prefix = "MultiCollectionDumpCheckpoint" scopes := map[string][]string{"scope1": {"collection1", "collection2"}} @@ -316,6 +352,8 @@ func TestDumpAcrossCollectionsDeletesCheckpointOnlyWhenAllComplete(t *testing.T) for range 3 { step <- struct{}{} } + // collection1 is not stepped, so wait for its feed to reach the end before stopping collection2 + require.Eventually(t, func() bool { return atomic.LoadInt32(&activeFeedCount) == 1 }, 10*time.Second, time.Millisecond) close(terminator) for done := false; !done; { select { @@ -325,11 +363,23 @@ func TestDumpAcrossCollectionsDeletesCheckpointOnlyWhenAllComplete(t *testing.T) } } - // collection2 was cut short, so the shared checkpoint is still needed + // collection2 was cut short, so the shared checkpoint keeps both collections' progress var checkpt checkpoint _, err := metadataStore.Get(ctx, prefix, &checkpt) require.NoError(t, err) - assert.NotEmpty(t, checkpt.LastCas) + assert.Len(t, checkpt.LastCas, 2) + + // resumed without a terminator, both collections stream to the end and remove their entries + args = newArgs(nil) + args.MetadataStore = metadataStore + doneChan = make(chan struct{}) + args.DoneChan = doneChan + require.NoError(t, bucket.StartDCPFeed(ctx, args, func(sgbucket.FeedEvent) bool { return true }, nil)) + <-doneChan + + _, err = metadataStore.Get(ctx, prefix, &checkpt) + require.Error(t, err) + require.IsType(t, sgbucket.MissingError{}, err) }) } @@ -397,6 +447,100 @@ func TestDumpKeepsCheckpointWhenTerminated(t *testing.T) { assert.NotEmpty(t, checkpt.LastCas) } +// TestDumpKeepsOtherFeedsCheckpointEntries verifies that a completed dump removes only its own collection's +// entry from a checkpoint document that a feed on another collection shares. +func TestDumpKeepsOtherFeedsCheckpointEntries(t *testing.T) { + ensureNoLeakedFeeds(t) + bucket := makeTestBucket(t) + ctx := t.Context() + c1 := bucket.DefaultDataStore(ctx).(*Collection) + c2, err := bucket.getOrCreateCollection(sgbucket.DataStoreNameImpl{Scope: "S", Collection: "C"}, true) + require.NoError(t, err) + addToCollection(t, c1, "c1-doc", 0, "V1") + addToCollection(t, c2, "c2-doc", 0, "V2") + + const prefix = "SharedDumpCheckpoint" + runFeedToEndOfBackfill(t, bucket, sgbucket.FeedArguments{ID: "id1", Backfill: sgbucket.FeedResume, CheckpointPrefix: prefix}) + runFeedToEndOfBackfill(t, bucket, sgbucket.FeedArguments{ID: "id2", Backfill: sgbucket.FeedResume, CheckpointPrefix: prefix, + Scopes: map[string][]string{"S": {"C"}}}) + + var checkpt checkpoint + _, err = c1.Get(ctx, prefix, &checkpt) + require.NoError(t, err) + require.Len(t, checkpt.LastCas, 2) + c2LastCas := checkpt.LastCas[c2.GetCollectionID()] + + runFeedToEndOfBackfill(t, bucket, sgbucket.FeedArguments{ID: "id1", Backfill: sgbucket.FeedResume, Dump: true, CheckpointPrefix: prefix}) + + checkpt = checkpoint{} + _, err = c1.Get(ctx, prefix, &checkpt) + require.NoError(t, err) + assert.Equal(t, map[uint32]uint64{c2.GetCollectionID(): c2LastCas}, checkpt.LastCas) +} + +// TestDumpWithoutResumeKeepsCheckpoint verifies that a dump that did not resume from the checkpoint leaves +// it in place for the feed that did write it. +func TestDumpWithoutResumeKeepsCheckpoint(t *testing.T) { + ensureNoLeakedFeeds(t) + bucket := makeTestBucket(t) + ctx := t.Context() + c := bucket.DefaultDataStore(ctx) + addToCollection(t, c, "able", 0, "A") + + const prefix = "NonResumeDumpCheckpoint" + runFeedToEndOfBackfill(t, bucket, sgbucket.FeedArguments{ID: "id1", Backfill: sgbucket.FeedResume, CheckpointPrefix: prefix}) + runFeedToEndOfBackfill(t, bucket, sgbucket.FeedArguments{ID: "id2", Backfill: 0, Dump: true, CheckpointPrefix: prefix}) + + var checkpt checkpoint + _, err := c.Get(ctx, prefix, &checkpt) + require.NoError(t, err) + assert.NotEmpty(t, checkpt.LastCas) +} + +// TestDumpStoppedByCorruptEventKeepsCheckpoint verifies that a dump that stops on an event it cannot convert +// counts as cut short, so it keeps the progress it made for the next resume. +func TestDumpStoppedByCorruptEventKeepsCheckpoint(t *testing.T) { + ensureNoLeakedFeeds(t) + bucket := makeTestBucket(t) + ctx := t.Context() + c := bucket.DefaultDataStore(ctx).(*Collection) + + addToCollection(t, c, "able", 0, "A") + addToCollection(t, c, "corrupt", 0, "B") + _, err := c.db().Exec(`UPDATE documents SET xattrs=? WHERE collection=? AND key=?`, + []byte(`not valid json`), c.id, "corrupt") + require.NoError(t, err) + + const prefix = "CorruptDumpCheckpoint" + runFeedToEndOfBackfill(t, bucket, sgbucket.FeedArguments{ID: "id1", Backfill: sgbucket.FeedResume, Dump: true, CheckpointPrefix: prefix}) + + ableCas, err := c.Get(ctx, "able", nil) + require.NoError(t, err) + var checkpt checkpoint + _, err = c.Get(ctx, prefix, &checkpt) + require.NoError(t, err) + assert.Equal(t, map[uint32]uint64{c.GetCollectionID(): ableCas}, checkpt.LastCas) +} + +// runFeedToEndOfBackfill runs a feed that discards its events, and waits until it stops. A feed that is not a +// dump is terminated once its backfill ends. +func runFeedToEndOfBackfill(t *testing.T, bucket *Bucket, args sgbucket.FeedArguments) { + args.DoneChan = make(chan struct{}) + args.MetadataStore = bucket.DefaultDataStore(t.Context()) + terminator := make(chan bool) + if !args.Dump { + args.Terminator = terminator + } + callback := func(event sgbucket.FeedEvent) bool { + if event.Opcode == sgbucket.FeedOpEndBackfill && !args.Dump { + close(terminator) + } + return true + } + require.NoError(t, bucket.StartDCPFeed(t.Context(), args, callback, nil)) + <-args.DoneChan +} + // TestResumeFromCheckpoint verifies that a resumed feed delivers only the documents written while it // was stopped, and that it sees the checkpoint document itself as a mutation. func TestResumeFromCheckpoint(t *testing.T) { diff --git a/queue.go b/queue.go index 1e16548..4e4fe0b 100644 --- a/queue.go +++ b/queue.go @@ -41,8 +41,8 @@ func (q *queue[T]) push(value T) (ok bool) { } // Removes the last/oldest value from the queue; if the queue is empty, blocks. -// If the queue is closed while blocking, returns a default-initialized T. -func (q *queue[T]) pull() (result T) { +// If the queue is closed while blocking, returns a default-initialized T and ok=false. +func (q *queue[T]) pull() (result T, ok bool) { q.cond.L.Lock() defer q.cond.L.Unlock() for q.list != nil && q.list.Len() == 0 { @@ -52,17 +52,11 @@ func (q *queue[T]) pull() (result T) { last := q.list.Back() q.list.Remove(last) result = last.Value.(T) + ok = true } return } -// closed returns true if the queue was closed, rather than drained to its end-of-feed marker. -func (q *queue[T]) closed() bool { - q.cond.L.Lock() - defer q.cond.L.Unlock() - return q.list == nil -} - func (q *queue[T]) close() { q.cond.L.Lock() if q.list != nil { diff --git a/queue_test.go b/queue_test.go new file mode 100644 index 0000000..28b1687 --- /dev/null +++ b/queue_test.go @@ -0,0 +1,57 @@ +// Copyright 2026-Present Couchbase, Inc. +// +// Use of this software is governed by the Business Source License included +// in the file licenses/BSL-Couchbase.txt. As of the Change Date specified +// in that file, in accordance with the Business Source License, use of this +// software will be governed by the Apache License, Version 2.0, included in +// the file licenses/APL2.txt. + +package rosmar + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestQueuePullReportsClose verifies that pull tells a closed queue apart from a nil value. A feed relies +// on this to tell a terminated feed from one that pulled its end-of-feed marker. +func TestQueuePullReportsClose(t *testing.T) { + t.Run("nil value", func(t *testing.T) { + var q queue[*int] + q.init() + q.push(nil) + v, ok := q.pull() + assert.Nil(t, v) + assert.True(t, ok) + }) + + t.Run("closed with values queued", func(t *testing.T) { + var q queue[*int] + q.init() + q.push(new(int)) + q.close() + v, ok := q.pull() + assert.Nil(t, v) + assert.False(t, ok, "close drops queued values") + }) + + t.Run("closed from another goroutine", func(t *testing.T) { + var q queue[*int] + q.init() + pulled := make(chan bool) + go func() { + _, ok := q.pull() + pulled <- ok + }() + q.close() + select { + case ok := <-pulled: + assert.False(t, ok) + case <-time.After(5 * time.Second): + require.FailNow(t, "pull did not return after close") + } + }) +} From 8d7a9115353b463d309358d9d7f97aac5cd0a486 Mon Sep 17 00:00:00 2001 From: Ritesh Kumar Date: Mon, 28 Sep 2026 13:34:04 +0530 Subject: [PATCH 6/6] Run one feed per collection and stop a terminated feed at its next event A feed that names a collection twice now runs one feed on it. Two feeds on the same collection delivered each document twice, and the second feed's done channel replaced the first one's. A feed now checks its Terminator before each event. Before, it stopped only when a separate goroutine closed the queue. If that goroutine ran late, a terminated dump reached its end and counted as completed. The multi-collection test also waits for collection1's backfill to end, because the feed count alone can be 1 before that feed starts. Co-Authored-By: Claude Opus 5.5 --- feeds.go | 30 ++++++++++++++++++++++-------- feeds_test.go | 49 +++++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 69 insertions(+), 10 deletions(-) diff --git a/feeds.go b/feeds.go index 17c4af3..87a7469 100644 --- a/feeds.go +++ b/feeds.go @@ -27,14 +27,14 @@ var activeFeedCount int32 // for tests func (bucket *Bucket) StartDCPFeed(ctx context.Context, args sgbucket.FeedArguments, callback sgbucket.FeedEventCallbackFunc, dbStats *expvar.Map) error { traceEnter("StartDCPFeed", "bucket=%s, args=%+v", bucket.GetName(), args) // Validate requested collections exist before starting feeds. No scopes means the default - // collection, if it exists. - requestedCollections := make([]*Collection, 0) + // collection, if it exists. A collection named twice gets one feed. + requestedCollections := map[*Collection]struct{}{} if len(args.Scopes) == 0 { collection, err := bucket.getCollection(defaultDataStoreName) if err != nil { return err } - requestedCollections = append(requestedCollections, collection) + requestedCollections[collection] = struct{}{} } for scopeName, collections := range args.Scopes { for _, collectionName := range collections { @@ -42,14 +42,14 @@ func (bucket *Bucket) StartDCPFeed(ctx context.Context, args sgbucket.FeedArgume if err != nil { return fmt.Errorf("couldn't open collection %s:%s for DCP feed: %w", scopeName, collectionName, err) } - requestedCollections = append(requestedCollections, collection) + requestedCollections[collection] = struct{}{} } } doneChan := args.DoneChan - doneChans := map[*Collection]chan struct{}{} + doneChans := make(map[*Collection]chan struct{}, len(requestedCollections)) startedFeeds := make([]*dcpFeed, 0, len(requestedCollections)) - for _, collection := range requestedCollections { + for collection := range requestedCollections { // Not bothering to remove scopes from args for the single collection feeds // here because it's ignored by Collection.startDCPFeed collectionID := collection.GetCollectionID() @@ -76,8 +76,8 @@ func (bucket *Bucket) StartDCPFeed(ctx context.Context, args sgbucket.FeedArgume // coalesce doneChans go func() { - for _, collection := range requestedCollections { - <-doneChans[collection] + for _, done := range doneChans { + <-done } // The collections share one checkpoint document, so a dump saves it once they have all stopped. if args.Dump { @@ -352,6 +352,10 @@ func (feed *dcpFeed) run() { collectionID := feed.collection.GetCollectionID() feedContent := feed.args.FeedContent for { + // Stop before the next event once the Terminator is closed, without waiting for the goroutine above. + if feed.terminated() { + break + } e, ok := feed.events.pull() if !ok { break // terminated @@ -382,6 +386,16 @@ func (feed *dcpFeed) run() { } } +// terminated reports whether the feed's Terminator is closed. A feed with no Terminator is never terminated. +func (feed *dcpFeed) terminated() bool { + select { + case <-feed.args.Terminator: + return true + default: + return false + } +} + // allFeedsCompleted reports whether every feed drained to its end-of-feed marker. func allFeedsCompleted(feeds []*dcpFeed) bool { for _, feed := range feeds { diff --git a/feeds_test.go b/feeds_test.go index 64d1e6d..af30e5d 100644 --- a/feeds_test.go +++ b/feeds_test.go @@ -341,9 +341,12 @@ func TestDumpAcrossCollectionsDeletesCheckpointOnlyWhenAllComplete(t *testing.T) // step collection2 one event at a time, so it is still running when the feed is stopped step := make(chan struct{}) + var c1BackfillEnded atomic.Bool callback := func(event sgbucket.FeedEvent) bool { if event.CollectionID == c2ID && event.Opcode == sgbucket.FeedOpMutation { <-step + } else if event.CollectionID != c2ID && event.Opcode == sgbucket.FeedOpEndBackfill { + c1BackfillEnded.Store(true) } return true } @@ -352,8 +355,11 @@ func TestDumpAcrossCollectionsDeletesCheckpointOnlyWhenAllComplete(t *testing.T) for range 3 { step <- struct{}{} } - // collection1 is not stepped, so wait for its feed to reach the end before stopping collection2 - require.Eventually(t, func() bool { return atomic.LoadInt32(&activeFeedCount) == 1 }, 10*time.Second, time.Millisecond) + // collection1 is not stepped, so wait for its feed to reach the end and stop before stopping collection2. + // The feed count alone can also be 1 before collection1's feed has started. + require.Eventually(t, func() bool { + return c1BackfillEnded.Load() && atomic.LoadInt32(&activeFeedCount) == 1 + }, 10*time.Second, time.Millisecond) close(terminator) for done := false; !done; { select { @@ -1339,6 +1345,45 @@ func TestStartDCPFeedUnknownCollection(t *testing.T) { requireNoRegisteredFeeds(t, bucket) } +// TestStartDCPFeedDuplicateCollection verifies that a bucket-level feed that names a collection twice runs one +// feed on it, so each document is delivered once. +func TestStartDCPFeedDuplicateCollection(t *testing.T) { + ctx := t.Context() + ensureNoLeakedFeeds(t) + bucket := makeTestBucket(t) + + c, err := bucket.NamedDataStore(ctx, dsName("scope1", "collection1")) + require.NoError(t, err) + addToCollection(t, c, "able", 0, "A") + addToCollection(t, c, "baker", 0, "B") + + var mutex sync.Mutex + delivered := map[string]int{} + callback := func(event sgbucket.FeedEvent) bool { + if event.Opcode == sgbucket.FeedOpMutation { + mutex.Lock() + delivered[string(event.Key)]++ + mutex.Unlock() + } + return true + } + doneChan := make(chan struct{}) + args := sgbucket.FeedArguments{ + Backfill: 0, + Dump: true, + Scopes: map[string][]string{"scope1": {"collection1", "collection1"}}, + DoneChan: doneChan, + } + require.NoError(t, bucket.StartDCPFeed(ctx, args, callback, nil)) + <-doneChan + // wait for every feed goroutine, so the deliveries of a second feed are counted too + require.Eventually(t, func() bool { return atomic.LoadInt32(&activeFeedCount) == 0 }, 10*time.Second, time.Millisecond) + + mutex.Lock() + defer mutex.Unlock() + assert.Equal(t, map[string]int{"able": 1, "baker": 1}, delivered) +} + // TestStartDCPFeedMissingDefaultCollection verifies that a bucket-level feed with no scopes fails when the // default collection does not exist, instead of creating it. func TestStartDCPFeedMissingDefaultCollection(t *testing.T) {