GO-7561 concurrent AddLog of the same log returns ErrLogExists instead of WriteConflict

db/db.go

@@ -17,6 +17,7 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/readconcern"
 	"go.mongodb.org/mongo-driver/mongo/readpref"
 	"go.uber.org/zap"
+	"storj.io/drpc/drpcerr"
 
 	consensus "github.com/anyproto/any-sync-consensusnode"
 )
@@ -128,7 +129,33 @@ func (s *service) ensureCollections(ctx context.Context, db *mongo.Database) err
 	return nil
 }
 
+const (
+	// txMaxAttempts bounds how many times a transaction that failed with a transient error is run again
+	txMaxAttempts = 10
+	// txRetryDelay is the wait before the second attempt; it grows linearly with each attempt
+	txRetryDelay = 10 * time.Millisecond
+	// transientTxErrorLabel marks an error after which the whole transaction can be run again,
+	// such as a WriteConflict with a concurrent transaction on the same document
+	transientTxErrorLabel = "TransientTransactionError"
+)
+
+// tx runs f in a transaction. A transaction that fails with a transient error is run again from the start,
+// so f must not depend on state an earlier attempt changed.
 func (s *service) tx(ctx context.Context, f func(txCtx mongo.SessionContext) error) (err error) {
+	for attempt := 1; ; attempt++ {
+		err = s.txOnce(ctx, f)
+		if err == nil || attempt == txMaxAttempts || !isTransientTxError(err) {
+			return err
+		}
+		select {
+		case <-ctx.Done():
+			return err
+		case <-time.After(time.Duration(attempt) * txRetryDelay):
+		}
+	}
+}
+
+func (s *service) txOnce(ctx context.Context, f func(txCtx mongo.SessionContext) error) (err error) {
 	return s.client.UseSessionWithOptions(
 		ctx,
 		options.Session().SetDefaultReadConcern(readconcern.Majority()),
@@ -144,32 +171,49 @@ func (s *service) tx(ctx context.Context, f func(txCtx mongo.SessionContext) err
 		})
 }
 
