Touch CAS fixes (#75)

Author: bbrks

collection+xattrs.go

@@ -448,6 +448,81 @@ func (c *Collection) getRawWithXattrs(key string, xattrKeys []string) (sgbucket.
 	return rawDoc, nil
 }
 
+// TouchXattrWithCas sets a single property within a system xattr, bumping the document's CAS.
+//
+// This mirrors gocb's subdoc UpsertSpec with StoreSemanticsReplace and a CAS guard, which is what Couchbase Server
+// uses to touch a metadata document during config rollback (see XattrBootstrapPersistence.touchConfigRollback in sync_gateway).
+//
+// Returns sgbucket.CasMismatchErr if the supplied cas does not match the current document CAS.
+func (c *Collection) TouchXattrWithCas(_ context.Context, key, xattrKey, property, value string, cas CAS) (casOut CAS, err error) {
+	traceEnter("TouchXattrWithCas", "%q, %q.%q=%q, cas=0x%x", key, xattrKey, property, value, cas)
+	defer func() { traceExit("TouchXattrWithCas", err, "0x%x", casOut) }()
+	err = c.withNewCas(func(txn *sql.Tx, newCas CAS) (*event, error) {
+		var (
+			bodyVal     []byte
+			existingCas CAS
+			rawXattrs   []byte
+			revSeqNo    uint64
+			isJSON      bool
+		)
+		row := txn.QueryRow(`SELECT value, cas, xattrs, revSeqNo, isJSON FROM documents WHERE collection=?1 AND key=?2`, c.id, key)
+		if err := scan(row, &bodyVal, &existingCas, &rawXattrs, &revSeqNo, &isJSON); err != nil {
+			return nil, remapKeyError(err, key)
+		}
+		if existingCas != cas {
+			return nil, sgbucket.CasMismatchErr{Expected: cas, Actual: existingCas}
+		}
+
+		xattrMap := semiParsedXattrs{}
+		if len(rawXattrs) > 0 {
+			if err := json.Unmarshal(rawXattrs, &xattrMap); err != nil {
+				return nil, fmt.Errorf("document %q xattrs are unreadable: %w", key, err)
+			}
+		}
+		// JSON-merge xattrKey.<property> = value into the xattr blob.
+		var xattrBody map[string]json.RawMessage
+		if existing, ok := xattrMap[xattrKey]; ok && len(existing) > 0 {
+			if err := json.Unmarshal(existing, &xattrBody); err != nil {
+				return nil, fmt.Errorf("document %q xattr %q is not a JSON object: %w", key, xattrKey, err)
+			}
+		}
+		if xattrBody == nil {
+			xattrBody = map[string]json.RawMessage{}
+		}
+		encodedValue, err := json.Marshal(value)
+		if err != nil {
+			return nil, err
+		}
+		xattrBody[property] = encodedValue
+		encodedXattr, err := json.Marshal(xattrBody)
+		if err != nil {
+			return nil, err
+		}
+		xattrMap[xattrKey] = encodedXattr
+		newRawXattrs, err := json.Marshal(xattrMap)
+		if err != nil {
+			return nil, err
+		}
+
+		revSeqNo++
+		if _, err := txn.Exec(
+			`UPDATE documents SET cas=?1, xattrs=?2, revSeqNo=?3 WHERE collection=?4 AND key=?5`,
+			newCas, newRawXattrs, revSeqNo, c.id, key); err != nil {
+			return nil, err
+		}
+		casOut = newCas
+		return &event{
+			key:      key,
+			value:    bodyVal,
+			cas:      newCas,
+			isJSON:   isJSON,
+			xattrs:   newRawXattrs,
+			revSeqNo: revSeqNo,
+		}, nil
+	})
+	return
+}
+
 // DeleteWithXattrs a document's body and xattrs simultaneously.
 func (c *Collection) DeleteWithXattrs(ctx context.Context, key string, xattrKeys []string) error {
 	err := c.withNewCas(func(txn *sql.Tx, newCas CAS) (*event, error) {

collection.go

@@ -122,15 +122,33 @@ func (c *Collection) getRaw(q queryable, key string) (val []byte, cas CAS, revSe
 
 func (c *Collection) GetAndTouchRaw(_ context.Context, key string, exp Exp) (val []byte, cas CAS, err error) {
 	traceEnter("GetAndTouchRaw", "%q, %d", key, exp)
-	err = c.withNewCas(func(txn *sql.Tx, newCas CAS) (e *event, err error) {
-		exp = absoluteExpiry(exp)
+	newExp := absoluteExpiry(exp)
+	err = c.bucket.inTransaction(func(txn *sql.Tx) error {
 		var revSeqNo int64
-		val, cas, revSeqNo, err = c.getRaw(txn, key)
-		if err == nil {
-			revSeqNo++
-			_, err = txn.Exec(`UPDATE documents SET exp=?1, revSeqNo=?2 WHERE key=?3`, exp, revSeqNo, key)
+		var existingExp Exp
+		row := txn.QueryRow(
+			`SELECT value, cas, revSeqNo, exp FROM documents WHERE collection=? AND key=?`,
+			c.id, key)
+		if scanErr := scan(row, &val, &cas, &revSeqNo, &existingExp); scanErr != nil {
+			return remapKeyError(scanErr, key)
 		}
-		return
+		if val == nil {
+			return sgbucket.MissingError{Key: key}
+		}
+		if existingExp == newExp {
+			// No-op: CB Server's KV TOUCH/GAT preserves CAS and revSeqNo when the
+			// expiry value is unchanged.
+			return nil
+		}
+		revSeqNo++
+		newCas := CAS(hlc.Now())
+		if _, execErr := txn.Exec(
+			`UPDATE documents SET exp=?1, revSeqNo=?2, cas=?3 WHERE collection=?4 AND key=?5`,
+			newExp, revSeqNo, newCas, c.id, key); execErr != nil {
+			return execErr
+		}
+		cas = newCas
+		return c.setLastCas(txn, newCas)
 	})
 	traceExit("GetAndTouchRaw", err, "cas=0x%x, val %s", cas, val)
 	return

collection_test.go

@@ -921,13 +921,31 @@ func TestRevSeqNo(t *testing.T) {
 	require.NoError(t, col.Set(ctx, docID, 0, nil, []byte(`{"foo": 4`)))
 	assertRevSeqNo(t, col, docID, `"4"`)
 
-	_, _, err := col.GetAndTouchRaw(ctx, docID, 0)
+	// Doc was last written with exp=0. A Touch/GetAndTouchRaw that doesn't change
+	// the expiry is a no-op on CB Server: CAS and revSeqNo are preserved.
+	_, casBeforeNoOpTouch, err := col.GetRaw(ctx, docID)
+	require.NoError(t, err)
+
+	_, casAfterNoOpGAT, err := col.GetAndTouchRaw(ctx, docID, 0)
+	require.NoError(t, err)
+	assertRevSeqNo(t, col, docID, `"4"`)
+	require.Equal(t, casBeforeNoOpTouch, casAfterNoOpGAT, "no-op GetAndTouchRaw should not bump CAS")
+
+	casAfterNoOpTouch, err := col.Touch(ctx, docID, 0)
+	require.NoError(t, err)
+	assertRevSeqNo(t, col, docID, `"4"`)
+	require.Equal(t, casBeforeNoOpTouch, casAfterNoOpTouch, "no-op Touch should not bump CAS")
+
+	// Changing the expiry is a metadata mutation: CB Server bumps both CAS and revSeqNo.
+	casAfterChangingTouch, err := col.Touch(ctx, docID, 60)
 	require.NoError(t, err)
 	assertRevSeqNo(t, col, docID, `"5"`)
+	require.NotEqual(t, casBeforeNoOpTouch, casAfterChangingTouch, "Touch should bump CAS when exp changes")
 
-	_, err = col.Touch(ctx, docID, 0)
+	_, casAfterChangingGAT, err := col.GetAndTouchRaw(ctx, docID, 120)
 	require.NoError(t, err)
 	assertRevSeqNo(t, col, docID, `"6"`)
+	require.NotEqual(t, casAfterChangingTouch, casAfterChangingGAT, "GetAndTouchRaw should bump CAS when exp changes")
 
 	writeWithXattrsDocID := "writeWithXattrs"
 	_, err = col.WriteWithXattrs(ctx, writeWithXattrsDocID, 0, 0, []byte(`{"foo": 1}`), nil, nil, nil)

xattrs_test.go

@@ -18,6 +18,66 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestTouchXattrWithCas(t *testing.T) {
+	ctx := t.Context()
+	ensureNoLeakedFeeds(t)
+	coll := makeTestBucket(t).DefaultDataStore(ctx).(*Collection)
+
+	const docID = "doc1"
+	bodyBytes := []byte(`{"hello":"world"}`)
+	xattrsInput := map[string][]byte{
+		syncXattrName: []byte(`{"existing":"value"}`),
+	}
+	originalCas, err := coll.WriteWithXattrs(ctx, docID, 0, 0, bodyBytes, xattrsInput, nil, nil)
+	require.NoError(t, err)
+
+	// Stale CAS should fail.
+	_, err = coll.TouchXattrWithCas(ctx, docID, syncXattrName, "name", "db1", originalCas+1)
+	require.ErrorAs(t, err, &sgbucket.CasMismatchErr{})
+
+	// Correct CAS succeeds and bumps CAS.
+	newCas, err := coll.TouchXattrWithCas(ctx, docID, syncXattrName, "name", "db1", originalCas)
+	require.NoError(t, err)
+	require.NotEqual(t, originalCas, newCas)
+
+	// Verify the xattr property is now visible and the existing properties were preserved.
+	gotBody, gotXattrs, getCas, err := coll.GetWithXattrs(ctx, docID, []string{syncXattrName})
+	require.NoError(t, err)
+	require.Equal(t, newCas, getCas)
+	require.Equal(t, bodyBytes, gotBody)
+	var xattr map[string]string
+	require.NoError(t, json.Unmarshal(gotXattrs[syncXattrName], &xattr))
+	require.Equal(t, "db1", xattr["name"])
+	require.Equal(t, "value", xattr["existing"])
+
+	// The old CAS is now stale.
+	_, err = coll.TouchXattrWithCas(ctx, docID, syncXattrName, "name", "db2", originalCas)
+	require.ErrorAs(t, err, &sgbucket.CasMismatchErr{})
+}
+
+// TestTouchXattrWithCasCreatesXattr verifies that TouchXattrWithCas can populate an xattr on a doc
+// that does not yet have one.
+func TestTouchXattrWithCasCreatesXattr(t *testing.T) {
+	ctx := t.Context()
+	ensureNoLeakedFeeds(t)
+	coll := makeTestBucket(t).DefaultDataStore(ctx).(*Collection)
+
+	const docID = "doc1"
+	require.NoError(t, coll.SetRaw(ctx, docID, 0, nil, []byte(`{"a":1}`)))
+	_, cas, err := coll.GetRaw(ctx, docID)
+	require.NoError(t, err)
+
+	newCas, err := coll.TouchXattrWithCas(ctx, docID, syncXattrName, "name", "db1", cas)
+	require.NoError(t, err)
+	require.NotEqual(t, cas, newCas)
+
+	_, gotXattrs, _, err := coll.GetWithXattrs(ctx, docID, []string{syncXattrName})
+	require.NoError(t, err)
+	var xattr map[string]string
+	require.NoError(t, json.Unmarshal(gotXattrs[syncXattrName], &xattr))
+	require.Equal(t, "db1", xattr["name"])
+}
+
 func TestSetXattrs(t *testing.T) {
 	ctx := t.Context()
 	ensureNoLeakedFeeds(t)