Skip to content

Commit

Permalink
CBG-3797 Attachment handling for HLV push replication (#6702)
Browse files Browse the repository at this point in the history 
HLV clients don't consider revpos, and evaluate whether they need to request an attachment based on the existing set of attachments on the document.

SGW still needs to persist revpos into _attachments to support revtree clients.  For new attachments added by HLV client, revpos is set to the generation of SGW's computed revTreeID for the incoming revision.

Co-authored-by: Gregory Newman-Smith <gregory.newmansmith@couchbase.com>
  • Loading branch information
@adamcfraser @gregns1 @bbrks
2 people authored and bbrks committed Apr 16, 2024
1 parent 8acc2a5 commit 37d3cb5
Show file tree
Hide file tree
Showing 8 changed files with 164 additions and 95 deletions.
64 changes: 36 additions & 28 deletions db/blip_handler.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -808,7 +808,7 @@ func (bh *blipHandler) handleProposeChanges(rq *blip.Message) error {
}
var status ProposedRevStatus
var currentRev string
if bh.activeCBMobileSubprotocol >= CBMobileReplicationV4 {
if bh.useHLV() {
status, currentRev = bh.collection.CheckProposedVersion(bh.loggingCtx, docID, rev, parentRevID)
} else {
status, currentRev = bh.collection.CheckProposedRev(bh.loggingCtx, docID, rev, parentRevID)
Expand Down Expand Up @@ -954,7 +954,7 @@ func (bh *blipHandler) processRev(rq *blip.Message, stats *processRevStats) (err
}
}()

if bh.activeCBMobileSubprotocol >= CBMobileReplicationV4 && bh.conflictResolver != nil {
if bh.useHLV() && bh.conflictResolver != nil {
return base.HTTPErrorf(http.StatusNotImplemented, "conflict resolver handling (ISGR) not yet implemented for v4 protocol")
}

Expand Down Expand Up @@ -1026,17 +1026,18 @@ func (bh *blipHandler) processRev(rq *blip.Message, stats *processRevStats) (err
}

var history []string
historyStr := rq.Properties[RevMessageHistory]
var incomingHLV HybridLogicalVector
// Build history/HLV
if bh.activeCBMobileSubprotocol < CBMobileReplicationV4 {
if !bh.useHLV() {
newDoc.RevID = rev
history = []string{rev}
if historyStr := rq.Properties[RevMessageHistory]; historyStr != "" {
if historyStr != "" {
history = append(history, strings.Split(historyStr, ",")...)
}
} else {
versionVectorStr := rev
if historyStr := rq.Properties[RevMessageHistory]; historyStr != "" {
if historyStr != "" {
versionVectorStr += ";" + historyStr
}
incomingHLV, err = extractHLVFromBlipMessage(versionVectorStr)
Expand Down Expand Up @@ -1066,7 +1067,7 @@ func (bh *blipHandler) processRev(rq *blip.Message, stats *processRevStats) (err
// due to no-conflict write restriction, but we still need to enforce security here to prevent leaking data about previous
// revisions to malicious actors (in the scenario where that user has write but not read access).
var deltaSrcRev DocumentRevision
if bh.activeCBMobileSubprotocol >= CBMobileReplicationV4 {
if bh.useHLV() {
cv := Version{}
cv.SourceID, cv.Value = incomingHLV.GetCurrentVersion()
deltaSrcRev, err = bh.collection.GetCV(bh.loggingCtx, docID, &cv, RevCacheOmitBody)
Expand Down Expand Up @@ -1138,31 +1139,32 @@ func (bh *blipHandler) processRev(rq *blip.Message, stats *processRevStats) (err

var rawBucketDoc *sgbucket.BucketDocument

// Pull out attachments
// Attachment processing
if injectedAttachmentsForDelta || bytes.Contains(bodyBytes, []byte(BodyAttachments)) {
// temporarily error here if V4
if bh.activeCBMobileSubprotocol >= CBMobileReplicationV4 {
return base.HTTPErrorf(http.StatusNotImplemented, "attachment handling not yet supported for v4 protocol")
}

body := newDoc.Body(bh.loggingCtx)

var currentBucketDoc *Document

// Look at attachments with revpos > the last common ancestor's
minRevpos := 1
if len(history) > 0 {
currentDoc, rawDoc, err := bh.collection.GetDocumentWithRaw(bh.loggingCtx, docID, DocUnmarshalSync)
// If we're able to obtain current doc data then we should use the common ancestor generation++ for min revpos
// as we will already have any attachments on the common ancestor so don't need to ask for them.
// Otherwise we'll have to go as far back as we can in the doc history and choose the last entry in there.
if err == nil {
commonAncestor := currentDoc.History.findAncestorFromSet(currentDoc.CurrentRev, history)
minRevpos, _ = ParseRevID(bh.loggingCtx, commonAncestor)
minRevpos++
rawBucketDoc = rawDoc
currentBucketDoc = currentDoc
} else {
minRevpos, _ = ParseRevID(bh.loggingCtx, history[len(history)-1])
minRevpos := 0
if historyStr != "" {
// fetch current bucket doc. Treats error as not found
currentBucketDoc, rawBucketDoc, _ = bh.collection.GetDocumentWithRaw(bh.loggingCtx, docID, DocUnmarshalSync)

// For revtree clients, can use revPos as an optimization. HLV always compares incoming
// attachments with current attachments on the document
if !bh.useHLV() {
// Look at attachments with revpos > the last common ancestor's
// If we're able to obtain current doc data then we should use the common ancestor generation++ for min revpos
// as we will already have any attachments on the common ancestor so don't need to ask for them.
// Otherwise we'll have to go as far back as we can in the doc history and choose the last entry in there.
if currentBucketDoc != nil {
commonAncestor := currentBucketDoc.History.findAncestorFromSet(currentBucketDoc.CurrentRev, history)
minRevpos, _ = ParseRevID(bh.loggingCtx, commonAncestor)
minRevpos++
} else {
minRevpos, _ = ParseRevID(bh.loggingCtx, history[len(history)-1])
}
}
}

Expand All @@ -1180,7 +1182,9 @@ func (bh *blipHandler) processRev(rq *blip.Message, stats *processRevStats) (err
if !ok {
// If we don't have this attachment already, ensure incoming revpos is greater than minRevPos, otherwise
// update to ensure it's fetched and uploaded
bodyAtts[name].(map[string]interface{})["revpos"], _ = ParseRevID(bh.loggingCtx, rev)
if minRevpos > 0 {
bodyAtts[name].(map[string]interface{})["revpos"], _ = ParseRevID(bh.loggingCtx, rev)
}
continue
}

Expand Down Expand Up @@ -1251,7 +1255,7 @@ func (bh *blipHandler) processRev(rq *blip.Message, stats *processRevStats) (err
// If the doc is a tombstone we want to allow conflicts when running SGR2
// bh.conflictResolver != nil represents an active SGR2 and BLIPClientTypeSGR2 represents a passive SGR2
forceAllowConflictingTombstone := newDoc.Deleted && (bh.conflictResolver != nil || bh.clientType == BLIPClientTypeSGR2)
if bh.activeCBMobileSubprotocol >= CBMobileReplicationV4 {
if bh.useHLV() {
_, _, _, err = bh.collection.PutExistingCurrentVersion(bh.loggingCtx, newDoc, incomingHLV, rawBucketDoc)
} else if bh.conflictResolver != nil {
_, _, err = bh.collection.PutExistingRevWithConflictResolution(bh.loggingCtx, newDoc, history, true, bh.conflictResolver, forceAllowConflictingTombstone, rawBucketDoc, ExistingVersionWithUpdateToHLV)
Expand Down Expand Up @@ -1585,3 +1589,7 @@ func allowedAttachmentKey(docID, digest string, activeCBMobileSubprotocol CBMobi
func (bh *blipHandler) logEndpointEntry(profile, endpoint string) {
base.InfofCtx(bh.loggingCtx, base.KeySyncMsg, "#%d: Type:%s %s", bh.serialNumber, profile, endpoint)
}

func (bh *blipHandler) useHLV() bool {
return bh.activeCBMobileSubprotocol >= CBMobileReplicationV4
}
2 changes: 1 addition & 1 deletion db/revision_cache_interface.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -332,7 +332,7 @@ func revCacheLoaderForDocument(ctx context.Context, backingStore RevisionCacheBa
func revCacheLoaderForDocumentCV(ctx context.Context, backingStore RevisionCacheBackingStore, doc *Document, cv Version) (bodyBytes []byte, body Body, history Revisions, channels base.Set, removed bool, attachments AttachmentsMeta, deleted bool, expiry *time.Time, revid string, hlv *HybridLogicalVector, err error) {
if bodyBytes, body, attachments, err = backingStore.getCurrentVersion(ctx, doc); err != nil {
// TODO: CBG-3814 - pending support of channel removal for CV
base.ErrorfCtx(ctx, "pending CBG-3213 support of channel removal for CV: %v", err)
base.ErrorfCtx(ctx, "pending CBG-3814 support of channel removal for CV: %v", err)
}

if err = doc.HasCurrentVersion(ctx, cv); err != nil {
Expand Down
2 changes: 1 addition & 1 deletion db/util_testing.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -682,7 +682,7 @@ func (c *DatabaseCollection) RequireCurrentVersion(t *testing.T, key string, sou
}

// GetDocumentCurrentVersion fetches the document by key and returns the current version
func (c *DatabaseCollection) GetDocumentCurrentVersion(t *testing.T, key string) (source string, version string) {
func (c *DatabaseCollection) GetDocumentCurrentVersion(t testing.TB, key string) (source string, version string) {
ctx := base.TestCtx(t)
doc, err := c.GetDocument(ctx, key, DocUnmarshalSync)
require.NoError(t, err)
Expand Down
23 changes: 9 additions & 14 deletions rest/attachment_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2265,7 +2265,6 @@ func TestUpdateExistingAttachment(t *testing.T) {
}

btcRunner := NewBlipTesterClientRunner(t)
btcRunner.SkipSubtest[VersionVectorSubtestName] = true // attachments not yet replicated in V4 protocol (CBG-3797)
const (
doc1ID = "doc1"
doc2ID = "doc2"
Expand All @@ -2279,8 +2278,8 @@ func TestUpdateExistingAttachment(t *testing.T) {
btc := btcRunner.NewBlipTesterClientOptsWithRT(rt, opts)
defer btc.Close()

doc1Version := rt.PutDoc(doc1ID, `{}`)
doc2Version := rt.PutDoc(doc2ID, `{}`)
doc1Version := btc.PutDoc(doc1ID, `{}`)
doc2Version := btc.PutDoc(doc2ID, `{}`)

require.NoError(t, rt.WaitForPendingChanges())

Expand Down Expand Up @@ -2328,7 +2327,6 @@ func TestPushUnknownAttachmentAsStub(t *testing.T) {
}
const doc1ID = "doc1"
btcRunner := NewBlipTesterClientRunner(t)
btcRunner.SkipSubtest[VersionVectorSubtestName] = true // attachments not yet replicated in V4 protocol (CBG-3797)

btcRunner.Run(func(t *testing.T, SupportedBLIPProtocols []string) {
rt := NewRestTester(t, rtConfig)
Expand All @@ -2338,7 +2336,7 @@ func TestPushUnknownAttachmentAsStub(t *testing.T) {
btc := btcRunner.NewBlipTesterClientOptsWithRT(rt, &opts)
defer btc.Close()
// Add doc1 and doc2
doc1Version := btc.rt.PutDoc(doc1ID, `{}`)
doc1Version := btc.PutDoc(doc1ID, `{}`)

require.NoError(t, btc.rt.WaitForPendingChanges())

Expand Down Expand Up @@ -2378,7 +2376,6 @@ func TestMinRevPosWorkToAvoidUnnecessaryProveAttachment(t *testing.T) {
}

btcRunner := NewBlipTesterClientRunner(t)
btcRunner.SkipSubtest[VersionVectorSubtestName] = true // attachments not yet replicated in V4 protocol (CBG-3797)
const docID = "doc"

btcRunner.Run(func(t *testing.T, SupportedBLIPProtocols []string) {
Expand All @@ -2389,7 +2386,7 @@ func TestMinRevPosWorkToAvoidUnnecessaryProveAttachment(t *testing.T) {
btc := btcRunner.NewBlipTesterClientOptsWithRT(rt, &opts)
defer btc.Close()
// Push an initial rev with attachment data
initialVersion := btc.rt.PutDoc(docID, `{"_attachments": {"hello.txt": {"data": "aGVsbG8gd29ybGQ="}}}`)
initialVersion := btc.PutDoc(docID, `{"_attachments": {"hello.txt": {"data": "aGVsbG8gd29ybGQ="}}}`)
err := btc.rt.WaitForPendingChanges()
assert.NoError(t, err)

Expand All @@ -2401,7 +2398,7 @@ func TestMinRevPosWorkToAvoidUnnecessaryProveAttachment(t *testing.T) {
// Push a revision with a bunch of history simulating doc updated on mobile device
// Note this references revpos 1 and therefore SGW has it - Shouldn't need proveAttachment
proveAttachmentBefore := btc.pushReplication.replicationStats.ProveAttachment.Value()
revid, err := btcRunner.PushRevWithHistory(btc.id, docID, initialVersion.RevTreeID, []byte(`{"_attachments": {"hello.txt": {"revpos":1,"stub":true,"digest":"sha1-Kq5sNclPz7QV2+lfQIuc6R7oRu0="}}}`), 25, 5)
revid, err := btcRunner.PushRevWithHistory(btc.id, docID, initialVersion.GetRev(btc.UseHLV()), []byte(`{"_attachments": {"hello.txt": {"revpos":1,"stub":true,"digest":"sha1-Kq5sNclPz7QV2+lfQIuc6R7oRu0="}}}`), 25, 5)
assert.NoError(t, err)
proveAttachmentAfter := btc.pushReplication.replicationStats.ProveAttachment.Value()
assert.Equal(t, proveAttachmentBefore, proveAttachmentAfter)
Expand All @@ -2420,7 +2417,6 @@ func TestAttachmentWithErroneousRevPos(t *testing.T) {
}

btcRunner := NewBlipTesterClientRunner(t)
btcRunner.SkipSubtest[VersionVectorSubtestName] = true // attachments not yet replicated in V4 protocol (CBG-3797)

btcRunner.Run(func(t *testing.T, SupportedBLIPProtocols []string) {
rt := NewRestTester(t, rtConfig)
Expand All @@ -2431,7 +2427,7 @@ func TestAttachmentWithErroneousRevPos(t *testing.T) {
defer btc.Close()
// Create rev 1 with the hello.txt attachment
const docID = "doc"
version := btc.rt.PutDoc(docID, `{"val": "val", "_attachments": {"hello.txt": {"data": "aGVsbG8gd29ybGQ="}}}`)
version := btc.PutDoc(docID, `{"val": "val", "_attachments": {"hello.txt": {"data": "aGVsbG8gd29ybGQ="}}}`)
err := btc.rt.WaitForPendingChanges()
assert.NoError(t, err)

Expand All @@ -2446,7 +2442,7 @@ func TestAttachmentWithErroneousRevPos(t *testing.T) {
btcRunner.AttachmentsLock(btc.id).Unlock()

// Put doc with an erroneous revpos 1 but with a different digest, referring to the above attachment
_, err = btcRunner.PushRevWithHistory(btc.id, docID, version.RevTreeID, []byte(`{"_attachments": {"hello.txt": {"revpos":1,"stub":true,"length": 19,"digest":"sha1-l+N7VpXGnoxMm8xfvtWPbz2YvDc="}}}`), 1, 0)
_, err = btcRunner.PushRevWithHistory(btc.id, docID, version.GetRev(btc.UseHLV()), []byte(`{"_attachments": {"hello.txt": {"revpos":1,"stub":true,"length": 19,"digest":"sha1-l+N7VpXGnoxMm8xfvtWPbz2YvDc="}}}`), 1, 0)
require.NoError(t, err)

// Ensure message and attachment is pushed up
Expand Down Expand Up @@ -2602,7 +2598,6 @@ func TestCBLRevposHandling(t *testing.T) {
}

btcRunner := NewBlipTesterClientRunner(t)
btcRunner.SkipSubtest[VersionVectorSubtestName] = true // attachments not yet replicated in V4 protocol (CBG-3797)
const (
doc1ID = "doc1"
doc2ID = "doc2"
Expand All @@ -2616,8 +2611,8 @@ func TestCBLRevposHandling(t *testing.T) {
btc := btcRunner.NewBlipTesterClientOptsWithRT(rt, &opts)
defer btc.Close()

doc1Version := btc.rt.PutDoc(doc1ID, `{}`)
doc2Version := btc.rt.PutDoc(doc2ID, `{}`)
doc1Version := btc.PutDoc(doc1ID, `{}`)
doc2Version := btc.PutDoc(doc2ID, `{}`)
require.NoError(t, btc.rt.WaitForPendingChanges())

err := btcRunner.StartOneshotPull(btc.id)
Expand Down
Loading

0 comments on commit 37d3cb5

Please sign in to comment.