+// isTransientTxError reports whether err carries the transient transaction label, also when err wraps it
+func isTransientTxError(err error) bool {
+	var labeled mongo.LabeledError
+	return errors.As(err, &labeled) && labeled.HasErrorLabel(transientTxErrorLabel)
+}
+
 func (s *service) AddLog(ctx context.Context, l consensus.Log) (err error) {
-	return s.tx(ctx, func(txCtx mongo.SessionContext) error {
+	err = s.tx(ctx, func(txCtx mongo.SessionContext) error {
+		// the log is stored without payloads; a retried transaction needs them again, so l stays intact
+		stored := consensus.Log{Id: l.Id, Records: make([]consensus.Record, len(l.Records))}
 		for i, record := range l.Records {
 			if err := s.savePayload(txCtx, consensus.NewPayload(l.Id, record.Id, record.Payload)); err != nil {
 				return err
 			}
-			l.Records[i].Payload = nil
+			record.Payload = nil
+			stored.Records[i] = record
 		}
-		_, err := s.logColl.InsertOne(txCtx, l)
-		if err != nil {
-			if mongo.IsDuplicateKeyError(err) {
-				return consensuserr.ErrLogExists
-			} else {
-				return errors.Join(consensuserr.ErrUnexpected, err)
-			}
+		_, err := s.logColl.InsertOne(txCtx, stored)
+		if mongo.IsDuplicateKeyError(err) {
+			return consensuserr.ErrLogExists
 		}
-		return nil
+		return err
 	})
+	return consensusErr("addLog", l.Id, err)
+}
+
+// consensusErr passes a consensus error on and replaces any other, such as a mongo error, with ErrUnexpected.
+// A transaction returns mongo errors as they are, so that tx can see their labels; the client needs the rpc code
+// of a consensus error, which an error joined with a mongo error would not carry.
+func consensusErr(op, logId string, err error) error {
+	if err == nil || drpcerr.Code(err) != 0 {
+		return err
+	}
+	log.Error(op+" error", zap.String("logId", logId), zap.Error(err))
+	return consensuserr.ErrUnexpected
 }
 
 type findLogQuery struct {
 	Id string `bson:"_id"`
 }
 
 func (s *service) DeleteLog(ctx context.Context, logId string) (err error) {
-	return s.tx(ctx, func(txCtx mongo.SessionContext) error {
+	err = s.tx(ctx, func(txCtx mongo.SessionContext) error {
 		res, err := s.logColl.DeleteOne(txCtx, findLogQuery{Id: logId})
 		if err != nil {
 			return err
@@ -184,6 +228,7 @@ func (s *service) DeleteLog(ctx context.Context, logId string) (err error) {
 		}
 		return nil
 	})
+	return consensusErr("deleteLog", logId, err)
 }
 
 type findRecordQuery struct {
@@ -201,23 +246,23 @@ type updateOp struct {
 }
 
 func (s *service) AddRecord(ctx context.Context, logId string, record consensus.Record) error {
-	return s.tx(ctx, func(txCtx mongo.SessionContext) (err error) {
+	err := s.tx(ctx, func(txCtx mongo.SessionContext) (err error) {
 		// save payload in a separate collection to avoid the one doc size limit
 		if err = s.savePayload(txCtx, consensus.NewPayload(logId, record.Id, record.Payload)); err != nil {
 			return err
 		}
 
-		// try to add record to the log
+		// try to add record to the log; the stored record goes without its payload, and record stays intact
+		// for a retried transaction
 		var upd updateOp
-		record.Payload = nil
-		upd.Push.Records.Each = []consensus.Record{record}
+		stored := record
+		stored.Payload = nil
+		upd.Push.Records.Each = []consensus.Record{stored}
 		result, err := s.logColl.UpdateOne(txCtx, findRecordQuery{
 			Id:           logId,
 			LastRecordId: record.PrevId,
 		}, upd)
 		if err != nil {
-			log.Error("addRecord update error", zap.Error(err))
-			err = consensuserr.ErrUnexpected
 			return
 		}
 		if result.ModifiedCount == 0 {
@@ -226,6 +271,7 @@ func (s *service) AddRecord(ctx context.Context, logId string, record consensus.
 		}
 		return
 	})
+	return consensusErr("addRecord", logId, err)
 }
 
 func (s *service) savePayload(ctx context.Context, payload consensus.Payload) (err error) {

db/db_test.go

@@ -2,6 +2,7 @@ package db
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -11,6 +12,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"storj.io/drpc/drpcerr"
 
 	consensus "github.com/anyproto/any-sync-consensusnode"
 )
@@ -331,3 +334,105 @@ func (c *testConfig) Name() string { return "config" }
 func (c *testConfig) GetDB() Config {
 	return Config{Connect: "mongodb://localhost:27017/?w=majority", Database: "consensus_test", LogCollection: "log"}
 }
+
+// TestService_AddLog_ConcurrentSameLog covers several nodes creating the same space log at once:
+// exactly one AddLog creates it, every other one reports ErrLogExists, never a WriteConflict.
+func TestService_AddLog_ConcurrentSameLog(t *testing.T) {
+	fx := newFixture(t, nil)
+	defer fx.Finish(t)
+
+	const rounds, n = 20, 5
+	for r := range rounds {
+		logId := fmt.Sprintf("same-log-%d", r)
+		errs := make(chan error, n)
+		for range n {
+			go func() {
+				errs <- fx.AddLog(ctx, consensus.Log{
+					Id:      logId,
+					Records: []consensus.Record{{Id: logId + "-root", Payload: []byte("root")}},
+				})
+			}()
+		}
+		// collect every result before asserting, so that no AddLog runs during the fixture teardown
+		results := make([]error, 0, n)
+		for range n {
+			results = append(results, <-errs)
+		}
+		var created int
+		for _, err := range results {
+			if err == nil {
+				created++
+				continue
+			}
+			assert.ErrorIs(t, err, consensuserr.ErrLogExists, "round %d", r)
+		}
+		assert.Equal(t, 1, created, "round %d", r)
+	}
+}
+
+func TestService_tx(t *testing.T) {
+	transient := mongo.CommandError{Code: 112, Name: "WriteConflict", Labels: []string{transientTxErrorLabel}}
+
+	// failing returns a transaction body that fails with err the given number of times, then succeeds
+	failing := func(times int, err error) (func(mongo.SessionContext) error, *int) {
+		var calls int
+		return func(mongo.SessionContext) error {
+			calls++
+			if calls <= times {
+				return err
+			}
+			return nil
+		}, &calls
+	}
+
+	t.Run("a transient error is retried", func(t *testing.T) {
+		fx := newFixture(t, nil)
+		defer fx.Finish(t)
+		f, calls := failing(2, transient)
+		require.NoError(t, fx.Service.(*service).tx(ctx, f))
+		assert.Equal(t, 3, *calls)
+	})
+	t.Run("a wrapped transient error is retried", func(t *testing.T) {
+		fx := newFixture(t, nil)
+		defer fx.Finish(t)
+		f, calls := failing(1, fmt.Errorf("insert: %w", transient))
+		require.NoError(t, fx.Service.(*service).tx(ctx, f))
+		assert.Equal(t, 2, *calls)
+	})
+	t.Run("retries stop after txMaxAttempts", func(t *testing.T) {
+		fx := newFixture(t, nil)
+		defer fx.Finish(t)
+		f, calls := failing(txMaxAttempts, transient)
+		err := fx.Service.(*service).tx(ctx, f)
+		assert.True(t, isTransientTxError(err))
+		assert.Equal(t, txMaxAttempts, *calls)
+	})
+	t.Run("another error is not retried", func(t *testing.T) {
+		fx := newFixture(t, nil)
+		defer fx.Finish(t)
+		f, calls := failing(1, consensuserr.ErrConflict)
+		assert.ErrorIs(t, fx.Service.(*service).tx(ctx, f), consensuserr.ErrConflict)
+		assert.Equal(t, 1, *calls)
+	})
+}
+
+func TestService_AddLog_KeepsInput(t *testing.T) {
+	fx := newFixture(t, nil)
+	defer fx.Finish(t)
+	l := consensus.Log{Id: "logOne", Records: []consensus.Record{{Id: "recordOne", Payload: []byte("payload")}}}
+	require.NoError(t, fx.AddLog(ctx, l))
+	// a retried transaction saves the payloads again
+	assert.Equal(t, []byte("payload"), l.Records[0].Payload)
+}
+
+func TestConsensusErr(t *testing.T) {
+	assert.NoError(t, consensusErr("op", "logId", nil))
+	for _, err := range []error{consensuserr.ErrLogExists, consensuserr.ErrConflict, consensuserr.ErrUnexpected} {
+		assert.Equal(t, err, consensusErr("op", "logId", err))
+	}
+	// a mongo error reaches the client as ErrUnexpected, with its rpc code
+	err := consensusErr("op", "logId", errors.Join(errors.New("context"), mongo.CommandError{Code: 112, Name: "WriteConflict"}))
+	assert.Equal(t, consensuserr.ErrUnexpected, err)
+	assert.Equal(t, drpcerr.Code(consensuserr.ErrUnexpected), drpcerr.Code(err))
+	assert.NotZero(t, drpcerr.Code(err))
+}

stream/service.go

@@ -2,12 +2,14 @@ package stream
 
 import (
 	"context"
+	"errors"
 	"sync/atomic"
 	"time"
 
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/anyproto/any-sync/app/ocache"
+	"github.com/anyproto/any-sync/consensus/consensusproto/consensuserr"
 	"github.com/anyproto/any-sync/metric"
 	"github.com/cheggaaa/mb/v3"
 	"go.uber.org/zap"
@@ -84,6 +86,12 @@ func (s *service) NewStream() *Stream {
 // AddStream to object with given logId
 func (s *service) AddStream(ctx context.Context, logId string, stream *Stream) (err error) {
 	obj, err := s.getObject(ctx, logId)
+	if errors.Is(err, consensuserr.ErrLogNotFound) {
+		// A node watches a log right after creating it. The lookup can join a load that another watch started
+		// before the log was committed and share its ErrLogNotFound; a failed load leaves the cache, so a second
+		// lookup reads the db again.
+		obj, err = s.getObject(ctx, logId)
+	}
 	if err != nil {
 		return err
 	}

stream/stream.go

@@ -63,6 +63,10 @@ func (s *Stream) WatchIds(ctx context.Context, logIds []string) {
 	for _, logId := range newIds {
 		if addErr := s.s.AddStream(ctx, logId, s); addErr != nil {
 			log.Info("can't add stream for log", zap.String("logId", logId), zap.Error(addErr))
+			// the stream is not attached to the log, so the id is not watched: a later WatchIds tries again
+			s.mu.Lock()
+			delete(s.logIds, logId)
+			s.mu.Unlock()
 			_ = s.mb.Add(ctx, consensus.Log{
 				Id:  logId,
 				Err: addErr,