From 763f0e6032e75cb6d28e0ac319401e9afc225ecb Mon Sep 17 00:00:00 2001 From: Roman Khafizianov Date: Mon, 5 Oct 2026 18:16:33 +0200 Subject: [PATCH 1/3] GO-7561 nodespace: retry creating the space's consensus log Every node that stores a space creates its consensus log when it loads the space, usually at the same time as the others, and then watches it. A failed AddLog was only logged, and the watch that followed could miss a log that was not created yet. A failed AddLog is now retried a few times until the log is created or found existing; Forbidden and InvalidPayload are not retried. A failure is still only logged, so the space loads, and the warning now names the space. --- nodespace/space.go | 54 +++++++++++++++++++++++++++++++--- nodespace/space_test.go | 65 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 115 insertions(+), 4 deletions(-) create mode 100644 nodespace/space_test.go diff --git a/nodespace/space.go b/nodespace/space.go index a54b9b56..1b1f679e 100644 --- a/nodespace/space.go +++ b/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, s.Id(), &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, spaceId string, 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, spaceId, 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 + } +} diff --git a/nodespace/space_test.go b/nodespace/space_test.go new file mode 100644 index 00000000..54de6d1d --- /dev/null +++ b/nodespace/space_test.go @@ -0,0 +1,65 @@ +package nodespace + +import ( + "context" + "errors" + "testing" + "time" + + "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/assert" + "go.uber.org/mock/gomock" +) + +func TestNodeSpace_addLog(t *testing.T) { + const spaceId = "spaceId" + root := &consensusproto.RawRecordWithId{Id: "aclId", Payload: []byte("root")} + + newSpace := func(t *testing.T) (*nodeSpace, *mock_consensusclient.MockService) { + cons := mock_consensusclient.NewMockService(gomock.NewController(t)) + return &nodeSpace{consClient: cons, log: log.With()}, cons + } + + t.Run("a created log needs one attempt", func(t *testing.T) { + s, cons := newSpace(t) + cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(nil) + s.addLog(context.Background(), spaceId, root) + }) + t.Run("an existing log needs one attempt", func(t *testing.T) { + s, cons := newSpace(t) + cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrLogExists) + s.addLog(context.Background(), spaceId, root) + }) + t.Run("a failed attempt is retried until the log exists", func(t *testing.T) { + s, cons := newSpace(t) + gomock.InOrder( + cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrUnexpected), + cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrLogExists), + ) + s.addLog(context.Background(), spaceId, root) + }) + t.Run("attempts stop after addLogAttempts", func(t *testing.T) { + s, cons := newSpace(t) + cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(errors.New("network")).Times(addLogAttempts) + s.addLog(context.Background(), spaceId, root) + }) + t.Run("a permanent error is not retried", func(t *testing.T) { + s, cons := newSpace(t) + cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrForbidden) + s.addLog(context.Background(), spaceId, root) + }) + t.Run("a done context stops the retries", func(t *testing.T) { + s, cons := newSpace(t) + ctx, cancel := context.WithCancel(context.Background()) + cons.EXPECT().AddLog(gomock.Any(), spaceId, root).DoAndReturn( + func(context.Context, string, *consensusproto.RawRecordWithId) error { + cancel() + return consensuserr.ErrUnexpected + }) + start := time.Now() + s.addLog(ctx, spaceId, root) + assert.Less(t, time.Since(start), addLogRetryDelay) + }) +} From 1064acd2717786e2d4b54e47ac9340145511fb97 Mon Sep 17 00:00:00 2001 From: Roman Khafizianov Date: Mon, 5 Oct 2026 18:38:54 +0200 Subject: [PATCH 2/3] GO-7561 nodespace: test that Init creates the log before watching it Covers Init itself, so the retry cannot be bypassed there, and drops a wall-clock assertion from the cancellation test; the mock already checks the single attempt. --- nodespace/space_test.go | 47 +++++++++++++++++++++++++++++++++++++---- 1 file changed, 43 insertions(+), 4 deletions(-) diff --git a/nodespace/space_test.go b/nodespace/space_test.go index 54de6d1d..00a39ba3 100644 --- a/nodespace/space_test.go +++ b/nodespace/space_test.go @@ -4,12 +4,13 @@ import ( "context" "errors" "testing" - "time" + "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/assert" + "github.com/stretchr/testify/require" "go.uber.org/mock/gomock" ) @@ -58,8 +59,46 @@ func TestNodeSpace_addLog(t *testing.T) { cancel() return consensuserr.ErrUnexpected }) - start := time.Now() + // one attempt only, which the mock checks s.addLog(ctx, spaceId, root) - assert.Less(t, time.Since(start), addLogRetryDelay) + }) +} + +func TestNodeSpace_Init(t *testing.T) { + const spaceId = "spaceId" + root := &consensusproto.RawRecordWithId{Id: "aclId", Payload: []byte("root")} + + newSpace := func(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(spaceId).AnyTimes() + sp.EXPECT().Acl().Return(acl).AnyTimes() + acl.EXPECT().Root().Return(root).AnyTimes() + acl.EXPECT().Id().Return(root.Id).AnyTimes() + s, err := newNodeSpace(sp, cons, nil, nil) + require.NoError(t, err) + return s, sp, cons + } + + t.Run("the log is created before the space watches it", func(t *testing.T) { + s, sp, cons := newSpace(t) + gomock.InOrder( + sp.EXPECT().Init(gomock.Any()).Return(nil), + cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrUnexpected), + cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrLogExists), + cons.EXPECT().Watch(spaceId, 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 := newSpace(t) + gomock.InOrder( + sp.EXPECT().Init(gomock.Any()).Return(nil), + cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrUnexpected).Times(addLogAttempts), + cons.EXPECT().Watch(spaceId, s).Return(nil), + ) + require.NoError(t, s.Init(context.Background())) }) } From cc7b30a511e4a0cbcbc2bdcee835e10af1b8a89f Mon Sep 17 00:00:00 2001 From: Roman Khafizianov Date: Wed, 7 Oct 2026 14:36:01 +0200 Subject: [PATCH 3/3] GO-7561 nodespace: addLog uses the space's own id The spaceId parameter was always s.Id(); the tests now use a mock space. --- nodespace/space.go | 6 +-- nodespace/space_test.go | 94 +++++++++++++++++++---------------------- 2 files changed, 47 insertions(+), 53 deletions(-) diff --git a/nodespace/space.go b/nodespace/space.go index 1b1f679e..0205a6d2 100644 --- a/nodespace/space.go +++ b/nodespace/space.go @@ -69,7 +69,7 @@ func (s *nodeSpace) Init(ctx context.Context) (err error) { return } // TODO: call a coordinator? - s.addLog(ctx, s.Id(), &consensusproto.RawRecordWithId{ + s.addLog(ctx, &consensusproto.RawRecordWithId{ Payload: s.Acl().Root().Payload, Id: s.Acl().Id(), }) @@ -108,13 +108,13 @@ const ( // 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, spaceId string, root *consensusproto.RawRecordWithId) { +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, spaceId, root) + err = s.consClient.AddLog(ctx, s.Id(), root) switch rpcerr.Unwrap(err) { case nil, consensuserr.ErrLogExists: return diff --git a/nodespace/space_test.go b/nodespace/space_test.go index 00a39ba3..0e5dcfb0 100644 --- a/nodespace/space_test.go +++ b/nodespace/space_test.go @@ -14,90 +14,84 @@ import ( "go.uber.org/mock/gomock" ) -func TestNodeSpace_addLog(t *testing.T) { - const spaceId = "spaceId" - root := &consensusproto.RawRecordWithId{Id: "aclId", Payload: []byte("root")} +const testSpaceId = "spaceId" + +var testRoot = &consensusproto.RawRecordWithId{Id: "aclId", Payload: []byte("root")} - newSpace := func(t *testing.T) (*nodeSpace, *mock_consensusclient.MockService) { - cons := mock_consensusclient.NewMockService(gomock.NewController(t)) - return &nodeSpace{consClient: cons, log: log.With()}, cons - } +// 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 := newSpace(t) - cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(nil) - s.addLog(context.Background(), spaceId, root) + 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 := newSpace(t) - cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrLogExists) - s.addLog(context.Background(), spaceId, root) + 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 := newSpace(t) + s, _, cons := newTestSpace(t) gomock.InOrder( - cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrUnexpected), - cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrLogExists), + cons.EXPECT().AddLog(gomock.Any(), testSpaceId, testRoot).Return(consensuserr.ErrUnexpected), + cons.EXPECT().AddLog(gomock.Any(), testSpaceId, testRoot).Return(consensuserr.ErrLogExists), ) - s.addLog(context.Background(), spaceId, root) + s.addLog(context.Background(), testRoot) }) t.Run("attempts stop after addLogAttempts", func(t *testing.T) { - s, cons := newSpace(t) - cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(errors.New("network")).Times(addLogAttempts) - s.addLog(context.Background(), spaceId, root) + 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 := newSpace(t) - cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrForbidden) - s.addLog(context.Background(), spaceId, root) + 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 := newSpace(t) + s, _, cons := newTestSpace(t) ctx, cancel := context.WithCancel(context.Background()) - cons.EXPECT().AddLog(gomock.Any(), spaceId, root).DoAndReturn( + 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, spaceId, root) + s.addLog(ctx, testRoot) }) } func TestNodeSpace_Init(t *testing.T) { - const spaceId = "spaceId" - root := &consensusproto.RawRecordWithId{Id: "aclId", Payload: []byte("root")} - - newSpace := func(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(spaceId).AnyTimes() - sp.EXPECT().Acl().Return(acl).AnyTimes() - acl.EXPECT().Root().Return(root).AnyTimes() - acl.EXPECT().Id().Return(root.Id).AnyTimes() - s, err := newNodeSpace(sp, cons, nil, nil) - require.NoError(t, err) - return s, sp, cons - } - t.Run("the log is created before the space watches it", func(t *testing.T) { - s, sp, cons := newSpace(t) + s, sp, cons := newTestSpace(t) gomock.InOrder( sp.EXPECT().Init(gomock.Any()).Return(nil), - cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrUnexpected), - cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrLogExists), - cons.EXPECT().Watch(spaceId, s).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 := newSpace(t) + s, sp, cons := newTestSpace(t) gomock.InOrder( sp.EXPECT().Init(gomock.Any()).Return(nil), - cons.EXPECT().AddLog(gomock.Any(), spaceId, root).Return(consensuserr.ErrUnexpected).Times(addLogAttempts), - cons.EXPECT().Watch(spaceId, s).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())) })