consensus: fix race

accountservice/mock_accountservice/helper.go

@@ -3,7 +3,7 @@ package mock_accountservice
 import (
 	"github.com/anyproto/any-sync/accountservice"
 	"github.com/anyproto/any-sync/commonspace/object/accountdata"
-	"github.com/golang/mock/gomock"
+	"go.uber.org/mock/gomock"
 )
 
 func NewAccountServiceWithAccount(ctrl *gomock.Controller, acc *accountdata.AccountKeys) *MockService {

accountservice/mock_accountservice/mock_accountservice.go

@@ -9,7 +9,7 @@ import (
 
 	app "github.com/anyproto/any-sync/app"
 	accountdata "github.com/anyproto/any-sync/commonspace/object/accountdata"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockService is a mock of Service interface.

app/ldiff/mock_ldiff/mock_ldiff.go

@@ -9,7 +9,7 @@ import (
 	reflect "reflect"
 
 	ldiff "github.com/anyproto/any-sync/app/ldiff"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockDiff is a mock of Diff interface.

commonspace/credentialprovider/mock_credentialprovider/mock_credentialprovider.go

@@ -10,7 +10,7 @@ import (
 
 	app "github.com/anyproto/any-sync/app"
 	spacesyncproto "github.com/anyproto/any-sync/commonspace/spacesyncproto"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockCredentialProvider is a mock of CredentialProvider interface.

commonspace/deletionstate/deletionstate_test.go

@@ -3,8 +3,8 @@ package deletionstate
 import (
 	"github.com/anyproto/any-sync/commonspace/spacestorage"
 	"github.com/anyproto/any-sync/commonspace/spacestorage/mock_spacestorage"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"sort"
 	"testing"
 )

commonspace/deletionstate/mock_deletionstate/mock_deletionstate.go

@@ -9,7 +9,7 @@ import (
 
 	app "github.com/anyproto/any-sync/app"
 	deletionstate "github.com/anyproto/any-sync/commonspace/deletionstate"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockObjectDeletionState is a mock of ObjectDeletionState interface.

commonspace/headsync/diffsyncer.go

@@ -26,8 +26,6 @@ type DiffSyncer interface {
 	Close() error
 }
 
-const logPeriodSecs = 200
-
 func newDiffSyncer(hs *headSync) DiffSyncer {
 	return &diffSyncer{
 		diff:               hs.diff,
@@ -37,7 +35,7 @@ func newDiffSyncer(hs *headSync) DiffSyncer {
 		peerManager:        hs.peerManager,
 		clientFactory:      spacesyncproto.ClientFactoryFunc(spacesyncproto.NewDRPCSpaceSyncClient),
 		credentialProvider: hs.credentialProvider,
-		log:                newSyncLogger(hs.log, logPeriodSecs),
+		log:                log,
 		syncStatus:         hs.syncStatus,
 		deletionState:      hs.deletionState,
 	}
@@ -50,7 +48,7 @@ type diffSyncer struct {
 	treeManager        treemanager.TreeManager
 	storage            spacestorage.SpaceStorage
 	clientFactory      spacesyncproto.ClientFactory
-	log                syncLogger
+	log                logger.CtxLogger
 	deletionState      deletionstate.ObjectDeletionState
 	credentialProvider credentialprovider.CredentialProvider
 	syncStatus         syncstatus.StatusUpdater
@@ -102,7 +100,7 @@ func (d *diffSyncer) Sync(ctx context.Context) error {
 			d.log.ErrorCtx(ctx, "can't sync with peer", zap.String("peer", p.Id()), zap.Error(err))
 		}
 	}
-	d.log.DebugCtx(ctx, "diff done", zap.String("spaceId", d.spaceId), zap.Duration("dur", time.Since(st)))
+	d.log.InfoCtx(ctx, "diff done", zap.String("spaceId", d.spaceId), zap.Duration("dur", time.Since(st)))
 	return nil
 }
 
@@ -147,7 +145,12 @@ func (d *diffSyncer) syncWithPeer(ctx context.Context, p peer.Peer) (err error)
 	if err != nil {
 		return err
 	}
-	d.log.logSyncDone(p.Id(), len(newIds), len(changedIds), len(removedIds), totalLen-len(existingIds)-len(missingIds))
+	d.log.Info("sync done:", zap.Int("newIds", len(newIds)),
+		zap.Int("changedIds", len(changedIds)),
+		zap.Int("removedIds", len(removedIds)),
+		zap.Int("already deleted ids", totalLen-len(existingIds)-len(missingIds)),
+		zap.String("peerId", p.Id()),
+	)
 	return
 }
 

commonspace/headsync/diffsyncer_test.go

@@ -14,8 +14,8 @@ import (
 	"github.com/anyproto/any-sync/commonspace/spacesyncproto"
 	"github.com/anyproto/any-sync/consensus/consensusproto"
 	"github.com/anyproto/any-sync/net/peer"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"storj.io/drpc"
 )
 

commonspace/headsync/headsync.go

@@ -6,7 +6,7 @@ import (
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/app/ldiff"
 	"github.com/anyproto/any-sync/app/logger"
-	"github.com/anyproto/any-sync/commonspace/config"
+	config2 "github.com/anyproto/any-sync/commonspace/config"
 	"github.com/anyproto/any-sync/commonspace/credentialprovider"
 	"github.com/anyproto/any-sync/commonspace/deletionstate"
 	"github.com/anyproto/any-sync/commonspace/object/treemanager"
@@ -70,7 +70,7 @@ var createDiffSyncer = newDiffSyncer
 
 func (h *headSync) Init(a *app.App) (err error) {
 	shared := a.MustComponent(spacestate.CName).(*spacestate.SpaceState)
-	cfg := a.MustComponent("config").(config.ConfigGetter)
+	cfg := a.MustComponent("config").(config2.ConfigGetter)
 	h.spaceId = shared.SpaceId
 	h.spaceIsDeleted = shared.SpaceIsDeleted
 	h.syncPeriod = cfg.GetSpace().SyncPeriod

commonspace/headsync/headsync_test.go

@@ -23,8 +23,8 @@ import (
 	"github.com/anyproto/any-sync/commonspace/syncstatus"
 	"github.com/anyproto/any-sync/nodeconf"
 	"github.com/anyproto/any-sync/nodeconf/mock_nodeconf"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"sync/atomic"
 	"testing"
 )

commonspace/headsync/mock_headsync/mock_headsync.go

@@ -8,7 +8,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockDiffSyncer is a mock of DiffSyncer interface.

commonspace/headsync/synclogger.go

@@ -1,41 +0,0 @@
-package headsync
-
-import (
-	"time"
-
-	"github.com/anyproto/any-sync/app/logger"
-	"go.uber.org/zap"
-)
-
-type syncLogger struct {
-	lastLogged  map[string]time.Time
-	logInterval time.Duration
-	logger.CtxLogger
-}
-
-func newSyncLogger(log logger.CtxLogger, syncLogPeriodSecs int) syncLogger {
-	return syncLogger{
-		lastLogged:  map[string]time.Time{},
-		logInterval: time.Duration(syncLogPeriodSecs) * time.Second,
-		CtxLogger:   log,
-	}
-}
-
-func (s syncLogger) logSyncDone(peerId string, newIds, changedIds, removedIds, deltedIds int) {
-	now := time.Now()
-	differentIds := newIds + changedIds + removedIds + deltedIds
-	// always logging if some ids are different or there are no log interval
-	if differentIds == 0 && s.logInterval > 0 {
-		lastLogged := s.lastLogged[peerId]
-		if now.Before(lastLogged.Add(s.logInterval)) {
-			return
-		}
-	}
-	s.lastLogged[peerId] = now
-	s.Info("sync done:", zap.Int("newIds", newIds),
-		zap.Int("changedIds", changedIds),
-		zap.Int("removedIds", removedIds),
-		zap.Int("already deleted ids", deltedIds),
-		zap.String("peerId", peerId),
-	)
-}

commonspace/object/acl/list/mock_list/mock_list.go

@@ -10,7 +10,7 @@ import (
 	list "github.com/anyproto/any-sync/commonspace/object/acl/list"
 	consensusproto "github.com/anyproto/any-sync/consensus/consensusproto"
 	crypto "github.com/anyproto/any-sync/util/crypto"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockAclList is a mock of AclList interface.

commonspace/object/acl/liststorage/mock_liststorage/mock_liststorage.go

@@ -9,7 +9,7 @@ import (
 	reflect "reflect"
 
 	consensusproto "github.com/anyproto/any-sync/consensus/consensusproto"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockListStorage is a mock of ListStorage interface.

commonspace/object/tree/objecttree/mock_objecttree/mock_objecttree.go

@@ -13,7 +13,7 @@ import (
 	objecttree "github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
 	treechangeproto "github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	treestorage "github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockObjectTree is a mock of ObjectTree interface.

commonspace/object/tree/synctree/mock_synctree/mock_synctree.go

@@ -15,7 +15,7 @@ import (
 	treechangeproto "github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	treestorage "github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
 	spacesyncproto "github.com/anyproto/any-sync/commonspace/spacesyncproto"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockSyncTree is a mock of SyncTree interface.

commonspace/object/tree/synctree/synctree_test.go

@@ -11,8 +11,8 @@ import (
 	"github.com/anyproto/any-sync/commonspace/objectsync"
 	"github.com/anyproto/any-sync/commonspace/syncstatus"
 	"github.com/anyproto/any-sync/nodeconf"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"testing"
 )
 

commonspace/object/tree/synctree/synctreehandler_test.go

@@ -9,8 +9,8 @@ import (
 	"github.com/anyproto/any-sync/commonspace/object/tree/synctree/mock_synctree"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	"github.com/anyproto/any-sync/commonspace/syncstatus"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 )
 
 type testObjTreeMock struct {

commonspace/object/tree/synctree/treeremotegetter_test.go

@@ -12,8 +12,8 @@ import (
 	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/peer/mock_peer"
 	"github.com/gogo/protobuf/proto"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 )
 
 type treeRemoteGetterFixture struct {

commonspace/object/tree/synctree/treesyncprotocol_test.go

@@ -8,8 +8,8 @@ import (
 	"github.com/anyproto/any-sync/commonspace/object/tree/objecttree/mock_objecttree"
 	"github.com/anyproto/any-sync/commonspace/object/tree/synctree/mock_synctree"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"testing"
 )
 

commonspace/object/tree/synctree/updatelistener/mock_updatelistener/mock_updatelistener.go

@@ -8,7 +8,7 @@ import (
 	reflect "reflect"
 
 	objecttree "github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockUpdateListener is a mock of UpdateListener interface.

commonspace/object/tree/treestorage/mock_treestorage/mock_treestorage.go

@@ -9,7 +9,7 @@ import (
 	reflect "reflect"
 
 	treechangeproto "github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockTreeStorage is a mock of TreeStorage interface.

commonspace/object/treemanager/mock_treemanager/mock_treemanager.go

@@ -11,7 +11,7 @@ import (
 	app "github.com/anyproto/any-sync/app"
 	objecttree "github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
 	treemanager "github.com/anyproto/any-sync/commonspace/object/treemanager"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockTreeManager is a mock of TreeManager interface.

commonspace/objectsync/mock_objectsync/mock_objectsync.go

@@ -12,7 +12,7 @@ import (
 	app "github.com/anyproto/any-sync/app"
 	objectsync "github.com/anyproto/any-sync/commonspace/objectsync"
 	spacesyncproto "github.com/anyproto/any-sync/commonspace/spacesyncproto"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockObjectSync is a mock of ObjectSync interface.

commonspace/objecttreebuilder/mock_objecttreebuilder/mock_objecttreebuilder.go

@@ -12,7 +12,7 @@ import (
 	updatelistener "github.com/anyproto/any-sync/commonspace/object/tree/synctree/updatelistener"
 	treestorage "github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
 	objecttreebuilder "github.com/anyproto/any-sync/commonspace/objecttreebuilder"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockTreeBuilder is a mock of TreeBuilder interface.

commonspace/peermanager/mock_peermanager/mock_peermanager.go

@@ -11,7 +11,7 @@ import (
 	app "github.com/anyproto/any-sync/app"
 	spacesyncproto "github.com/anyproto/any-sync/commonspace/spacesyncproto"
 	peer "github.com/anyproto/any-sync/net/peer"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockPeerManager is a mock of PeerManager interface.

commonspace/requestmanager/requestmanager_test.go

@@ -12,8 +12,8 @@ import (
 	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/peer/mock_peer"
 	"github.com/anyproto/any-sync/net/pool/mock_pool"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"storj.io/drpc"
 	"storj.io/drpc/drpcconn"
 )

commonspace/settings/deleter_test.go

@@ -6,7 +6,7 @@ import (
 	"github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
 	"github.com/anyproto/any-sync/commonspace/object/treemanager/mock_treemanager"
 	"github.com/anyproto/any-sync/commonspace/spacestorage/mock_spacestorage"
-	"github.com/golang/mock/gomock"
+	"go.uber.org/mock/gomock"
 	"testing"
 )
 

commonspace/settings/deletionmanager_test.go

@@ -6,8 +6,8 @@ import (
 	"github.com/anyproto/any-sync/commonspace/object/treemanager/mock_treemanager"
 	"github.com/anyproto/any-sync/commonspace/settings/mock_settings"
 	"github.com/anyproto/any-sync/commonspace/settings/settingsstate"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"testing"
 )
 

commonspace/settings/mock_settings/mock_settings.go

@@ -9,7 +9,7 @@ import (
 	reflect "reflect"
 
 	settingsstate "github.com/anyproto/any-sync/commonspace/settings/settingsstate"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockDeletionManager is a mock of DeletionManager interface.

commonspace/settings/settingsobject_test.go

@@ -16,8 +16,8 @@ import (
 	"github.com/anyproto/any-sync/commonspace/settings/settingsstate"
 	"github.com/anyproto/any-sync/commonspace/settings/settingsstate/mock_settingsstate"
 	"github.com/anyproto/any-sync/commonspace/spacestorage/mock_spacestorage"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"sync"
 	"testing"
 	"time"

commonspace/settings/settingsstate/mock_settingsstate/mock_settingsstate.go

@@ -9,7 +9,7 @@ import (
 
 	objecttree "github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
 	settingsstate "github.com/anyproto/any-sync/commonspace/settings/settingsstate"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockStateBuilder is a mock of StateBuilder interface.

commonspace/settings/settingsstate/statebuilder_test.go

@@ -4,8 +4,8 @@ import (
 	"github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
 	"github.com/anyproto/any-sync/commonspace/object/tree/objecttree/mock_objecttree"
 	"github.com/anyproto/any-sync/commonspace/spacesyncproto"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"testing"
 )
 

commonspace/spacestorage/mock_spacestorage/mock_spacestorage.go

@@ -13,7 +13,7 @@ import (
 	treechangeproto "github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	treestorage "github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
 	spacesyncproto "github.com/anyproto/any-sync/commonspace/spacesyncproto"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockSpaceStorage is a mock of SpaceStorage interface.

commonspace/spacesyncproto/mock_spacesyncproto/mock_spacesyncproto.go

@@ -9,7 +9,7 @@ import (
 	reflect "reflect"
 
 	spacesyncproto "github.com/anyproto/any-sync/commonspace/spacesyncproto"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 	drpc "storj.io/drpc"
 )
 

consensus/consensusclient/client.go

@@ -36,13 +36,13 @@ type Watcher interface {
 
 type Service interface {
 	// AddLog adds new log to consensus servers
-	AddLog(ctx context.Context, clog *consensusproto.Log) (err error)
+	AddLog(ctx context.Context, rec *consensusproto.RawRecordWithId) (err error)
 	// AddRecord adds new record to consensus servers
-	AddRecord(ctx context.Context, logId []byte, clog *consensusproto.RawRecord) (record *consensusproto.RawRecordWithId, err error)
+	AddRecord(ctx context.Context, logId string, rec *consensusproto.RawRecord) (record *consensusproto.RawRecordWithId, err error)
 	// Watch starts watching to given logId and calls watcher when any relative event received
-	Watch(logId []byte, w Watcher) (err error)
+	Watch(logId string, w Watcher) (err error)
 	// UnWatch stops watching given logId and removes watcher
-	UnWatch(logId []byte) (err error)
+	UnWatch(logId string) (err error)
 	app.ComponentRunnable
 }
 
@@ -86,10 +86,10 @@ func (s *service) doClient(ctx context.Context, fn func(cl consensusproto.DRPCCo
 	return fn(consensusproto.NewDRPCConsensusClient(dc))
 }
 
-func (s *service) AddLog(ctx context.Context, clog *consensusproto.Log) (err error) {
+func (s *service) AddLog(ctx context.Context, rec *consensusproto.RawRecordWithId) (err error) {
 	return s.doClient(ctx, func(cl consensusproto.DRPCConsensusClient) error {
 		if _, err = cl.LogAdd(ctx, &consensusproto.LogAddRequest{
-			Log: clog,
+			Record: rec,
 		}); err != nil {
 			return rpcerr.Unwrap(err)
 		}
@@ -97,11 +97,11 @@ func (s *service) AddLog(ctx context.Context, clog *consensusproto.Log) (err err
 	})
 }
 
-func (s *service) AddRecord(ctx context.Context, logId []byte, clog *consensusproto.RawRecord) (record *consensusproto.RawRecordWithId, err error) {
+func (s *service) AddRecord(ctx context.Context, logId string, rec *consensusproto.RawRecord) (record *consensusproto.RawRecordWithId, err error) {
 	err = s.doClient(ctx, func(cl consensusproto.DRPCConsensusClient) error {
 		if record, err = cl.RecordAdd(ctx, &consensusproto.RecordAddRequest{
 			LogId:  logId,
-			Record: clog,
+			Record: rec,
 		}); err != nil {
 			return rpcerr.Unwrap(err)
 		}
@@ -110,30 +110,30 @@ func (s *service) AddRecord(ctx context.Context, logId []byte, clog *consensuspr
 	return
 }
 
-func (s *service) Watch(logId []byte, w Watcher) (err error) {
+func (s *service) Watch(logId string, w Watcher) (err error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	if _, ok := s.watchers[string(logId)]; ok {
+	if _, ok := s.watchers[logId]; ok {
 		return ErrWatcherExists
 	}
-	s.watchers[string(logId)] = w
+	s.watchers[logId] = w
 	if s.stream != nil {
-		if wErr := s.stream.WatchIds([][]byte{logId}); wErr != nil {
+		if wErr := s.stream.WatchIds([]string{logId}); wErr != nil {
 			log.Warn("WatchIds error", zap.Error(wErr))
 		}
 	}
 	return
 }
 
-func (s *service) UnWatch(logId []byte) (err error) {
+func (s *service) UnWatch(logId string) (err error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	if _, ok := s.watchers[string(logId)]; !ok {
+	if _, ok := s.watchers[logId]; !ok {
 		return ErrWatcherNotExists
 	}
-	delete(s.watchers, string(logId))
+	delete(s.watchers, logId)
 	if s.stream != nil {
-		if wErr := s.stream.UnwatchIds([][]byte{logId}); wErr != nil {
+		if wErr := s.stream.UnwatchIds([]string{logId}); wErr != nil {
 			log.Warn("UnWatchIds error", zap.Error(wErr))
 		}
 	}
@@ -182,9 +182,9 @@ func (s *service) streamWatcher() {
 
 		// collect ids and setup stream
 		s.mu.Lock()
-		var logIds = make([][]byte, 0, len(s.watchers))
+		var logIds = make([]string, 0, len(s.watchers))
 		for id := range s.watchers {
-			logIds = append(logIds, []byte(id))
+			logIds = append(logIds, id)
 		}
 		s.stream = st
 		s.mu.Unlock()
@@ -212,17 +212,19 @@ func (s *service) streamReader() error {
 		if len(events) == 0 {
 			return s.stream.Err()
 		}
+		s.mu.Lock()
 		for _, e := range events {
-			if w, ok := s.watchers[string(e.LogId)]; ok {
+			if w, ok := s.watchers[e.LogId]; ok {
 				if e.Error == nil {
 					w.AddConsensusRecords(e.Records)
 				} else {
 					w.AddConsensusError(rpcerr.Err(uint64(e.Error.Error)))
 				}
 			} else {
-				log.Warn("received unexpected log id", zap.Binary("logId", e.LogId))
+				log.Warn("received unexpected log id", zap.String("logId", e.LogId))
 			}
 		}
+		s.mu.Unlock()
 	}
 }
 

consensus/consensusclient/client_test.go

@@ -12,9 +12,9 @@ import (
 	"github.com/anyproto/any-sync/nodeconf/mock_nodeconf"
 	"github.com/anyproto/any-sync/testutil/accounttest"
 	"github.com/anyproto/any-sync/util/cidutil"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"sync"
 	"testing"
 	"time"
@@ -24,13 +24,13 @@ func TestService_Watch(t *testing.T) {
 	t.Run("not found error", func(t *testing.T) {
 		fx := newFixture(t).run(t)
 		defer fx.Finish()
-		var logId = []byte{'1'}
+		var logId = "1"
 		w := &testWatcher{ready: make(chan struct{})}
 		require.NoError(t, fx.Watch(logId, w))
 		st := fx.testServer.waitStream(t)
 		req, err := st.Recv()
 		require.NoError(t, err)
-		assert.Equal(t, [][]byte{logId}, req.WatchIds)
+		assert.Equal(t, []string{logId}, req.WatchIds)
 		require.NoError(t, st.Send(&consensusproto.LogWatchEvent{
 			LogId: logId,
 			Error: &consensusproto.Err{
@@ -44,7 +44,7 @@ func TestService_Watch(t *testing.T) {
 	t.Run("watcherExists error", func(t *testing.T) {
 		fx := newFixture(t).run(t)
 		defer fx.Finish()
-		var logId = []byte{'1'}
+		var logId = "1"
 		w := &testWatcher{}
 		require.NoError(t, fx.Watch(logId, w))
 		require.Error(t, fx.Watch(logId, w))
@@ -55,20 +55,20 @@ func TestService_Watch(t *testing.T) {
 	t.Run("watch", func(t *testing.T) {
 		fx := newFixture(t).run(t)
 		defer fx.Finish()
-		var logId1 = []byte{'1'}
+		var logId1 = "1"
 		w := &testWatcher{}
 		require.NoError(t, fx.Watch(logId1, w))
 		st := fx.testServer.waitStream(t)
 		req, err := st.Recv()
 		require.NoError(t, err)
-		assert.Equal(t, [][]byte{logId1}, req.WatchIds)
+		assert.Equal(t, []string{logId1}, req.WatchIds)
 
-		var logId2 = []byte{'2'}
+		var logId2 = "2"
 		w = &testWatcher{}
 		require.NoError(t, fx.Watch(logId2, w))
 		req, err = st.Recv()
 		require.NoError(t, err)
-		assert.Equal(t, [][]byte{logId2}, req.WatchIds)
+		assert.Equal(t, []string{logId2}, req.WatchIds)
 
 		fx.testServer.releaseStream <- nil
 	})
@@ -78,14 +78,14 @@ func TestService_UnWatch(t *testing.T) {
 	t.Run("no watcher", func(t *testing.T) {
 		fx := newFixture(t).run(t)
 		defer fx.Finish()
-		require.Error(t, fx.UnWatch([]byte{'1'}))
+		require.Error(t, fx.UnWatch("1"))
 	})
 	t.Run("success", func(t *testing.T) {
 		fx := newFixture(t).run(t)
 		defer fx.Finish()
 		w := &testWatcher{}
-		require.NoError(t, fx.Watch([]byte{'1'}, w))
-		assert.NoError(t, fx.UnWatch([]byte{'1'}))
+		require.NoError(t, fx.Watch("1", w))
+		assert.NoError(t, fx.UnWatch("1"))
 	})
 }
 
@@ -113,13 +113,13 @@ func TestService_Init(t *testing.T) {
 func TestService_AddLog(t *testing.T) {
 	fx := newFixture(t).run(t)
 	defer fx.Finish()
-	assert.NoError(t, fx.AddLog(ctx, &consensusproto.Log{}))
+	assert.NoError(t, fx.AddLog(ctx, &consensusproto.RawRecordWithId{}))
 }
 
 func TestService_AddRecord(t *testing.T) {
 	fx := newFixture(t).run(t)
 	defer fx.Finish()
-	rec, err := fx.AddRecord(ctx, []byte{'1'}, &consensusproto.RawRecord{})
+	rec, err := fx.AddRecord(ctx, "1", &consensusproto.RawRecord{})
 	require.NoError(t, err)
 	assert.NotEmpty(t, rec)
 }

consensus/consensusclient/mock_consensusclient/mock_consensusclient.go

@@ -11,7 +11,7 @@ import (
 	app "github.com/anyproto/any-sync/app"
 	consensusclient "github.com/anyproto/any-sync/consensus/consensusclient"
 	consensusproto "github.com/anyproto/any-sync/consensus/consensusproto"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockService is a mock of Service interface.
@@ -38,7 +38,7 @@ func (m *MockService) EXPECT() *MockServiceMockRecorder {
 }
 
 // AddLog mocks base method.
-func (m *MockService) AddLog(arg0 context.Context, arg1 *consensusproto.Log) error {
+func (m *MockService) AddLog(arg0 context.Context, arg1 *consensusproto.RawRecordWithId) error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "AddLog", arg0, arg1)
 	ret0, _ := ret[0].(error)
@@ -52,7 +52,7 @@ func (mr *MockServiceMockRecorder) AddLog(arg0, arg1 interface{}) *gomock.Call {
 }
 
 // AddRecord mocks base method.
-func (m *MockService) AddRecord(arg0 context.Context, arg1 []byte, arg2 *consensusproto.RawRecord) (*consensusproto.RawRecordWithId, error) {
+func (m *MockService) AddRecord(arg0 context.Context, arg1 string, arg2 *consensusproto.RawRecord) (*consensusproto.RawRecordWithId, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "AddRecord", arg0, arg1, arg2)
 	ret0, _ := ret[0].(*consensusproto.RawRecordWithId)
@@ -123,7 +123,7 @@ func (mr *MockServiceMockRecorder) Run(arg0 interface{}) *gomock.Call {
 }
 
 // UnWatch mocks base method.
-func (m *MockService) UnWatch(arg0 []byte) error {
+func (m *MockService) UnWatch(arg0 string) error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "UnWatch", arg0)
 	ret0, _ := ret[0].(error)
@@ -137,7 +137,7 @@ func (mr *MockServiceMockRecorder) UnWatch(arg0 interface{}) *gomock.Call {
 }
 
 // Watch mocks base method.
-func (m *MockService) Watch(arg0 []byte, arg1 consensusclient.Watcher) error {
+func (m *MockService) Watch(arg0 string, arg1 consensusclient.Watcher) error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "Watch", arg0, arg1)
 	ret0, _ := ret[0].(error)

consensus/consensusclient/stream.go

@@ -23,13 +23,13 @@ type stream struct {
 	err       error
 }
 
-func (s *stream) WatchIds(logIds [][]byte) (err error) {
+func (s *stream) WatchIds(logIds []string) (err error) {
 	return s.rpcStream.Send(&consensusproto.LogWatchRequest{
 		WatchIds: logIds,
 	})
 }
 
-func (s *stream) UnwatchIds(logIds [][]byte) (err error) {
+func (s *stream) UnwatchIds(logIds []string) (err error) {
 	return s.rpcStream.Send(&consensusproto.LogWatchRequest{
 		UnwatchIds: logIds,
 	})

consensus/consensusproto/consensus.pb.go

@@ -29,6 +29,8 @@ const (
 	ErrCodes_LogExists      ErrCodes = 1
 	ErrCodes_LogNotFound    ErrCodes = 2
 	ErrCodes_RecordConflict ErrCodes = 3
+	ErrCodes_Forbidden      ErrCodes = 4
+	ErrCodes_InvalidPayload ErrCodes = 5
 	ErrCodes_ErrorOffset    ErrCodes = 500
 )
 
@@ -37,6 +39,8 @@ var ErrCodes_name = map[int32]string{
 	1:   "LogExists",
 	2:   "LogNotFound",
 	3:   "RecordConflict",
+	4:   "Forbidden",
+	5:   "InvalidPayload",
 	500: "ErrorOffset",
 }
 
@@ -45,6 +49,8 @@ var ErrCodes_value = map[string]int32{
 	"LogExists":      1,
 	"LogNotFound":    2,
 	"RecordConflict": 3,
+	"Forbidden":      4,
+	"InvalidPayload": 5,
 	"ErrorOffset":    500,
 }
 
@@ -57,8 +63,7 @@ func (ErrCodes) EnumDescriptor() ([]byte, []int) {
 }
 
 type Log struct {
-	Id      []byte             `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
-	Payload []byte             `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"`
+	Id      string             `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
 	Records []*RawRecordWithId `protobuf:"bytes,3,rep,name=records,proto3" json:"records,omitempty"`
 }
 
@@ -95,18 +100,11 @@ func (m *Log) XXX_DiscardUnknown() {
 
 var xxx_messageInfo_Log proto.InternalMessageInfo
 
-func (m *Log) GetId() []byte {
+func (m *Log) GetId() string {
 	if m != nil {
 		return m.Id
 	}
-	return nil
-}
-
-func (m *Log) GetPayload() []byte {
-	if m != nil {
-		return m.Payload
-	}
-	return nil
+	return ""
 }
 
 func (m *Log) GetRecords() []*RawRecordWithId {
@@ -344,7 +342,8 @@ func (m *Ok) XXX_DiscardUnknown() {
 var xxx_messageInfo_Ok proto.InternalMessageInfo
 
 type LogAddRequest struct {
-	Log *Log `protobuf:"bytes,1,opt,name=log,proto3" json:"log,omitempty"`
+	// first record in the log, consensus node not sign it
+	Record *RawRecordWithId `protobuf:"bytes,1,opt,name=record,proto3" json:"record,omitempty"`
 }
 
 func (m *LogAddRequest) Reset()         { *m = LogAddRequest{} }
@@ -380,15 +379,15 @@ func (m *LogAddRequest) XXX_DiscardUnknown() {
 
 var xxx_messageInfo_LogAddRequest proto.InternalMessageInfo
 
-func (m *LogAddRequest) GetLog() *Log {
+func (m *LogAddRequest) GetRecord() *RawRecordWithId {
 	if m != nil {
-		return m.Log
+		return m.Record
 	}
 	return nil
 }
 
 type RecordAddRequest struct {
-	LogId  []byte     `protobuf:"bytes,1,opt,name=logId,proto3" json:"logId,omitempty"`
+	LogId  string     `protobuf:"bytes,1,opt,name=logId,proto3" json:"logId,omitempty"`
 	Record *RawRecord `protobuf:"bytes,2,opt,name=record,proto3" json:"record,omitempty"`
 }
 
@@ -425,11 +424,11 @@ func (m *RecordAddRequest) XXX_DiscardUnknown() {
 
 var xxx_messageInfo_RecordAddRequest proto.InternalMessageInfo
 
-func (m *RecordAddRequest) GetLogId() []byte {
+func (m *RecordAddRequest) GetLogId() string {
 	if m != nil {
 		return m.LogId
 	}
-	return nil
+	return ""
 }
 
 func (m *RecordAddRequest) GetRecord() *RawRecord {
@@ -440,8 +439,8 @@ func (m *RecordAddRequest) GetRecord() *RawRecord {
 }
 
 type LogWatchRequest struct {
-	WatchIds   [][]byte `protobuf:"bytes,1,rep,name=watchIds,proto3" json:"watchIds,omitempty"`
-	UnwatchIds [][]byte `protobuf:"bytes,2,rep,name=unwatchIds,proto3" json:"unwatchIds,omitempty"`
+	WatchIds   []string `protobuf:"bytes,1,rep,name=watchIds,proto3" json:"watchIds,omitempty"`
+	UnwatchIds []string `protobuf:"bytes,2,rep,name=unwatchIds,proto3" json:"unwatchIds,omitempty"`
 }
 
 func (m *LogWatchRequest) Reset()         { *m = LogWatchRequest{} }
@@ -477,14 +476,14 @@ func (m *LogWatchRequest) XXX_DiscardUnknown() {
 
 var xxx_messageInfo_LogWatchRequest proto.InternalMessageInfo
 
-func (m *LogWatchRequest) GetWatchIds() [][]byte {
+func (m *LogWatchRequest) GetWatchIds() []string {
 	if m != nil {
 		return m.WatchIds
 	}
 	return nil
 }
 
-func (m *LogWatchRequest) GetUnwatchIds() [][]byte {
+func (m *LogWatchRequest) GetUnwatchIds() []string {
 	if m != nil {
 		return m.UnwatchIds
 	}
@@ -492,7 +491,7 @@ func (m *LogWatchRequest) GetUnwatchIds() [][]byte {
 }
 
 type LogWatchEvent struct {
-	LogId   []byte             `protobuf:"bytes,1,opt,name=logId,proto3" json:"logId,omitempty"`
+	LogId   string             `protobuf:"bytes,1,opt,name=logId,proto3" json:"logId,omitempty"`
 	Records []*RawRecordWithId `protobuf:"bytes,2,rep,name=records,proto3" json:"records,omitempty"`
 	Error   *Err               `protobuf:"bytes,3,opt,name=error,proto3" json:"error,omitempty"`
 }
@@ -530,11 +529,11 @@ func (m *LogWatchEvent) XXX_DiscardUnknown() {
 
 var xxx_messageInfo_LogWatchEvent proto.InternalMessageInfo
 
-func (m *LogWatchEvent) GetLogId() []byte {
+func (m *LogWatchEvent) GetLogId() string {
 	if m != nil {
 		return m.LogId
 	}
-	return nil
+	return ""
 }
 
 func (m *LogWatchEvent) GetRecords() []*RawRecordWithId {
@@ -614,46 +613,47 @@ func init() {
 }
 
 var fileDescriptor_b8d7f1c16b400059 = []byte{
-	// 618 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x54, 0xcf, 0x6e, 0xd3, 0x4e,
-	0x10, 0xce, 0xda, 0x6d, 0x1a, 0x4f, 0x7e, 0x4d, 0xfd, 0x1b, 0x10, 0x32, 0x11, 0x75, 0x23, 0x4b,
-	0x48, 0xa5, 0x42, 0x29, 0x04, 0x81, 0x84, 0x7a, 0x82, 0x2a, 0x48, 0x91, 0x42, 0x8b, 0x8c, 0x50,
-	0x25, 0xb8, 0x60, 0xbc, 0x5b, 0xd7, 0x34, 0xf5, 0x86, 0xdd, 0x4d, 0xff, 0x3c, 0x03, 0x17, 0x5e,
-	0x80, 0xf7, 0xe1, 0xd8, 0x23, 0x47, 0xd4, 0x5e, 0x78, 0x00, 0x1e, 0x00, 0x79, 0x1d, 0x3b, 0x6e,
-	0x9a, 0x80, 0xb8, 0x24, 0x3b, 0xdf, 0xfc, 0xfb, 0x66, 0xe6, 0x93, 0x61, 0x33, 0xe4, 0x89, 0x64,
-	0x89, 0x1c, 0xc9, 0xc9, 0x6b, 0x28, 0xb8, 0xe2, 0x9b, 0xfa, 0xb7, 0x84, 0xb6, 0x35, 0x80, 0x8d,
-	0x02, 0x78, 0x95, 0xda, 0xde, 0x47, 0x30, 0xfb, 0x3c, 0xc2, 0x06, 0x18, 0x31, 0x75, 0x48, 0x8b,
-	0xac, 0xff, 0xe7, 0x1b, 0x31, 0x45, 0x07, 0x96, 0x86, 0xc1, 0xd9, 0x80, 0x07, 0xd4, 0x31, 0x34,
-	0x98, 0x9b, 0xf8, 0x14, 0x96, 0x04, 0x0b, 0xb9, 0xa0, 0xd2, 0x31, 0x5b, 0xe6, 0x7a, 0xbd, 0xb3,
-	0xd6, 0xbe, 0x5a, 0xb2, 0xed, 0x07, 0x27, 0xbe, 0x8e, 0xd8, 0x8b, 0xd5, 0x41, 0x8f, 0xfa, 0x79,
-	0xbc, 0xf7, 0x95, 0x80, 0x55, 0x38, 0xcb, 0x2d, 0xc8, 0xd5, 0x16, 0x77, 0xc0, 0x92, 0x71, 0x94,
-	0x04, 0x6a, 0x24, 0xd8, 0xb8, 0xfd, 0x04, 0xc0, 0x0d, 0xb0, 0x83, 0x30, 0x64, 0x43, 0xc5, 0x45,
-	0x8f, 0xb2, 0x44, 0xc5, 0xea, 0xcc, 0x31, 0x75, 0xd0, 0x35, 0x1c, 0xef, 0xc3, 0xff, 0x39, 0xf6,
-	0xba, 0xa8, 0xb8, 0xa0, 0x83, 0xaf, 0x3b, 0xbc, 0x2d, 0x58, 0x99, 0xe2, 0xfe, 0x07, 0x92, 0xd9,
-	0xc6, 0x52, 0x76, 0x56, 0xba, 0x31, 0x2f, 0x81, 0xea, 0x78, 0xb0, 0x5b, 0x50, 0x1d, 0x0a, 0x76,
-	0xdc, 0xcb, 0x52, 0x2c, 0x7f, 0x6c, 0x61, 0x13, 0x6a, 0x71, 0x4e, 0x38, 0x9b, 0xaa, 0xb0, 0x11,
-	0x61, 0x81, 0x06, 0x2a, 0x18, 0x0f, 0xa2, 0xdf, 0xe9, 0x1a, 0x54, 0x7c, 0xc4, 0xa4, 0x0a, 0x8e,
-	0x86, 0x9a, 0xb4, 0xe9, 0x4f, 0x00, 0x6f, 0x01, 0x8c, 0xdd, 0x43, 0xef, 0x09, 0x2c, 0xf7, 0x79,
-	0xf4, 0x8c, 0x52, 0x9f, 0x7d, 0x1a, 0x31, 0xa9, 0xf0, 0x2e, 0x98, 0x03, 0x1e, 0xe9, 0xce, 0xf5,
-	0xce, 0x8d, 0xe9, 0xd3, 0xf4, 0x79, 0xe4, 0xa7, 0x7e, 0xef, 0x1d, 0xd8, 0x19, 0xdb, 0x52, 0xea,
-	0x4d, 0x58, 0x1c, 0xf0, 0xa8, 0x97, 0x4f, 0x9a, 0x19, 0xf8, 0x10, 0xaa, 0xd9, 0xfd, 0x34, 0xe7,
-	0x7a, 0xe7, 0xf6, 0xdc, 0x73, 0xfb, 0xe3, 0x40, 0xef, 0x25, 0xac, 0xf4, 0x79, 0xb4, 0x17, 0xa8,
-	0xf0, 0x20, 0xaf, 0xdd, 0x84, 0xda, 0x49, 0x6a, 0xf7, 0xa8, 0x74, 0x48, 0xcb, 0x4c, 0x67, 0xcf,
-	0x6d, 0x74, 0x01, 0x46, 0x49, 0xe1, 0x35, 0xb4, 0xb7, 0x84, 0x78, 0x9f, 0x89, 0x1e, 0x52, 0xd7,
-	0xeb, 0x1e, 0xb3, 0x64, 0x1e, 0xd3, 0x92, 0x32, 0x8d, 0x7f, 0x53, 0x26, 0xde, 0x83, 0x45, 0x26,
-	0x04, 0x17, 0x7a, 0xff, 0x33, 0xf6, 0xd6, 0x15, 0xc2, 0xcf, 0x22, 0xbc, 0xc7, 0x60, 0x76, 0x85,
-	0xc0, 0x76, 0x9e, 0x91, 0x52, 0x68, 0x74, 0x9c, 0x19, 0x19, 0xdb, 0x9c, 0x32, 0x39, 0x4e, 0xdb,
-	0x78, 0x0f, 0xb5, 0x1c, 0xc2, 0x06, 0xc0, 0x9b, 0x84, 0x9d, 0x0e, 0x59, 0xa8, 0x18, 0xb5, 0x2b,
-	0xb8, 0x0c, 0x56, 0x9f, 0x47, 0xdd, 0xd3, 0x58, 0x2a, 0x69, 0x13, 0x5c, 0x81, 0x7a, 0x9f, 0x47,
-	0x3b, 0x5c, 0xbd, 0xe0, 0xa3, 0x84, 0xda, 0x06, 0x22, 0x34, 0x32, 0xda, 0xdb, 0x3c, 0xd9, 0x1f,
-	0xc4, 0xa1, 0xb2, 0x4d, 0xb4, 0xa1, 0xde, 0x4d, 0x0b, 0xef, 0xee, 0xef, 0x4b, 0xa6, 0xec, 0x5f,
-	0x66, 0xe7, 0x27, 0x01, 0x6b, 0x3b, 0x27, 0x81, 0x5b, 0x50, 0xcd, 0x84, 0x81, 0xab, 0x33, 0x44,
-	0x30, 0xb9, 0x7a, 0x13, 0xa7, 0xdd, 0xbb, 0x87, 0xb8, 0x03, 0x56, 0xa1, 0x0e, 0x6c, 0x5d, 0xdb,
-	0xe2, 0x94, 0x70, 0x9a, 0x7f, 0xdb, 0x33, 0xee, 0x40, 0x2d, 0x3f, 0x20, 0xae, 0xcd, 0xa0, 0x53,
-	0x96, 0x4a, 0x73, 0x75, 0x5e, 0x80, 0xbe, 0xfd, 0x3a, 0x79, 0x40, 0x9e, 0x77, 0xbe, 0x5d, 0xb8,
-	0xe4, 0xfc, 0xc2, 0x25, 0x3f, 0x2e, 0x5c, 0xf2, 0xe5, 0xd2, 0xad, 0x9c, 0x5f, 0xba, 0x95, 0xef,
-	0x97, 0x6e, 0xe5, 0xad, 0x33, 0xef, 0x7b, 0xf8, 0xa1, 0xaa, 0xff, 0x1e, 0xfd, 0x0e, 0x00, 0x00,
-	0xff, 0xff, 0xbc, 0x83, 0xcb, 0xc1, 0x32, 0x05, 0x00, 0x00,
+	// 625 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x54, 0xcd, 0x6e, 0xd3, 0x40,
+	0x10, 0xce, 0xda, 0x69, 0x1a, 0x4f, 0x68, 0x6a, 0x06, 0x84, 0x4c, 0x44, 0xdd, 0xc8, 0xa7, 0x50,
+	0xa1, 0x14, 0x82, 0x10, 0x42, 0x3d, 0x41, 0x95, 0x8a, 0x48, 0xa1, 0xad, 0x8c, 0x50, 0x25, 0x38,
+	0xb9, 0xde, 0xad, 0x6b, 0x35, 0xf5, 0x86, 0xdd, 0x4d, 0x7f, 0xce, 0x1c, 0xb9, 0xf0, 0x02, 0xbc,
+	0x0f, 0xc7, 0x1e, 0x39, 0xa2, 0xf6, 0xc2, 0x03, 0xf0, 0x00, 0xc8, 0xeb, 0xd8, 0x4d, 0xd3, 0x86,
+	0x9f, 0x4b, 0xb2, 0xf3, 0xcd, 0xdf, 0x37, 0x33, 0x9f, 0x0c, 0xab, 0x21, 0x4f, 0x24, 0x4b, 0xe4,
+	0x48, 0x5e, 0xbe, 0x86, 0x82, 0x2b, 0xbe, 0xaa, 0x7f, 0x27, 0xd0, 0xb6, 0x06, 0xb0, 0x5e, 0x00,
+	0xdb, 0xa9, 0xed, 0x6d, 0x83, 0xd9, 0xe7, 0x11, 0xd6, 0xc1, 0x88, 0xa9, 0x43, 0x9a, 0xa4, 0x65,
+	0xf9, 0x46, 0x4c, 0xf1, 0x05, 0xcc, 0x0b, 0x16, 0x72, 0x41, 0xa5, 0x63, 0x36, 0xcd, 0x56, 0xad,
+	0xb3, 0xdc, 0xbe, 0x9a, 0xd8, 0xf6, 0x83, 0x63, 0x5f, 0x47, 0xec, 0xc4, 0x6a, 0xbf, 0x47, 0xfd,
+	0x3c, 0xde, 0xfb, 0x4a, 0xc0, 0x2a, 0x9c, 0xe8, 0xc0, 0xfc, 0x30, 0x38, 0x1d, 0xf0, 0x20, 0xab,
+	0x7e, 0xcb, 0xcf, 0x4d, 0x7c, 0x00, 0x96, 0x8c, 0xa3, 0x24, 0x50, 0x23, 0xc1, 0x1c, 0x43, 0xfb,
+	0x2e, 0x01, 0x5c, 0x01, 0x3b, 0x08, 0x43, 0x36, 0x54, 0x5c, 0xf4, 0x28, 0x4b, 0x54, 0xac, 0x4e,
+	0x1d, 0x53, 0x07, 0x5d, 0xc3, 0xf1, 0x11, 0xdc, 0xce, 0xb1, 0xb7, 0x45, 0xc5, 0xb2, 0x0e, 0xbe,
+	0xee, 0xf0, 0xd6, 0x60, 0x71, 0x8a, 0xfb, 0x1f, 0x48, 0x66, 0x7b, 0x31, 0xf2, 0xbd, 0x78, 0x09,
+	0x54, 0xc6, 0x83, 0xdd, 0x83, 0xca, 0x50, 0xb0, 0xa3, 0x5e, 0xbe, 0xb5, 0xb1, 0x85, 0x0d, 0xa8,
+	0xc6, 0x39, 0xe1, 0x6c, 0xaa, 0xc2, 0x46, 0x84, 0x32, 0x0d, 0x54, 0x30, 0x1e, 0x44, 0xbf, 0xd3,
+	0x35, 0xa8, 0xf8, 0x90, 0x49, 0x15, 0x1c, 0x0e, 0x35, 0x69, 0xd3, 0xbf, 0x04, 0xbc, 0x32, 0x18,
+	0x5b, 0x07, 0xde, 0x6b, 0x58, 0xe8, 0xf3, 0xe8, 0x25, 0xa5, 0x3e, 0xfb, 0x38, 0x62, 0x52, 0xe1,
+	0x73, 0xa8, 0x64, 0xeb, 0xd6, 0xcd, 0xff, 0xe1, 0x3a, 0xe3, 0x70, 0xef, 0x03, 0xd8, 0x19, 0x3e,
+	0x51, 0xec, 0x2e, 0xcc, 0x0d, 0x78, 0x54, 0x0c, 0x92, 0x19, 0xf8, 0xa4, 0x68, 0x61, 0xe8, 0x16,
+	0xf7, 0x67, 0xb6, 0x28, 0x8a, 0xbf, 0x81, 0xc5, 0x3e, 0x8f, 0x76, 0x02, 0x15, 0xee, 0xe7, 0xb5,
+	0x1b, 0x50, 0x3d, 0x4e, 0xed, 0x1e, 0x95, 0x0e, 0x69, 0x9a, 0x2d, 0xcb, 0x2f, 0x6c, 0x74, 0x01,
+	0x46, 0x49, 0xe1, 0x35, 0xb4, 0x77, 0x02, 0xf1, 0x3e, 0x13, 0x3d, 0xb6, 0xae, 0xd7, 0x3d, 0x62,
+	0xc9, 0x2c, 0xa6, 0x13, 0x5a, 0x35, 0xfe, 0x4f, 0xab, 0xf8, 0x10, 0xe6, 0x98, 0x10, 0x5c, 0xe8,
+	0x8b, 0xd4, 0x3a, 0x77, 0xa6, 0x13, 0xbb, 0x42, 0xf8, 0x59, 0x84, 0xf7, 0x0c, 0xcc, 0xae, 0x10,
+	0xd8, 0xce, 0x33, 0x52, 0x0a, 0xf5, 0x8e, 0x73, 0x43, 0xc6, 0x3a, 0xa7, 0x4c, 0x8e, 0xd3, 0x56,
+	0x3e, 0x11, 0xa8, 0xe6, 0x18, 0xd6, 0x01, 0xde, 0x25, 0xec, 0x64, 0xc8, 0x42, 0xc5, 0xa8, 0x5d,
+	0xc2, 0x05, 0xb0, 0xfa, 0x3c, 0xea, 0x9e, 0xc4, 0x52, 0x49, 0x9b, 0xe0, 0x22, 0xd4, 0xfa, 0x3c,
+	0xda, 0xe4, 0x6a, 0x83, 0x8f, 0x12, 0x6a, 0x1b, 0x88, 0x50, 0xcf, 0x78, 0xaf, 0xf3, 0x64, 0x6f,
+	0x10, 0x87, 0xca, 0x36, 0xd3, 0x9c, 0x0d, 0x2e, 0x76, 0x63, 0x4a, 0x59, 0x62, 0x97, 0xd3, 0x90,
+	0x5e, 0x72, 0x14, 0x0c, 0x62, 0xba, 0x9d, 0x49, 0xd6, 0x9e, 0x43, 0x1b, 0x6a, 0xdd, 0xb4, 0xf9,
+	0xd6, 0xde, 0x9e, 0x64, 0xca, 0xfe, 0x65, 0x76, 0x7e, 0x12, 0xb0, 0xd6, 0x73, 0xa2, 0xb8, 0x06,
+	0x95, 0x4c, 0x4e, 0xb8, 0x34, 0x4d, 0xff, 0x8a, 0xcc, 0x1a, 0x38, 0xed, 0xde, 0x3a, 0xc0, 0x4d,
+	0xb0, 0x0a, 0x05, 0x61, 0xf3, 0xda, 0xa6, 0xa7, 0xc4, 0xd5, 0xf8, 0xdb, 0x2d, 0x70, 0x13, 0xaa,
+	0xf9, 0x91, 0x71, 0xf9, 0x06, 0x3a, 0x93, 0x72, 0x6a, 0x2c, 0xcd, 0x0a, 0xd0, 0xfa, 0x68, 0x91,
+	0xc7, 0xe4, 0x55, 0xe7, 0xdb, 0xb9, 0x4b, 0xce, 0xce, 0x5d, 0xf2, 0xe3, 0xdc, 0x25, 0x5f, 0x2e,
+	0xdc, 0xd2, 0xd9, 0x85, 0x5b, 0xfa, 0x7e, 0xe1, 0x96, 0xde, 0x3b, 0xb3, 0xbe, 0x95, 0xbb, 0x15,
+	0xfd, 0xf7, 0xf4, 0x77, 0x00, 0x00, 0x00, 0xff, 0xff, 0x20, 0xdd, 0xae, 0xec, 0x4e, 0x05, 0x00,
+	0x00,
 }
 
 func (m *Log) Marshal() (dAtA []byte, err error) {
@@ -690,13 +690,6 @@ func (m *Log) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 			dAtA[i] = 0x1a
 		}
 	}
-	if len(m.Payload) > 0 {
-		i -= len(m.Payload)
-		copy(dAtA[i:], m.Payload)
-		i = encodeVarintConsensus(dAtA, i, uint64(len(m.Payload)))
-		i--
-		dAtA[i] = 0x12
-	}
 	if len(m.Id) > 0 {
 		i -= len(m.Id)
 		copy(dAtA[i:], m.Id)
@@ -887,9 +880,9 @@ func (m *LogAddRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
-	if m.Log != nil {
+	if m.Record != nil {
 		{
-			size, err := m.Log.MarshalToSizedBuffer(dAtA[:i])
+			size, err := m.Record.MarshalToSizedBuffer(dAtA[:i])
 			if err != nil {
 				return 0, err
 			}
@@ -1090,10 +1083,6 @@ func (m *Log) Size() (n int) {
 	if l > 0 {
 		n += 1 + l + sovConsensus(uint64(l))
 	}
-	l = len(m.Payload)
-	if l > 0 {
-		n += 1 + l + sovConsensus(uint64(l))
-	}
 	if len(m.Records) > 0 {
 		for _, e := range m.Records {
 			l = e.Size()
@@ -1184,8 +1173,8 @@ func (m *LogAddRequest) Size() (n int) {
 	}
 	var l int
 	_ = l
-	if m.Log != nil {
-		l = m.Log.Size()
+	if m.Record != nil {
+		l = m.Record.Size()
 		n += 1 + l + sovConsensus(uint64(l))
 	}
 	return n
@@ -1215,14 +1204,14 @@ func (m *LogWatchRequest) Size() (n int) {
 	var l int
 	_ = l
 	if len(m.WatchIds) > 0 {
-		for _, b := range m.WatchIds {
-			l = len(b)
+		for _, s := range m.WatchIds {
+			l = len(s)
 			n += 1 + l + sovConsensus(uint64(l))
 		}
 	}
 	if len(m.UnwatchIds) > 0 {
-		for _, b := range m.UnwatchIds {
-			l = len(b)
+		for _, s := range m.UnwatchIds {
+			l = len(s)
 			n += 1 + l + sovConsensus(uint64(l))
 		}
 	}
@@ -1303,7 +1292,7 @@ func (m *Log) Unmarshal(dAtA []byte) error {
 			if wireType != 2 {
 				return fmt.Errorf("proto: wrong wireType = %d for field Id", wireType)
 			}
-			var byteLen int
+			var stringLen uint64
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowConsensus
@@ -1313,59 +1302,23 @@ func (m *Log) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				byteLen |= int(b&0x7F) << shift
+				stringLen |= uint64(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
-			if byteLen < 0 {
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
 				return ErrInvalidLengthConsensus
 			}
-			postIndex := iNdEx + byteLen
+			postIndex := iNdEx + intStringLen
 			if postIndex < 0 {
 				return ErrInvalidLengthConsensus
 			}
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.Id = append(m.Id[:0], dAtA[iNdEx:postIndex]...)
-			if m.Id == nil {
-				m.Id = []byte{}
-			}
-			iNdEx = postIndex
-		case 2:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Payload", wireType)
-			}
-			var byteLen int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowConsensus
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				byteLen |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if byteLen < 0 {
-				return ErrInvalidLengthConsensus
-			}
-			postIndex := iNdEx + byteLen
-			if postIndex < 0 {
-				return ErrInvalidLengthConsensus
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.Payload = append(m.Payload[:0], dAtA[iNdEx:postIndex]...)
-			if m.Payload == nil {
-				m.Payload = []byte{}
-			}
+			m.Id = string(dAtA[iNdEx:postIndex])
 			iNdEx = postIndex
 		case 3:
 			if wireType != 2 {
@@ -1974,7 +1927,7 @@ func (m *LogAddRequest) Unmarshal(dAtA []byte) error {
 		switch fieldNum {
 		case 1:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Log", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field Record", wireType)
 			}
 			var msglen int
 			for shift := uint(0); ; shift += 7 {
@@ -2001,10 +1954,10 @@ func (m *LogAddRequest) Unmarshal(dAtA []byte) error {
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			if m.Log == nil {
-				m.Log = &Log{}
+			if m.Record == nil {
+				m.Record = &RawRecordWithId{}
 			}
-			if err := m.Log.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+			if err := m.Record.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
 				return err
 			}
 			iNdEx = postIndex
@@ -2062,7 +2015,7 @@ func (m *RecordAddRequest) Unmarshal(dAtA []byte) error {
 			if wireType != 2 {
 				return fmt.Errorf("proto: wrong wireType = %d for field LogId", wireType)
 			}
-			var byteLen int
+			var stringLen uint64
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowConsensus
@@ -2072,25 +2025,23 @@ func (m *RecordAddRequest) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				byteLen |= int(b&0x7F) << shift
+				stringLen |= uint64(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
-			if byteLen < 0 {
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
 				return ErrInvalidLengthConsensus
 			}
-			postIndex := iNdEx + byteLen
+			postIndex := iNdEx + intStringLen
 			if postIndex < 0 {
 				return ErrInvalidLengthConsensus
 			}
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.LogId = append(m.LogId[:0], dAtA[iNdEx:postIndex]...)
-			if m.LogId == nil {
-				m.LogId = []byte{}
-			}
+			m.LogId = string(dAtA[iNdEx:postIndex])
 			iNdEx = postIndex
 		case 2:
 			if wireType != 2 {
@@ -2182,7 +2133,7 @@ func (m *LogWatchRequest) Unmarshal(dAtA []byte) error {
 			if wireType != 2 {
 				return fmt.Errorf("proto: wrong wireType = %d for field WatchIds", wireType)
 			}
-			var byteLen int
+			var stringLen uint64
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowConsensus
@@ -2192,29 +2143,29 @@ func (m *LogWatchRequest) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				byteLen |= int(b&0x7F) << shift
+				stringLen |= uint64(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
-			if byteLen < 0 {
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
 				return ErrInvalidLengthConsensus
 			}
-			postIndex := iNdEx + byteLen
+			postIndex := iNdEx + intStringLen
 			if postIndex < 0 {
 				return ErrInvalidLengthConsensus
 			}
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.WatchIds = append(m.WatchIds, make([]byte, postIndex-iNdEx))
-			copy(m.WatchIds[len(m.WatchIds)-1], dAtA[iNdEx:postIndex])
+			m.WatchIds = append(m.WatchIds, string(dAtA[iNdEx:postIndex]))
 			iNdEx = postIndex
 		case 2:
 			if wireType != 2 {
 				return fmt.Errorf("proto: wrong wireType = %d for field UnwatchIds", wireType)
 			}
-			var byteLen int
+			var stringLen uint64
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowConsensus
@@ -2224,23 +2175,23 @@ func (m *LogWatchRequest) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				byteLen |= int(b&0x7F) << shift
+				stringLen |= uint64(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
-			if byteLen < 0 {
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
 				return ErrInvalidLengthConsensus
 			}
-			postIndex := iNdEx + byteLen
+			postIndex := iNdEx + intStringLen
 			if postIndex < 0 {
 				return ErrInvalidLengthConsensus
 			}
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.UnwatchIds = append(m.UnwatchIds, make([]byte, postIndex-iNdEx))
-			copy(m.UnwatchIds[len(m.UnwatchIds)-1], dAtA[iNdEx:postIndex])
+			m.UnwatchIds = append(m.UnwatchIds, string(dAtA[iNdEx:postIndex]))
 			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
@@ -2296,7 +2247,7 @@ func (m *LogWatchEvent) Unmarshal(dAtA []byte) error {
 			if wireType != 2 {
 				return fmt.Errorf("proto: wrong wireType = %d for field LogId", wireType)
 			}
-			var byteLen int
+			var stringLen uint64
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowConsensus
@@ -2306,25 +2257,23 @@ func (m *LogWatchEvent) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				byteLen |= int(b&0x7F) << shift
+				stringLen |= uint64(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
-			if byteLen < 0 {
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
 				return ErrInvalidLengthConsensus
 			}
-			postIndex := iNdEx + byteLen
+			postIndex := iNdEx + intStringLen
 			if postIndex < 0 {
 				return ErrInvalidLengthConsensus
 			}
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.LogId = append(m.LogId[:0], dAtA[iNdEx:postIndex]...)
-			if m.LogId == nil {
-				m.LogId = []byte{}
-			}
+			m.LogId = string(dAtA[iNdEx:postIndex])
 			iNdEx = postIndex
 		case 2:
 			if wireType != 2 {

consensus/consensusproto/consensuserr/errors.go

@@ -9,8 +9,10 @@ import (
 var (
 	errGroup = rpcerr.ErrGroup(consensusproto.ErrCodes_ErrorOffset)
 
-	ErrUnexpected  = errGroup.Register(fmt.Errorf("unexpected consensus error"), uint64(consensusproto.ErrCodes_Unexpected))
-	ErrConflict    = errGroup.Register(fmt.Errorf("records conflict"), uint64(consensusproto.ErrCodes_RecordConflict))
-	ErrLogExists   = errGroup.Register(fmt.Errorf("log exists"), uint64(consensusproto.ErrCodes_LogExists))
-	ErrLogNotFound = errGroup.Register(fmt.Errorf("log not found"), uint64(consensusproto.ErrCodes_LogNotFound))
+	ErrUnexpected     = errGroup.Register(fmt.Errorf("unexpected consensus error"), uint64(consensusproto.ErrCodes_Unexpected))
+	ErrConflict       = errGroup.Register(fmt.Errorf("records conflict"), uint64(consensusproto.ErrCodes_RecordConflict))
+	ErrLogExists      = errGroup.Register(fmt.Errorf("log exists"), uint64(consensusproto.ErrCodes_LogExists))
+	ErrLogNotFound    = errGroup.Register(fmt.Errorf("log not found"), uint64(consensusproto.ErrCodes_LogNotFound))
+	ErrForbidden      = errGroup.Register(fmt.Errorf("forbidden"), uint64(consensusproto.ErrCodes_Forbidden))
+	ErrInvalidPayload = errGroup.Register(fmt.Errorf("invalid payload"), uint64(consensusproto.ErrCodes_InvalidPayload))
 )

consensus/consensusproto/protos/consensus.proto

@@ -8,13 +8,14 @@ enum ErrCodes {
     LogExists = 1;
     LogNotFound = 2;
     RecordConflict = 3;
+    Forbidden = 4;
+    InvalidPayload = 5;
     ErrorOffset = 500;
 }
 
 
 message Log {
-    bytes id = 1;
-    bytes payload = 2;
+    string id = 1;
     repeated RawRecordWithId records = 3;
 }
 
@@ -53,21 +54,22 @@ service Consensus {
 message Ok {}
 
 message LogAddRequest {
-    Log log = 1;
+    // first record in the log, consensus node not sign it
+    RawRecordWithId record = 1;
 }
 
 message RecordAddRequest {
-    bytes logId = 1;
+    string logId = 1;
     RawRecord record = 2;
 }
 
 message LogWatchRequest {
-    repeated bytes watchIds = 1;
-    repeated bytes unwatchIds = 2;
+    repeated string watchIds = 1;
+    repeated string unwatchIds = 2;
 }
 
 message LogWatchEvent {
-    bytes logId = 1;
+    string logId = 1;
     repeated RawRecordWithId records = 2;
     Err error = 3;
 }

coordinator/coordinatorclient/mock_coordinatorclient/mock_coordinatorclient.go

@@ -12,7 +12,7 @@ import (
 	treechangeproto "github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	coordinatorclient "github.com/anyproto/any-sync/coordinator/coordinatorclient"
 	coordinatorproto "github.com/anyproto/any-sync/coordinator/coordinatorproto"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockCoordinatorClient is a mock of CoordinatorClient interface.

go.mod

@@ -12,7 +12,6 @@ require (
 	github.com/gobwas/glob v0.2.3
 	github.com/goccy/go-graphviz v0.1.1
 	github.com/gogo/protobuf v1.3.2
-	github.com/golang/mock v1.6.0
 	github.com/google/uuid v1.3.0
 	github.com/hashicorp/yamux v0.1.1
 	github.com/huandu/skiplist v1.2.0
@@ -34,10 +33,11 @@ require (
 	github.com/tyler-smith/go-bip39 v1.1.0
 	github.com/zeebo/blake3 v0.2.3
 	go.uber.org/atomic v1.11.0
+	go.uber.org/mock v0.2.0
 	go.uber.org/zap v1.24.0
-	golang.org/x/crypto v0.11.0
+	golang.org/x/crypto v0.10.0
 	golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1
-	golang.org/x/net v0.12.0
+	golang.org/x/net v0.11.0
 	gopkg.in/yaml.v3 v3.0.1
 	storj.io/drpc v0.0.33
 )
@@ -99,7 +99,7 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/image v0.6.0 // indirect
 	golang.org/x/sync v0.2.0 // indirect
-	golang.org/x/sys v0.10.0 // indirect
+	golang.org/x/sys v0.9.0 // indirect
 	golang.org/x/tools v0.9.3 // indirect
 	google.golang.org/protobuf v1.30.0 // indirect
 	lukechampine.com/blake3 v1.2.1 // indirect

go.sum

@@ -54,7 +54,6 @@ github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69
 github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0 h1:DACJavvAHhabrF08vX0COfcOBJRhZ8lUbR+ZWIs0Y5g=
 github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0/go.mod h1:E/TSTwGwJL78qG/PmXZO1EjYhfJinVAhrmmHX6Z8B9k=
 github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
-github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
@@ -281,6 +280,8 @@ go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
 go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.uber.org/goleak v1.1.11-0.20210813005559-691160354723/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
 go.uber.org/goleak v1.1.12 h1:gZAh5/EyT/HQwlpkCy6wTpqfH9H8Lz8zbm3dZh+OyzA=
+go.uber.org/mock v0.2.0 h1:TaP3xedm7JaAgScZO7tlvlKrqT0p7I6OsdGB5YNSMDU=
+go.uber.org/mock v0.2.0/go.mod h1:J0y0rp9L3xiff1+ZBfKxlC1fz2+aO16tw0tsDOixfuM=
 go.uber.org/multierr v1.5.0/go.mod h1:FeouvMocqHpRaaGuG9EjoKcStLC43Zu/fmqdUMPcKYU=
 go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
@@ -298,8 +299,8 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200602180216-279210d13fed/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/crypto v0.11.0 h1:6Ewdq3tDic1mg5xRO4milcWCfMVQhI4NkqWWvqejpuA=
-golang.org/x/crypto v0.11.0/go.mod h1:xgJhtzW8F9jGdVFWZESrid1U1bjeNy4zgy5cRr/CIio=
+golang.org/x/crypto v0.10.0 h1:LKqV2xt9+kDzSTfOhx4FrkEBcMrAgHSYgzywV9zcGmM=
+golang.org/x/crypto v0.10.0/go.mod h1:o4eNf7Ede1fv+hwOwZsTHl9EsPFO6q6ZvYR8vYfY45I=
 golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1 h1:k/i9J1pBpvlfR+9QsetwPyERsqu1GIbi967PQMq3Ivc=
 golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1/go.mod h1:V1LtkGg67GoY2N1AnLN78QLrzxkLyJw7RJb1gzOOz9w=
 golang.org/x/image v0.6.0 h1:bR8b5okrPI3g/gyZakLZHeWxAR8Dn5CyxXv1hLH5g/4=
@@ -321,8 +322,8 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
-golang.org/x/net v0.12.0 h1:cfawfvKITfUsFCeJIHJrbSxpeu/E81khclypR0GVT50=
-golang.org/x/net v0.12.0/go.mod h1:zEVYFnQC7m/vmpQFELhcD1EWkZlX69l4oqgmer6hfKA=
+golang.org/x/net v0.11.0 h1:Gi2tvZIJyBtO9SDr1q9h5hEQCp/4L2RQ+ar0qjx2oNU=
+golang.org/x/net v0.11.0/go.mod h1:2L/ixqYpgIVXmeoSA/4Lu7BzTG4KIyPIryS4IsOd1oQ=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -346,8 +347,8 @@ golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.10.0 h1:SqMFp9UcQJZa+pmYuAKjd9xq1f0j5rLcDIk0mj4qAsA=
-golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.9.0 h1:KS/R3tvhPqvJvwcKfnBHJwwthS11LRhmM5D59eEXa0s=
+golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
@@ -356,7 +357,7 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
-golang.org/x/text v0.11.0 h1:LAntKIrcmeSKERyiOh0XMV39LXS8IE9UL2yP7+f5ij4=
+golang.org/x/text v0.10.0 h1:UpjohKhiEgNc0CSauXmwYftY1+LlaC75SJwh0SgCX58=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
@@ -365,7 +366,6 @@ golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
-golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=

net/peer/mock_peer/mock_peer.go

@@ -9,7 +9,7 @@ import (
 	reflect "reflect"
 	time "time"
 
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 	drpc "storj.io/drpc"
 )
 

net/peer/peer.go

@@ -106,10 +106,10 @@ func (p *peer) AcquireDrpcConn(ctx context.Context) (drpc.Conn, error) {
 	p.mu.Lock()
 	if len(p.inactive) == 0 {
 		wait := p.limiter.wait(len(p.active) + int(p.openingWaitCount.Load()))
-		p.openingWaitCount.Add(1)
-		defer p.openingWaitCount.Add(-1)
 		p.mu.Unlock()
 		if wait != nil {
+			p.openingWaitCount.Add(1)
+			defer p.openingWaitCount.Add(-1)
 			// throttle new connection opening
 			select {
 			case <-ctx.Done():

net/peer/peer_test.go

@@ -6,9 +6,9 @@ import (
 	"github.com/anyproto/any-sync/net/secureservice/handshake"
 	"github.com/anyproto/any-sync/net/secureservice/handshake/handshakeproto"
 	"github.com/anyproto/any-sync/net/transport/mock_transport"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"io"
 	"net"
 	_ "net/http/pprof"

net/pool/mock_pool/mock_pool.go

@@ -9,7 +9,7 @@ import (
 	reflect "reflect"
 
 	peer "github.com/anyproto/any-sync/net/peer"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockPool is a mock of Pool interface.

net/secureservice/secureservice_test.go

@@ -9,9 +9,9 @@ import (
 	"github.com/anyproto/any-sync/nodeconf"
 	"github.com/anyproto/any-sync/nodeconf/mock_nodeconf"
 	"github.com/anyproto/any-sync/testutil/testnodeconf"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"net"
 	"testing"
 )

net/transport/mock_transport/mock_transport.go

@@ -10,7 +10,7 @@ import (
 	reflect "reflect"
 
 	transport "github.com/anyproto/any-sync/net/transport"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockTransport is a mock of Transport interface.

net/transport/yamux/yamux_test.go

@@ -10,9 +10,9 @@ import (
 	"github.com/anyproto/any-sync/nodeconf/mock_nodeconf"
 	"github.com/anyproto/any-sync/testutil/accounttest"
 	"github.com/anyproto/any-sync/testutil/testnodeconf"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"io"
 	"net"
 	"sync"

nodeconf/mock_nodeconf/mock_nodeconf.go

@@ -11,7 +11,7 @@ import (
 	app "github.com/anyproto/any-sync/app"
 	nodeconf "github.com/anyproto/any-sync/nodeconf"
 	chash "github.com/anyproto/go-chash"
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockService is a mock of Service interface.

util/periodicsync/mock_periodicsync/mock_periodicsync.go

@@ -7,7 +7,7 @@ package mock_periodicsync
 import (
 	reflect "reflect"
 
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
 // MockPeriodicSync is a mock of PeriodicSync interface.

util/periodicsync/periodicsync_test.go

@@ -3,8 +3,8 @@ package periodicsync
 import (
 	"context"
 	"github.com/anyproto/any-sync/app/logger"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"testing"
 	"time"
 )