Skip to content

CBG-5141: Only fetch backup revision for fromRev for delta sync #8041

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bbrks merged 7 commits into from
Jan 30, 2026
Merged

CBG-5141: Only fetch backup revision for fromRev for delta sync #8041

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
99bf635
CBG-5141: Only fetch backup revision for fromRev for delta sync
gregns1 Jan 28, 2026
dbfd5ae
new direction
gregns1 Jan 29, 2026
30b2dd6
removed commented out test case
gregns1 Jan 29, 2026
6454d4e
fix failed loads not getting removed from cache for non default colle…
gregns1 Jan 29, 2026
6dfd21c
fix import issue
gregns1 Jan 29, 2026
de7cd71
query escape
gregns1 Jan 29, 2026
99e5629
fix tests
gregns1 Jan 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion db/attachment_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ func TestGetBackupRevisionWhenCurrentRevisionHasAttachments(t *testing.T) {
// can remove in CBG-4542
db.FlushRevisionCacheForTest()

docRev, err := collection.GetRev(ctx, "doc1", doc1.HLV.GetCurrentVersionString(), false, nil)
docRev, err := collection.revisionCache.GetWithCV(ctx, "doc1", doc1.HLV.ExtractCurrentVersionFromHLV(), false, true)
require.NoError(t, err)

// assert version is fetched and attachments is empty
Expand Down
11 changes: 6 additions & 5 deletions db/crud.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -357,7 +357,7 @@ func (db *DatabaseCollectionWithUser) getRev(ctx context.Context, docid, revOrCV
revision, getErr = db.revisionCache.GetWithRev(ctx, docid, *revID, RevCacheOmitDelta)
} else {
cv = &currentVersion
revision, getErr = db.revisionCache.GetWithCV(ctx, docid, cv, RevCacheOmitDelta)
revision, getErr = db.revisionCache.GetWithCV(ctx, docid, cv, RevCacheOmitDelta, false)
}
} else {
// No rev given, so load active revision
Expand Down Expand Up @@ -432,7 +432,7 @@ func (db *DatabaseCollectionWithUser) documentRevisionForRequest(ctx context.Con

func (db *DatabaseCollectionWithUser) GetCV(ctx context.Context, docid string, cv *Version, revTreeHistory bool) (revision DocumentRevision, err error) {
if cv != nil {
revision, err = db.revisionCache.GetWithCV(ctx, docid, cv, RevCacheOmitDelta)
revision, err = db.revisionCache.GetWithCV(ctx, docid, cv, RevCacheOmitDelta, false)
} else {
revision, err = db.revisionCache.GetActive(ctx, docid)
}
Expand Down Expand Up @@ -462,7 +462,8 @@ func (db *DatabaseCollectionWithUser) GetDelta(ctx context.Context, docID, fromR
if err != nil {
return nil, nil, err
}
initialFromRevision, err = db.revisionCache.GetWithCV(ctx, docID, &fromRevVrs, RevCacheIncludeDelta)
// It is possible delta source will not be resident in the cache and we may want to lookup to the bucket for a backup revision
initialFromRevision, err = db.revisionCache.GetWithCV(ctx, docID, &fromRevVrs, RevCacheIncludeDelta, true)
if err != nil {
return nil, nil, err
}
Expand Down Expand Up @@ -503,7 +504,7 @@ func (db *DatabaseCollectionWithUser) GetDelta(ctx context.Context, docID, fromR
// fromRevisionForDiff is a version of the fromRevision that is guarded by the delta lock that we will use to generate the delta (or check again for a newly cached delta)
var fromRevisionForDiff DocumentRevision
if fromRevIsCV {
fromRevisionForDiff, err = db.revisionCache.GetWithCV(ctx, docID, &fromRevVrs, RevCacheIncludeDelta)
fromRevisionForDiff, err = db.revisionCache.GetWithCV(ctx, docID, &fromRevVrs, RevCacheIncludeDelta, true)
if err != nil {
return nil, nil, err
}
Expand Down Expand Up @@ -537,7 +538,7 @@ func (db *DatabaseCollectionWithUser) GetDelta(ctx context.Context, docID, fromR
if err != nil {
return nil, nil, err
}
toRevision, err = db.revisionCache.GetWithCV(ctx, docID, &cv, RevCacheIncludeDelta)
toRevision, err = db.revisionCache.GetWithCV(ctx, docID, &cv, RevCacheIncludeDelta, false)
if err != nil {
return nil, nil, err
}
Expand Down
5 changes: 3 additions & 2 deletions db/database_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1090,14 +1090,14 @@ func TestFetchCurrentRevAfterFetchBackupRevByCV(t *testing.T) {
"k1": "v2",
BodyRev: rev1ID,
}
rev2ID, _, err := collection.Put(ctx, "doc1", rev2Body)
rev2ID, doc2, err := collection.Put(ctx, "doc1", rev2Body)
require.NoError(t, err, "Error creating doc")

// Flush the revision cache, this can be removed pending CBG-4542
db.FlushRevisionCacheForTest()

// fetch backup rev by cv and ensure we have no revID populated (no way to get revID from backup rev in CV)
docRev, err := collection.GetRev(ctx, "doc1", doc.CV(), true, nil)
docRev, err := collection.revisionCache.GetWithCV(ctx, "doc1", doc.HLV.ExtractCurrentVersionFromHLV(), false, true)
require.NoError(t, err, "Error fetching backup revision CV")
assert.Equal(t, "", docRev.RevID)
assert.Equal(t, `{"k1":"v1"}`, string(docRev.BodyBytes))
Expand All @@ -1107,6 +1107,7 @@ func TestFetchCurrentRevAfterFetchBackupRevByCV(t *testing.T) {
require.NoError(t, err, "error fetching current revision")
assert.Equal(t, rev2ID, docRev.RevID)
assert.Equal(t, `{"k1":"v2"}`, string(docRev.BodyBytes))
assert.Equal(t, doc2.HLV.GetCurrentVersionString(), docRev.CV.String())
}

func TestFetchCurrentRevAfterFetchBackupRevByRevID(t *testing.T) {
Expand Down
6 changes: 3 additions & 3 deletions db/hybrid_logical_vector_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -335,7 +335,7 @@ func TestHLVImport(t *testing.T) {
name: "HLV write from without mou",
preFunc: func(t *testing.T, collection *DatabaseCollectionWithUser, docID string) {
hlvHelper := NewHLVAgent(t, collection.dataStore, otherSource, "_vv")
_ = hlvHelper.InsertWithHLV(ctx, docID)
_ = hlvHelper.InsertWithHLV(ctx, docID, nil)
},
expectedMou: func(output *outputData) *MetadataOnlyUpdate {
return &MetadataOnlyUpdate{
Expand All @@ -356,7 +356,7 @@ func TestHLVImport(t *testing.T) {
name: "XDCR stamped with _mou",
preFunc: func(t *testing.T, collection *DatabaseCollectionWithUser, docID string) {
hlvHelper := NewHLVAgent(t, collection.dataStore, otherSource, "_vv")
cas := hlvHelper.InsertWithHLV(ctx, docID)
cas := hlvHelper.InsertWithHLV(ctx, docID, nil)

_, xattrs, _, err := collection.dataStore.GetWithXattrs(ctx, docID, []string{base.VirtualXattrRevSeqNo})
require.NoError(t, err)
Expand Down Expand Up @@ -387,7 +387,7 @@ func TestHLVImport(t *testing.T) {
name: "invalid _mou, but valid hlv",
preFunc: func(t *testing.T, collection *DatabaseCollectionWithUser, docID string) {
hlvHelper := NewHLVAgent(t, collection.dataStore, otherSource, "_vv")
cas := hlvHelper.InsertWithHLV(ctx, docID)
cas := hlvHelper.InsertWithHLV(ctx, docID, nil)

_, xattrs, _, err := collection.dataStore.GetWithXattrs(ctx, docID, []string{base.VirtualXattrRevSeqNo})
require.NoError(t, err)
Expand Down
5 changes: 2 additions & 3 deletions db/revision_cache_bypass.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ func (rc *BypassRevisionCache) GetWithRev(ctx context.Context, docID, revID stri
}

// GetWithCV fetches the Current Version for the given docID and CV immediately from the bucket.
func (rc *BypassRevisionCache) GetWithCV(ctx context.Context, docID string, cv *Version, collectionID uint32, includeDelta bool) (docRev DocumentRevision, err error) {
func (rc *BypassRevisionCache) GetWithCV(ctx context.Context, docID string, cv *Version, collectionID uint32, includeDelta bool, loadBackup bool) (docRev DocumentRevision, err error) {

docRev = DocumentRevision{
CV: cv,
Expand All @@ -73,12 +73,11 @@ func (rc *BypassRevisionCache) GetWithCV(ctx context.Context, docID string, cv *
}

var hlv *HybridLogicalVector
docRev.BodyBytes, docRev.History, docRev.Channels, docRev.Removed, docRev.Attachments, docRev.Deleted, docRev.Expiry, docRev.RevID, hlv, err = revCacheLoaderForDocumentCV(ctx, rc.backingStores[collectionID], doc, *cv)
docRev.BodyBytes, docRev.History, docRev.Channels, docRev.Removed, docRev.Attachments, docRev.Deleted, docRev.Expiry, docRev.RevID, hlv, err = revCacheLoaderForDocumentCV(ctx, rc.backingStores[collectionID], doc, *cv, loadBackup)
if err != nil {
return DocumentRevision{}, err
}
if hlv != nil {
docRev.CV = hlv.ExtractCurrentVersionFromHLV()
docRev.HlvHistory = hlv.ToHistoryForHLV()
}

Expand Down
26 changes: 15 additions & 11 deletions db/revision_cache_interface.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,9 @@ type RevisionCache interface {
GetWithRev(ctx context.Context, docID, revID string, collectionID uint32, includeDelta bool) (DocumentRevision, error)

// GetWithCV returns the given revision by CV, and stores if not already cached.
// When includeBody=true, the returned DocumentRevision will include a mutable shallow copy of the marshaled body.
// When includeDelta=true, the returned DocumentRevision will include delta - requires additional locking during retrieval.
GetWithCV(ctx context.Context, docID string, cv *Version, collectionID uint32, includeDelta bool) (DocumentRevision, error)
// When loadBackup=true, will load from backup revisions if requested version is not active document
GetWithCV(ctx context.Context, docID string, cv *Version, collectionID uint32, includeDelta bool, loadBackup bool) (DocumentRevision, error)

// GetActive returns the current revision for the given doc ID, and stores if not already cached.
GetActive(ctx context.Context, docID string, collectionID uint32) (docRev DocumentRevision, err error)
Expand Down Expand Up @@ -127,7 +127,7 @@ func DefaultRevisionCacheOptions() *RevisionCacheOptions {
type RevisionCacheBackingStore interface {
GetDocument(ctx context.Context, docid string, unmarshalLevel DocumentUnmarshalLevel) (doc *Document, err error)
getRevision(ctx context.Context, doc *Document, revid string) ([]byte, AttachmentsMeta, base.Set, error)
getCurrentVersion(ctx context.Context, doc *Document, cv Version) ([]byte, AttachmentsMeta, base.Set, bool, error)
getCurrentVersion(ctx context.Context, doc *Document, cv Version, loadBackup bool) ([]byte, AttachmentsMeta, base.Set, bool, error)
}

// collectionRevisionCache is a view of a revision cache for a collection.
Expand All @@ -150,8 +150,8 @@ func (c *collectionRevisionCache) GetWithRev(ctx context.Context, docID, revID s
}

// Get is for per collection access to Get method
func (c *collectionRevisionCache) GetWithCV(ctx context.Context, docID string, cv *Version, includeDelta bool) (DocumentRevision, error) {
return (*c.revCache).GetWithCV(ctx, docID, cv, c.collectionID, includeDelta)
func (c *collectionRevisionCache) GetWithCV(ctx context.Context, docID string, cv *Version, includeDelta bool, loadBackup bool) (DocumentRevision, error) {
return (*c.revCache).GetWithCV(ctx, docID, cv, c.collectionID, includeDelta, loadBackup)
}

// GetActive is for per collection access to GetActive method
Expand Down Expand Up @@ -421,7 +421,7 @@ func revCacheLoader(ctx context.Context, backingStore RevisionCacheBackingStore,

// revCacheLoaderForCv will load a document from the bucket using the CV, compare the fetched doc and the CV specified in the function,
// and will still return revid for purpose of populating the Rev ID lookup map on the cache
func revCacheLoaderForCv(ctx context.Context, backingStore RevisionCacheBackingStore, id IDandCV) (bodyBytes []byte, history Revisions, channels base.Set, removed bool, attachments AttachmentsMeta, deleted bool, expiry *time.Time, revid string, hlv *HybridLogicalVector, err error) {
func revCacheLoaderForCv(ctx context.Context, backingStore RevisionCacheBackingStore, id IDandCV, loadBackup bool) (bodyBytes []byte, history Revisions, channels base.Set, removed bool, attachments AttachmentsMeta, deleted bool, expiry *time.Time, revid string, hlv *HybridLogicalVector, err error) {
cv := Version{
Value: id.Version,
SourceID: id.Source,
Expand All @@ -431,7 +431,7 @@ func revCacheLoaderForCv(ctx context.Context, backingStore RevisionCacheBackingS
return bodyBytes, history, channels, removed, attachments, deleted, expiry, revid, hlv, err
}

return revCacheLoaderForDocumentCV(ctx, backingStore, doc, cv)
return revCacheLoaderForDocumentCV(ctx, backingStore, doc, cv, loadBackup)
}

// Common revCacheLoader functionality used either during a cache miss (from revCacheLoader), or directly when retrieving current rev from cache
Expand Down Expand Up @@ -471,8 +471,8 @@ func revCacheLoaderForDocument(ctx context.Context, backingStore RevisionCacheBa

// revCacheLoaderForDocumentCV used either during cache miss (from revCacheLoaderForCv), or used directly when getting current active CV from cache
// nolint:staticcheck
func revCacheLoaderForDocumentCV(ctx context.Context, backingStore RevisionCacheBackingStore, doc *Document, cv Version) (bodyBytes []byte, history Revisions, channels base.Set, removed bool, attachments AttachmentsMeta, deleted bool, expiry *time.Time, revid string, hlv *HybridLogicalVector, err error) {
if bodyBytes, attachments, channels, deleted, err = backingStore.getCurrentVersion(ctx, doc, cv); err != nil {
func revCacheLoaderForDocumentCV(ctx context.Context, backingStore RevisionCacheBackingStore, doc *Document, cv Version, loadBackup bool) (bodyBytes []byte, history Revisions, channels base.Set, removed bool, attachments AttachmentsMeta, deleted bool, expiry *time.Time, revid string, hlv *HybridLogicalVector, err error) {
if bodyBytes, attachments, channels, deleted, err = backingStore.getCurrentVersion(ctx, doc, cv, loadBackup); err != nil {
return nil, nil, nil, false, nil, false, nil, "", nil, err
}

Expand All @@ -481,9 +481,9 @@ func revCacheLoaderForDocumentCV(ctx context.Context, backingStore RevisionCache
if doc.HLV.ExtractCurrentVersionFromHLV().Equal(cv) {
revid = doc.GetRevTreeID()
deleted = doc.Deleted
hlv = doc.HLV
}

hlv = doc.HLV
validatedHistory, getHistoryErr := doc.History.getHistory(revid)
if getHistoryErr != nil {
return bodyBytes, history, channels, removed, attachments, deleted, doc.Expiry, revid, hlv, err
Expand All @@ -493,8 +493,12 @@ func revCacheLoaderForDocumentCV(ctx context.Context, backingStore RevisionCache
return bodyBytes, history, channels, removed, attachments, deleted, doc.Expiry, revid, hlv, err
}

func (c *DatabaseCollection) getCurrentVersion(ctx context.Context, doc *Document, cv Version) (bodyBytes []byte, attachments AttachmentsMeta, channels base.Set, deleted bool, err error) {
func (c *DatabaseCollection) getCurrentVersion(ctx context.Context, doc *Document, cv Version, loadBackup bool) (bodyBytes []byte, attachments AttachmentsMeta, channels base.Set, deleted bool, err error) {
if err = doc.HasCurrentVersion(ctx, cv); err != nil {
if !loadBackup {
// do not attempt to fetch backup revision by CV unless specified
return nil, nil, nil, false, ErrMissing
}
bodyBytes, channels, deleted, err = c.getOldRevisionJSON(ctx, doc.ID, base.Crc32cHashString([]byte(cv.String())))
if err != nil || bodyBytes == nil {
return nil, nil, nil, false, err
Expand Down
Loading
Loading