merge CBG-4411

Author: RIT3shSapata

base/error.go

@@ -84,6 +84,7 @@ var (
 	// ErrInvalidJSON is returned when the JSON being unmarshalled cannot be parsed.
 	ErrInvalidJSON = HTTPErrorf(http.StatusBadRequest, "Invalid JSON")
 
+	ErrSyncFnDryRun = &sgError{"Error returned from Sync Function:"}
 	ErrImportDryRun = &sgError{"Error occured during import dry run: "}
 )
 

db/blip.go

@@ -12,6 +12,7 @@ package db
 
 import (
 	"context"
+	"fmt"
 	"regexp"
 	"strings"
 
@@ -84,11 +85,13 @@ func defaultBlipLogger(ctx context.Context) blip.LogFn {
 }
 
 // blipRevMessageProperties returns a set of BLIP message properties for the given parameters.
-func blipRevMessageProperties(revisionHistory []string, deleted bool, seq SequenceID, replacedRevID string, revTreeProperty []string) blip.Properties {
+func blipRevMessageProperties(revisionHistory []string, deleted bool, seq SequenceID, replacedRevID string, revTreeProperty []string) (blip.Properties, error) {
 	properties := make(blip.Properties)
 
-	// TODO: Assert? db.SequenceID.MarshalJSON can never error
-	seqJSON, _ := base.JSONMarshal(seq)
+	seqJSON, err := base.JSONMarshal(seq)
+	if err != nil {
+		return nil, fmt.Errorf("could not marshal sequence %v: %v", seq, err)
+	}
 	properties[RevMessageSequence] = string(seqJSON)
 
 	if len(revisionHistory) > 0 {
@@ -107,7 +110,7 @@ func blipRevMessageProperties(revisionHistory []string, deleted bool, seq Sequen
 		properties[RevMessageReplacedRev] = replacedRevID
 	}
 
-	return properties
+	return properties, nil
 }
 
 // Returns true if this attachment is worth trying to compress.

db/blip_handler.go

@@ -937,7 +937,10 @@ func (bsc *BlipSyncContext) sendRevAsDelta(ctx context.Context, sender *blip.Sen
 			revTreeProperty = append(revTreeProperty, toHistory(redactedRev.History, knownRevs, maxHistory)...)
 		}
 
-		properties := blipRevMessageProperties(history, redactedRev.Deleted, seq, "", revTreeProperty)
+		properties, err := blipRevMessageProperties(history, redactedRev.Deleted, seq, "", revTreeProperty)
+		if err != nil {
+			return err
+		}
 		return bsc.sendRevisionWithProperties(ctx, sender, docID, revID, collectionIdx, redactedRev.BodyBytes, nil, properties, seq, nil)
 	}
 

db/blip_sync_context.go

@@ -597,7 +597,10 @@ func (bsc *BlipSyncContext) sendDelta(ctx context.Context, sender *blip.Sender,
 		revTreeProperty = append(revTreeProperty, revDelta.ToRevID)
 		revTreeProperty = append(revTreeProperty, revDelta.RevisionHistory...)
 	}
-	properties := blipRevMessageProperties(history, revDelta.ToDeleted, seq, "", revTreeProperty)
+	properties, err := blipRevMessageProperties(history, revDelta.ToDeleted, seq, "", revTreeProperty)
+	if err != nil {
+		return err
+	}
 	properties[RevMessageDeltaSrc] = deltaSrcRevID
 
 	if bsc.useHLV() {
@@ -629,7 +632,10 @@ func (bsc *BlipSyncContext) sendNoRev(sender *blip.Sender, docID, revID string,
 	if bsc.activeCBMobileSubprotocol <= CBMobileReplicationV2 && bsc.clientType == BLIPClientTypeSGR2 {
 		noRevRq.SetSeq(seq)
 	} else {
-		noRevRq.SetSequence(seq)
+		err := noRevRq.SetSequence(seq)
+		if err != nil {
+			return err
+		}
 	}
 
 	status, reason := base.ErrorAsHTTPStatus(err)
@@ -785,7 +791,10 @@ func (bsc *BlipSyncContext) sendRevision(ctx context.Context, sender *blip.Sende
 		revTreeHistoryProperty = append(revTreeHistoryProperty, toHistory(docRev.History, knownRevs, maxHistory)...)
 	}
 
-	properties := blipRevMessageProperties(history, docRev.Deleted, seq, replacedRevID, revTreeHistoryProperty)
+	properties, err := blipRevMessageProperties(history, docRev.Deleted, seq, replacedRevID, revTreeHistoryProperty)
+	if err != nil {
+		return err
+	}
 	if base.LogDebugEnabled(ctx, base.KeySync) {
 		replacedRevMsg := ""
 		if replacedRevID != "" {

db/blip_sync_messages.go

@@ -508,8 +508,13 @@ func (nrm *noRevMessage) SetSeq(seq SequenceID) {
 	nrm.Properties[NorevMessageSeq] = seq.String()
 }
 
-func (nrm *noRevMessage) SetSequence(sequence SequenceID) {
-	nrm.Properties[NorevMessageSequence] = sequence.String()
+func (nrm *noRevMessage) SetSequence(sequence SequenceID) error {
+	s, err := base.JSONMarshal(sequence)
+	if err != nil {
+		return err
+	}
+	nrm.Properties[NorevMessageSequence] = string(s)
+	return nil
 }
 
 func (nrm *noRevMessage) SetReason(reason string) {

db/blip_sync_messages_test.go

@@ -0,0 +1,82 @@
+// Copyright 2025-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 db
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/couchbase/sync_gateway/base"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBlipSequenceProperty(t *testing.T) {
+	testCases := []struct {
+		name                 string
+		seq                  SequenceID
+		expectedString       string
+		expectedISGRV2String string
+	}{
+		{
+			name:                 "Simple SequenceID",
+			seq:                  SequenceID{Seq: 1},
+			expectedString:       `1`,
+			expectedISGRV2String: `1`,
+		},
+		{
+			name:                 "compound sequenceID",
+			seq:                  SequenceID{LowSeq: 5, TriggeredBy: 10, Seq: 15},
+			expectedString:       `"5:10:15"`,
+			expectedISGRV2String: `5:10:15`,
+		},
+		{
+			name:                 "TriggeredBy only",
+			seq:                  SequenceID{TriggeredBy: 20, Seq: 25},
+			expectedString:       `"20:25"`,
+			expectedISGRV2String: `20:25`,
+		},
+		{
+			name:                 "LowSeq and Seq",
+			seq:                  SequenceID{LowSeq: 30, Seq: 35},
+			expectedString:       `"30::35"`,
+			expectedISGRV2String: `30::35`,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// test norev sequence property
+			norevMessage := NewNoRevMessage()
+			require.NoError(t, norevMessage.SetSequence(tc.seq))
+			require.Equal(t, tc.expectedString, norevMessage.Properties[NorevMessageSequence])
+			// this string is only used by ISGR when CB_Mobile < 3
+			norevMessage.SetSeq(tc.seq)
+			require.Equal(t, tc.expectedISGRV2String, norevMessage.Properties[NorevMessageSeq])
+
+			// test rev sequence property
+			properties, err := blipRevMessageProperties(nil, false, tc.seq, "", nil)
+			require.NoError(t, err)
+			require.Equal(t, tc.expectedString, properties[RevMessageSequence])
+			changeEntry := &ChangeEntry{
+				Seq: tc.seq,
+				ID:  "docID",
+			}
+			// changes message format
+			ctx := base.TestCtx(t)
+			bh := &blipHandler{
+				loggingCtx:      ctx,
+				BlipSyncContext: &BlipSyncContext{},
+			}
+			changeRow := bh.buildChangesRow(changeEntry, ChangeByVersionType{"rev": "1-abc"})
+			rowString, err := base.JSONMarshal(changeRow)
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf(`[%s,"docID","1-abc"]`, tc.expectedString), string(rowString))
+		})
+	}
+
+}

db/crud.go

@@ -1697,36 +1697,15 @@ func (db *DatabaseCollectionWithUser) PutExistingRevWithBody(ctx context.Context
 // SyncFnDryrun Runs a document through the sync function and returns expiry, channels doc was placed in, access map for users, roles, handler errors and sync fn exceptions.
 // If a docID is a non empty string, the document will be fetched from the bucket, otherwise the body will be used. If both are specified, this function returns an error.
 // The first error return value represents an error that occurs before the sync function is run. The second error return value represents an exception from the sync function.
-func (db *DatabaseCollectionWithUser) SyncFnDryrun(ctx context.Context, body Body, docID, syncFn string) (*channels.ChannelMapperOutput, error, error) {
-	doc := &Document{
-		ID:    docID,
-		_body: body,
-	}
-	oldDoc := doc
-	if docID != "" {
-		if docInBucket, err := db.GetDocument(ctx, docID, DocUnmarshalAll); err == nil {
-			oldDoc = docInBucket
-			if len(doc._body) == 0 {
-				body = oldDoc.Body(ctx)
-				doc._body = body
-				// If no body is given, use doc in bucket as doc with no old doc
-				oldDoc._body = nil
-			}
-			doc._body[BodyRev] = oldDoc.SyncData.GetRevTreeID()
-		} else {
-			return nil, err, nil
-		}
-	} else {
-		oldDoc._body = nil
-	}
+func (db *DatabaseCollectionWithUser) SyncFnDryrun(ctx context.Context, oldDoc *Document, body Body, docID, syncFn string) (*channels.ChannelMapperOutput, error) {
 
 	delete(body, BodyId)
 
 	// Get the revision ID to match, and the new generation number:
 	matchRev, _ := body[BodyRev].(string)
 	generation, _ := ParseRevID(ctx, matchRev)
 	if generation < 0 {
-		return nil, base.HTTPErrorf(http.StatusBadRequest, "Invalid revision ID"), nil
+		return nil, base.HTTPErrorf(http.StatusBadRequest, "Invalid revision ID")
 	}
 	generation++
 
@@ -1742,12 +1721,12 @@ func (db *DatabaseCollectionWithUser) SyncFnDryrun(ctx context.Context, body Bod
 
 	err := validateAPIDocUpdate(body)
 	if err != nil {
-		return nil, err, nil
+		return nil, err
 	}
 	bodyWithoutInternalProps, wasStripped := StripInternalProperties(body)
 	canonicalBytesForRevID, err := base.JSONMarshalCanonical(bodyWithoutInternalProps)
 	if err != nil {
-		return nil, err, nil
+		return nil, err
 	}
 
 	// We needed to keep _deleted around in the body until we generated a rev ID, but now we can ditch it.
@@ -1770,30 +1749,34 @@ func (db *DatabaseCollectionWithUser) SyncFnDryrun(ctx context.Context, body Bod
 	mutableBody, metaMap, _, err := db.prepareSyncFn(oldDoc, newDoc)
 	if err != nil {
 		base.InfofCtx(ctx, base.KeyDiagnostic, "Failed to prepare to run sync function: %v", err)
-		return nil, err, nil
+		return nil, err
 	}
 
 	syncOptions, err := MakeUserCtx(db.user, db.ScopeName, db.Name)
 	if err != nil {
-		return nil, err, nil
+		return nil, err
 	}
 	var output *channels.ChannelMapperOutput
+	var syncErr error
 	if syncFn == "" {
 		output, err = db.ChannelMapper.MapToChannelsAndAccess(ctx, mutableBody, string(oldDoc._rawBody), metaMap, syncOptions)
+		if err != nil {
+			syncErr = fmt.Errorf("%s%s", base.ErrSyncFnDryRun, err)
+		}
 	} else {
 		jsTimeout := time.Duration(base.DefaultJavascriptTimeoutSecs) * time.Second
 		syncRunner, err := channels.NewSyncRunner(ctx, syncFn, jsTimeout)
 		if err != nil {
-			return nil, fmt.Errorf("failed to create sync runner: %v", err), nil
+			return nil, fmt.Errorf("failed to create sync runner: %v", err)
 		}
 		jsOutput, err := syncRunner.Call(ctx, mutableBody, string(oldDoc._rawBody), metaMap, syncOptions)
 		if err != nil {
-			return nil, err, nil
+			syncErr = fmt.Errorf("%s%s", base.ErrSyncFnDryRun, err)
 		}
 		output = jsOutput.(*channels.ChannelMapperOutput)
 	}
 
-	return output, nil, err
+	return output, syncErr
 }
 
 // revTreeConflictCheck checks for conflicts in the rev tree history and returns the parent revid, currentRevIndex

docs/api/paths/diagnostic/keyspace-sync.yaml

@@ -7,16 +7,39 @@
 # the file licenses/APL2.txt.
 parameters:
   - $ref: ../../components/parameters.yaml#/keyspace
-get:
+post:
   summary: Run a doc body through the sync function and return sync data.
   description: |-
-    Run a document body through the sync function and return document sync data.
+    Runs a document body through the sync function and returns document sync
+    data. If no custom sync function is provided in the request body, the
+    default or user-defined sync function for the collection is used.
+    | Document | DocID | Behaviour                                                                                                                                                                            |
+    | -------- | ----- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
+    | Yes      | No    | The document passed will be considered as newDoc and oldDoc will be empty                                                                                                            |
+    | Yes      | Yes   | The document passed in the body will be newDoc and DocID will be read from the bucket/collection and will be passed as the oldDoc. If DocID doesn't exist, then oldDoc will be empty |
+    | No       | No    | Will throw an error                                                                                                                                                                  |
+    | No       | Yes   | The docID will be passed in as the newDoc and oldDoc will be empty. If the document is not found, an error will be returned                                                          |
+
     * Sync Gateway Application Read Only
   requestBody:
     content:
       application/json:
         schema:
-          $ref: ../../components/schemas.yaml#/Document
+          type: object
+          properties:
+            sync_function:
+              description: |-
+                A JavaScript function that defines custom access, channel, and
+                validation logic for documents. This function will be evaluated
+                by the Sync Gateway to determine document routing, access
+                grants, and validation outcomes during synchronization.
+              type: string
+              example: |-
+                function (doc, oldDoc) {
+                  channel(doc.channels);
+                }
+            doc:
+              $ref: ../../components/schemas.yaml#/Document
   responses:
     '200':
       description: Document Processed by sync function successfully

go.mod

@@ -37,7 +37,7 @@ require (
 	golang.org/x/crypto v0.45.0
 	golang.org/x/exp v0.0.0-20250606033433-dcc06ee1d476
 	golang.org/x/net v0.47.0
-	golang.org/x/oauth2 v0.33.0
+	golang.org/x/oauth2 v0.34.0
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1
 )
 

go.sum

@@ -270,8 +270,8 @@ golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwY
 golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY=
 golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
-golang.org/x/oauth2 v0.33.0 h1:4Q+qn+E5z8gPRJfmRy7C2gGG3T4jIprK6aSYgTXGRpo=
-golang.org/x/oauth2 v0.33.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA=
+golang.org/x/oauth2 v0.34.0 h1:hqK/t4AKgbqWkdkcAeI8XLmbK+4m4G5YeQRrmiotGlw=
+golang.org/x/oauth2 v0.34.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=