CBG-4249: Create an abstract DCPClient to be able to work with rosmar #7879

torcolvin · 2025-11-13T23:06:45Z

Create an abstract DCPClient to be able to work with rosmar

I expect that this can be extended to work as a sharded or non shared DCP client as well, but I think there's interesting work here.

I made this "work" but not for attachment compaction since this uses hierarchical paths in rosmar for xattr subdoc operations https://jira.issues.couchbase.com/browse/CBG-4232

TODO

ability to tag checkpoints with a groupID
ability to remove checkpoint documents https://github.com/couchbaselabs/rosmar/blob/3b9ac157a3cd52c076649d31cbe385e76386787d/feeds.go#L221
change https://github.com/couchbase/sg-bucket/blob/8c9a5acda2c223f50957df2a513a582fe7530bfd/tap.go#L82 so that sgbucket.DoneChan returns an error to avoid running another goroutine

Pre-review checklist

Removed debug logging (fmt.Print, log.Print, ...)
Logging sensitive data? Make sure it's tagged (e.g. base.UD(docID), base.MD(dbName))
Updated relevant information in the API specifications (such as endpoint descriptions, schemas, ...) in docs/api

Dependencies (if applicable)

Link upstream PRs
Update Go module dependencies when merged

Integration Tests

GSI=true,xattrs=true https://jenkins.sgwdev.com/job/SyncGatewayIntegration/0000/

Copilot

Pull Request Overview

This PR introduces an abstract DCPClient interface to support both Couchbase Server (via gocbcore) and Rosmar backends. The key change is creating a unified DCP client abstraction that dispatches to implementation-specific clients based on the underlying bucket type.

Key changes:

Created DCPClient interface with implementations GoCBDCPClient (renamed from DCPClient) and RosmarDCPClient
Unified DCPClientOptions struct replacing separate options for each implementation
Removed test skips for Rosmar/Walrus, enabling DCP-based tests to run against all bucket types

Reviewed Changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
base/abstract_dcp_client.go	New abstraction layer with `DCPClient` interface and factory function `NewDCPClient`
base/rosmar_dcp_client.go	New Rosmar-specific DCP client implementation
base/dcp_client.go	Renamed `DCPClient` to `GoCBDCPClient` and `DCPClientOptions` to `GoCBDCPClientOptions`
base/dcp_client_stream_observer.go	Updated receiver types from `DCPClient` to `GoCBDCPClient`
base/gocb_dcp_feed.go	Refactored to use new client creation pattern
db/background_mgr_resync_dcp.go	Updated to use new abstract client with scope-based collection specification
db/background_mgr_attachment_migration.go	Updated to use new abstract client with scope-based collection specification
db/attachment_compaction.go	Updated to use new abstract client with scope-based collection specification
db/util_testing.go	Updated to use new abstract client with scope-based collection specification
db/background_mgr_resync_dcp_test.go	Removed Walrus test skips
db/background_mgr_attachment_migration_test.go	Removed Walrus test skips
db/attachment_compaction_test.go	Removed Walrus test skips
base/dcp_client_test.go	Removed Walrus test skips and commented out unported tests
tools/cache_perf_tool/dcpDataGeneration.go	Updated type references to `GoCBDCPClient`

base/rosmar_dcp_client.go

base/dcp_client.go

db/attachment_compaction.go

db/attachment_compaction_test.go

base/dcp_sharded.go

Copilot

Pull request overview

Copilot reviewed 33 out of 33 changed files in this pull request and generated 9 comments.

db/database.go

base/dcp_common.go

base/abstract_dcp_client.go

base/rosmar_dcp_client.go

db/attachment_compaction.go

base/dcp_client.go

base/abstract_dcp_client.go

adamcfraser

A few initial comments, haven't done a full review.

adamcfraser · 2025-12-03T05:03:17Z

base/abstract_dcp_client.go

