GO-7561 nodespace: retry creating the space's consensus log

nodespace/space.go

@@ -69,13 +69,10 @@ func (s *nodeSpace) Init(ctx context.Context) (err error) {
 		return
 	}
 	// TODO: call a coordinator?
-	err = s.consClient.AddLog(ctx, s.Id(), &consensusproto.RawRecordWithId{
+	s.addLog(ctx, &consensusproto.RawRecordWithId{
 		Payload: s.Acl().Root().Payload,
 		Id:      s.Acl().Id(),
 	})
-	if err != nil && rpcerr.Unwrap(err) != consensuserr.ErrLogExists {
-		log.Warn("failed to add consensus record", zap.Error(err))
-	}
 	if err = s.consClient.Watch(s.Id(), s); err != nil {
 		_ = s.Space.Close()
 		return
@@ -100,3 +97,52 @@ func (s *nodeSpace) Close() (err error) {
 	}
 	return s.Space.Close()
 }
+
+const (
+	// addLogAttempts bounds the attempts to create the space's consensus log
+	addLogAttempts = 3
+	// addLogRetryDelay is the wait before the second attempt; it grows linearly with each attempt
+	addLogRetryDelay = 100 * time.Millisecond
+)
+
+// addLog creates the space's consensus log, or finds it created. Every node that stores the space creates it,
+// usually at the same time, and a watch on the log fails until it exists, so a failed attempt is retried.
+// A failure is only logged, so that the space still loads.
+func (s *nodeSpace) addLog(ctx context.Context, root *consensusproto.RawRecordWithId) {
+	var err error
+	for attempt := 1; attempt <= addLogAttempts; attempt++ {
+		if attempt > 1 && !sleepCtx(ctx, time.Duration(attempt-1)*addLogRetryDelay) {
+			break
+		}
+		err = s.consClient.AddLog(ctx, s.Id(), root)
+		switch rpcerr.Unwrap(err) {
+		case nil, consensuserr.ErrLogExists:
+			return
+		}
+		if isPermanentAddLogErr(err) {
+			break
+		}
+	}
+	s.log.Warn("failed to add consensus record", zap.Error(err))
+}
+
+// isPermanentAddLogErr reports whether another AddLog attempt would fail the same way
+func isPermanentAddLogErr(err error) bool {
+	switch rpcerr.Unwrap(err) {
+	case consensuserr.ErrForbidden, consensuserr.ErrInvalidPayload:
+		return true
+	}
+	return false
+}
+
+// sleepCtx waits for d and reports false when ctx is done first
+func sleepCtx(ctx context.Context, d time.Duration) bool {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return false
+	case <-timer.C:
+		return true
+	}
+}

nodespace/space_test.go

@@ -0,0 +1,98 @@
+package nodespace
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/anyproto/any-sync/commonspace/mock_commonspace"
+	"github.com/anyproto/any-sync/commonspace/object/acl/syncacl/mock_syncacl"
+	"github.com/anyproto/any-sync/consensus/consensusclient/mock_consensusclient"
+	"github.com/anyproto/any-sync/consensus/consensusproto"
+	"github.com/anyproto/any-sync/consensus/consensusproto/consensuserr"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+)
+
+const testSpaceId = "spaceId"
+
+var testRoot = &consensusproto.RawRecordWithId{Id: "aclId", Payload: []byte("root")}
+
+// newTestSpace returns a node space whose space and acl are mocks with testSpaceId and testRoot
+func newTestSpace(t *testing.T) (*nodeSpace, *mock_commonspace.MockSpace, *mock_consensusclient.MockService) {
+	ctrl := gomock.NewController(t)
+	sp := mock_commonspace.NewMockSpace(ctrl)
+	acl := mock_syncacl.NewMockSyncAcl(ctrl)
+	cons := mock_consensusclient.NewMockService(ctrl)
+	sp.EXPECT().Id().Return(testSpaceId).AnyTimes()
+	sp.EXPECT().Acl().Return(acl).AnyTimes()
+	acl.EXPECT().Root().Return(testRoot).AnyTimes()
+	acl.EXPECT().Id().Return(testRoot.Id).AnyTimes()
+	s, err := newNodeSpace(sp, cons, nil, nil)
+	require.NoError(t, err)
+	return s, sp, cons
+}
+
+func TestNodeSpace_addLog(t *testing.T) {
+	t.Run("a created log needs one attempt", func(t *testing.T) {
+		s, _, cons := newTestSpace(t)
+		cons.EXPECT().AddLog(gomock.Any(), testSpaceId, testRoot).Return(nil)
+		s.addLog(context.Background(), testRoot)
+	})
+	t.Run("an existing log needs one attempt", func(t *testing.T) {
+		s, _, cons := newTestSpace(t)
+		cons.EXPECT().AddLog(gomock.Any(), testSpaceId, testRoot).Return(consensuserr.ErrLogExists)
+		s.addLog(context.Background(), testRoot)
+	})
+	t.Run("a failed attempt is retried until the log exists", func(t *testing.T) {
+		s, _, cons := newTestSpace(t)
+		gomock.InOrder(
+			cons.EXPECT().AddLog(gomock.Any(), testSpaceId, testRoot).Return(consensuserr.ErrUnexpected),
+			cons.EXPECT().AddLog(gomock.Any(), testSpaceId, testRoot).Return(consensuserr.ErrLogExists),
+		)
+		s.addLog(context.Background(), testRoot)
+	})
+	t.Run("attempts stop after addLogAttempts", func(t *testing.T) {
+		s, _, cons := newTestSpace(t)
+		cons.EXPECT().AddLog(gomock.Any(), testSpaceId, testRoot).Return(errors.New("network")).Times(addLogAttempts)
+		s.addLog(context.Background(), testRoot)
+	})
+	t.Run("a permanent error is not retried", func(t *testing.T) {
+		s, _, cons := newTestSpace(t)
+		cons.EXPECT().AddLog(gomock.Any(), testSpaceId, testRoot).Return(consensuserr.ErrForbidden)
+		s.addLog(context.Background(), testRoot)
+	})
+	t.Run("a done context stops the retries", func(t *testing.T) {
+		s, _, cons := newTestSpace(t)
+		ctx, cancel := context.WithCancel(context.Background())
+		cons.EXPECT().AddLog(gomock.Any(), testSpaceId, testRoot).DoAndReturn(
+			func(context.Context, string, *consensusproto.RawRecordWithId) error {
+				cancel()
+				return consensuserr.ErrUnexpected
+			})
+		// one attempt only, which the mock checks
+		s.addLog(ctx, testRoot)
+	})
+}
+
+func TestNodeSpace_Init(t *testing.T) {
+	t.Run("the log is created before the space watches it", func(t *testing.T) {
+		s, sp, cons := newTestSpace(t)
+		gomock.InOrder(
+			sp.EXPECT().Init(gomock.Any()).Return(nil),
+			cons.EXPECT().AddLog(gomock.Any(), testSpaceId, testRoot).Return(consensuserr.ErrUnexpected),
+			cons.EXPECT().AddLog(gomock.Any(), testSpaceId, testRoot).Return(consensuserr.ErrLogExists),
+			cons.EXPECT().Watch(testSpaceId, s).Return(nil),
+		)
+		require.NoError(t, s.Init(context.Background()))
+	})
+	t.Run("the space loads when every attempt fails", func(t *testing.T) {
+		s, sp, cons := newTestSpace(t)
+		gomock.InOrder(
+			sp.EXPECT().Init(gomock.Any()).Return(nil),
+			cons.EXPECT().AddLog(gomock.Any(), testSpaceId, testRoot).Return(consensuserr.ErrUnexpected).Times(addLogAttempts),
+			cons.EXPECT().Watch(testSpaceId, s).Return(nil),
+		)
+		require.NoError(t, s.Init(context.Background()))
+	})
+}