+	InfofCtx(ctx, KeyDCP, "Started DCP Feed %q for bucket %q", feedName, MD(bucketName))
+	go func() {
+		select {
+		case err := <-doneChan:


I thought that for a one-shot feed, a closed doneChan wasn't an error case.

adamcfraser · 2025-12-03T05:09:42Z

base/dcp_client.go

+// Close is used externally to stop the DCP client.
+func (dc *GoCBDCPClient) Close() {
 	dc.close()
-	return dc.getCloseError()


Do we not care about reporting any potential close error, or is that being done elsewhere?

adamcfraser · 2025-12-03T05:10:38Z

base/dcp_client.go

 var ErrVbUUIDMismatch = errors.New("VbUUID mismatch when failOnRollback set")

-type DCPClient struct {
+type GoCBDCPClient struct {


Nitpick, take it or leave it, but I think GocbDCPClient and GocbDCPClientOptions are more readable.

adamcfraser · 2025-12-03T05:14:58Z

db/database.go

 	bucket, err := base.AsGocbV2Bucket(database.Bucket)
 	if err != nil {
-		return err
+		checkpoint := checkpointPrefix + ":" + feedPrefix


Is this Rosmar-specific handling - it seems to require some knowledge of the naming convention for rosmar DCP checkpoints. Seems like an unexpected place for rosmar-specific code.

bbrks

Dropping the few comments I had from yesterday's review

bbrks · 2025-12-02T14:18:34Z

db/util_testing.go

-		FailOnRollback:    false,
-		CollectionIDs:     slices.Collect(maps.Keys(collections)),
-		MetadataStoreType: base.DCPMetadataStoreInMemory,
+		collectionNames.Add(dataStoreName)


Since Add takes a variadic and Add itself is doing a reasonable amount of work to handle the de-dupe, it might make more sense to push this down outside of the loop?

bbrks · 2025-12-02T14:19:53Z

base/collection_common.go

+	for _, d := range ds {
+		if _, ok := c[d.ScopeName()]; !ok {
+			c[d.ScopeName()] = []string{}
+		} else if slices.Contains(c[d.ScopeName()], d.CollectionName()) {


I don't really like the slices.Contains inside the ds iteration - especially thinking about in the future when we're considering increasing the collection limits.

It kind of feels like this would be better suited as a map[string]map[string]struct{} or similar to avoid the slice iterations at insertion time, and handle dedupe.

bbrks · 2025-12-02T14:28:56Z

db/background_mgr_resync_dcp.go


+func getResyncDCPPrefix(resyncID string) string {
+	return fmt.Sprintf(
+		"sg-%v:resync:%v",


We should decide on whether CBG-4286 means we should make this prefix sg: or one that includes a version. I would be inclined to say sg: and then group version numbers in the trailing part of the name.

Copilot AI review requested due to automatic review settings November 13, 2025 23:06

torcolvin marked this pull request as draft November 13, 2025 23:06

Copilot AI reviewed Nov 13, 2025

View reviewed changes

Create an abstract DCP client, support rosmar DCP

e83d779

torcolvin force-pushed the dcp-abstract branch from ec497ab to 8aa5326 Compare December 1, 2025 20:20

github-advanced-security bot found potential problems Dec 1, 2025

View reviewed changes

base/dcp_sharded.go Fixed Show fixed Hide fixed

Unskip tests

dc06fb0

torcolvin force-pushed the dcp-abstract branch from 8aa5326 to dc06fb0 Compare December 2, 2025 02:06

torcolvin requested a review from Copilot December 2, 2025 02:06

torcolvin changed the title ~~Prototype: Create an abstract DCPClient to be able to work with rosmar~~ CBG-4249: Create an abstract DCPClient to be able to work with rosmar Dec 2, 2025

Copilot AI reviewed Dec 2, 2025

View reviewed changes

copilot suggestions

f7dec40

torcolvin mentioned this pull request Dec 2, 2025

CBG-4286: Add sg prefix to dcp feed names #7916

Open

6 tasks

adamcfraser reviewed Dec 3, 2025

View reviewed changes

bbrks reviewed Dec 3, 2025

View reviewed changes

CBG-4249: Create an abstract DCPClient to be able to work with rosmar #7879

Are you sure you want to change the base?

CBG-4249: Create an abstract DCPClient to be able to work with rosmar #7879

Conversation

torcolvin commented Nov 13, 2025

TODO

Pre-review checklist

Dependencies (if applicable)

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

adamcfraser left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbrks left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants