Merge branch 'drpc-services' into space-sync

# Conflicts:
#	Makefile
#	cmd/node/node.go
#	pkg/acl/aclchanges/aclpb/aclchanges.pb.go
#	pkg/acl/acltree/aclstatebuilder.go
#	pkg/acl/acltree/change.go
#	pkg/acl/example/plaintextdocument/document.go
#	pkg/acl/example/plaintextdocument/plaintextdocstate.go
#	pkg/acl/list/aclstate.go
#	pkg/acl/list/changebuilder.go
#	pkg/acl/testutils/testchanges/proto/test.pb.go
#	pkg/acl/testutils/treestoragebuilder/treestoragebuilder.go
#	pkg/acl/testutils/treestoragebuilder/treestoragebuildergraph_nix.go
#	pkg/acl/tree/treegraph_nix.go
#	service/document/service.go
#	service/space/service.go
#	service/space/space.go
#	service/sync/requesthandler/requesthandler.go
#	service/treecache/service.go
#	syncproto/helpers.go
#	syncproto/sync.pb.go

app/app.go

@@ -27,7 +27,7 @@ var (
 type Component interface {
 	// Init will be called first
 	// When returned error is not nil - app start will be aborted
-	Init(ctx context.Context, a *App) (err error)
+	Init(a *App) (err error)
 	// Name must return unique service name
 	Name() (name string)
 }
@@ -157,7 +157,7 @@ func (app *App) Start(ctx context.Context) (err error) {
 	}
 
 	for i, s := range app.components {
-		if err = s.Init(ctx, app); err != nil {
+		if err = s.Init(app); err != nil {
 			closeServices(i)
 			return fmt.Errorf("can't init service '%s': %v", s.Name(), err)
 		}

app/app_test.go

@@ -131,7 +131,7 @@ type testComponent struct {
 	ids  testIds
 }
 
-func (t *testComponent) Init(ctx context.Context, a *App) error {
+func (t *testComponent) Init(a *App) error {
 	t.ids.initId = t.seq.New()
 	return t.err
 }

cmd/node/node.go

@@ -6,20 +6,16 @@ import (
 	"fmt"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/account"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/dialer"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/pool"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/rpc/server"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/secure"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/nodeconf"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/config"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/account"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/node/nodespace"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/api"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/configuration"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/document"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/dialer"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/pool"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/rpc/server"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/secure"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/node"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/storage"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/sync/message"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/sync/requesthandler"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/treecache"
 	"go.uber.org/zap"
 	"net/http"
 	_ "net/http/pprof"
@@ -95,15 +91,17 @@ func main() {
 func Bootstrap(a *app.App) {
 	a.Register(account.New()).
 		Register(node.New()).
+		Register(nodeconf.New()).
 		Register(secure.New()).
-		Register(server.New()).
 		Register(dialer.New()).
-		Register(pool.NewPool()).
+		Register(pool.New()).
-		Register(storage.New()).
+		Register(nodespace.New()).
-		Register(configuration.New()).
+		Register(commonspace.New()).
-		Register(document.New()).
+		Register(server.New())
-		Register(message.New()).
+
-		Register(requesthandler.New()).
+	//Register(document.New()).
-		Register(treecache.New()).
+	//Register(message.New()).
-		Register(api.New())
+	//Register(requesthandler.New()).
+	//Register(treecache.New()).
+	//Register(api.New())
 }

common/account/service.go

@@ -1,7 +1,6 @@
 package account
 
 import (
-	"context"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/config"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/account"
@@ -33,7 +32,7 @@ func New() app.Component {
 	return &service{}
 }
 
-func (s *service) Init(ctx context.Context, a *app.App) (err error) {
+func (s *service) Init(a *app.App) (err error) {
 	cfg := a.MustComponent(config.CName).(*config.Config)
 
 	// decoding our keys

common/commonspace/periodicsync.go

@@ -0,0 +1,58 @@
+package commonspace
+
+import (
+	"context"
+	"go.uber.org/zap"
+	"time"
+)
+
+func newPeriodicSync(periodSeconds int, sync func(ctx context.Context) error, l *zap.Logger) *periodicSync {
+	ctx, cancel := context.WithCancel(context.Background())
+	ps := &periodicSync{
+		log:          l,
+		sync:         sync,
+		syncCtx:      ctx,
+		syncCancel:   cancel,
+		syncLoopDone: make(chan struct{}),
+	}
+	go ps.syncLoop(periodSeconds)
+	return ps
+}
+
+type periodicSync struct {
+	log          *zap.Logger
+	sync         func(ctx context.Context) error
+	syncCtx      context.Context
+	syncCancel   context.CancelFunc
+	syncLoopDone chan struct{}
+}
+
+func (p *periodicSync) syncLoop(periodSeconds int) {
+	period := time.Duration(periodSeconds) * time.Second
+	defer close(p.syncLoopDone)
+	doSync := func() {
+		ctx, cancel := context.WithTimeout(p.syncCtx, time.Minute)
+		defer cancel()
+		if err := p.sync(ctx); err != nil {
+			p.log.Warn("periodic sync error", zap.Error(err))
+		}
+	}
+	doSync()
+	if period > 0 {
+		ticker := time.NewTicker(period)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-p.syncCtx.Done():
+				return
+			case <-ticker.C:
+				doSync()
+			}
+		}
+	}
+}
+
+func (p *periodicSync) Close() {
+	p.syncCancel()
+	<-p.syncLoopDone
+}

common/commonspace/remotediff/remotediff.go

@@ -0,0 +1,98 @@
+package remotediff
+
+import (
+	"context"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace/spacesyncproto"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/ldiff"
+)
+
+type Client interface {
+	HeadSync(ctx context.Context, in *spacesyncproto.HeadSyncRequest) (*spacesyncproto.HeadSyncResponse, error)
+}
+
+func NewRemoteDiff(spaceId string, client Client) ldiff.Remote {
+	return remote{
+		spaceId: spaceId,
+		client:  client,
+	}
+}
+
+type remote struct {
+	spaceId string
+	client  Client
+}
+
+func (r remote) Ranges(ctx context.Context, ranges []ldiff.Range, resBuf []ldiff.RangeResult) (results []ldiff.RangeResult, err error) {
+	results = resBuf[:0]
+	pbRanges := make([]*spacesyncproto.HeadSyncRange, 0, len(ranges))
+	for _, rg := range ranges {
+		pbRanges = append(pbRanges, &spacesyncproto.HeadSyncRange{
+			From:  rg.From,
+			To:    rg.To,
+			Limit: uint32(rg.Limit),
+		})
+	}
+	req := &spacesyncproto.HeadSyncRequest{
+		SpaceId: r.spaceId,
+		Ranges:  pbRanges,
+	}
+	resp, err := r.client.HeadSync(ctx, req)
+	if err != nil {
+		return
+	}
+	for _, rr := range resp.Results {
+		var elms []ldiff.Element
+		if len(rr.Elements) > 0 {
+			elms = make([]ldiff.Element, 0, len(rr.Elements))
+		}
+		for _, e := range rr.Elements {
+			elms = append(elms, ldiff.Element{
+				Id:   e.Id,
+				Head: e.Head,
+			})
+		}
+		results = append(results, ldiff.RangeResult{
+			Hash:     rr.Hash,
+			Elements: elms,
+			Count:    int(rr.Count),
+		})
+	}
+	return
+}
+
+func HandlerRangeRequest(ctx context.Context, d ldiff.Diff, req *spacesyncproto.HeadSyncRequest) (resp *spacesyncproto.HeadSyncResponse, err error) {
+	ranges := make([]ldiff.Range, 0, len(req.Ranges))
+	for _, reqRange := range req.Ranges {
+		ranges = append(ranges, ldiff.Range{
+			From:  reqRange.From,
+			To:    reqRange.To,
+			Limit: int(reqRange.Limit),
+		})
+	}
+	res, err := d.Ranges(ctx, ranges, nil)
+	if err != nil {
+		return
+	}
+
+	resp = &spacesyncproto.HeadSyncResponse{
+		Results: make([]*spacesyncproto.HeadSyncResult, 0, len(res)),
+	}
+	for _, rangeRes := range res {
+		var elements []*spacesyncproto.HeadSyncResultElement
+		if len(rangeRes.Elements) > 0 {
+			elements = make([]*spacesyncproto.HeadSyncResultElement, 0, len(rangeRes.Elements))
+			for _, el := range rangeRes.Elements {
+				elements = append(elements, &spacesyncproto.HeadSyncResultElement{
+					Id:   el.Id,
+					Head: el.Head,
+				})
+			}
+		}
+		resp.Results = append(resp.Results, &spacesyncproto.HeadSyncResult{
+			Hash:     rangeRes.Hash,
+			Elements: elements,
+			Count:    uint32(rangeRes.Count),
+		})
+	}
+	return
+}

common/commonspace/remotediff/remotediff_test.go

@@ -0,0 +1,41 @@
+package remotediff
+
+import (
+	"context"
+	"fmt"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace/spacesyncproto"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/ldiff"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestRemote(t *testing.T) {
+	ldLocal := ldiff.New(8, 8)
+	ldRemote := ldiff.New(8, 8)
+	for i := 0; i < 100; i++ {
+		el := ldiff.Element{
+			Id:   fmt.Sprint(i),
+			Head: fmt.Sprint(i),
+		}
+		ldRemote.Set(el)
+		if i%10 != 0 {
+			ldLocal.Set(el)
+		}
+	}
+
+	rd := NewRemoteDiff("1", &mockClient{l: ldRemote})
+	newIds, changedIds, removedIds, err := ldLocal.Diff(context.Background(), rd)
+	require.NoError(t, err)
+	assert.Len(t, newIds, 10)
+	assert.Len(t, changedIds, 0)
+	assert.Len(t, removedIds, 0)
+}
+
+type mockClient struct {
+	l ldiff.Diff
+}
+
+func (m *mockClient) HeadSync(ctx context.Context, in *spacesyncproto.HeadSyncRequest) (*spacesyncproto.HeadSyncResponse, error) {
+	return HandlerRangeRequest(ctx, m.l, in)
+}

common/commonspace/rpchandler.go

@@ -0,0 +1,25 @@
+package commonspace
+
+import (
+	"context"
+	"fmt"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace/remotediff"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace/spacesyncproto"
+)
+
+type RpcHandler interface {
+	HeadSync(ctx context.Context, req *spacesyncproto.HeadSyncRequest) (*spacesyncproto.HeadSyncResponse, error)
+	Stream(stream spacesyncproto.DRPCSpace_StreamStream) error
+}
+
+type rpcHandler struct {
+	s *space
+}
+
+func (r *rpcHandler) HeadSync(ctx context.Context, req *spacesyncproto.HeadSyncRequest) (*spacesyncproto.HeadSyncResponse, error) {
+	return remotediff.HandlerRangeRequest(ctx, r.s.diff, req)
+}
+
+func (r *rpcHandler) Stream(stream spacesyncproto.DRPCSpace_StreamStream) error {
+	return fmt.Errorf("not implemented")
+}

common/commonspace/service.go

@@ -0,0 +1,49 @@
+package commonspace
+
+import (
+	"context"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/nodeconf"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/config"
+)
+
+const CName = "common.commonspace"
+
+var log = logger.NewNamed(CName)
+
+func New() Service {
+	return &service{}
+}
+
+type Service interface {
+	CreateSpace(ctx context.Context, id string) (sp Space, err error)
+	app.Component
+}
+
+type service struct {
+	config               config.Space
+	configurationService nodeconf.Service
+}
+
+func (s *service) Init(a *app.App) (err error) {
+	s.config = a.MustComponent(config.CName).(*config.Config).Space
+	s.configurationService = a.MustComponent(nodeconf.CName).(nodeconf.Service)
+	return nil
+}
+
+func (s *service) Name() (name string) {
+	return CName
+}
+
+func (s *service) CreateSpace(ctx context.Context, id string) (Space, error) {
+	sp := &space{
+		id:    id,
+		nconf: s.configurationService.GetLast(),
+		conf:  s.config,
+	}
+	if err := sp.Init(ctx); err != nil {
+		return nil, err
+	}
+	return sp, nil
+}

common/commonspace/space.go

@@ -0,0 +1,115 @@
+package commonspace
+
+import (
+	"context"
+	"fmt"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace/remotediff"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace/spacesyncproto"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/peer"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/nodeconf"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/config"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/ldiff"
+	"go.uber.org/zap"
+	"math/rand"
+	"sync"
+	"time"
+)
+
+type Space interface {
+	Id() string
+
+	SpaceSyncRpc() RpcHandler
+
+	Close() error
+}
+
+type space struct {
+	id           string
+	nconf        nodeconf.Configuration
+	conf         config.Space
+	diff         ldiff.Diff
+	rpc          *rpcHandler
+	periodicSync *periodicSync
+	mu           sync.RWMutex
+}
+
+func (s *space) Id() string {
+	return s.id
+}
+
+func (s *space) Init(ctx context.Context) error {
+	s.diff = ldiff.New(16, 16)
+	s.periodicSync = newPeriodicSync(s.conf.SyncPeriod, s.sync, log.With(zap.String("spaceId", s.id)))
+	s.rpc = &rpcHandler{s: s}
+	s.testFill()
+	return nil
+}
+
+func (s *space) SpaceSyncRpc() RpcHandler {
+	return s.rpc
+}
+
+func (s *space) testFill() {
+	var n = 1000
+	var els = make([]ldiff.Element, 0, n)
+	rand.Seed(time.Now().UnixNano())
+	for i := 0; i < n; i++ {
+		if rand.Intn(n) > 2 {
+			id := fmt.Sprintf("%s.%d", s.id, i)
+			head := "head." + id
+			if rand.Intn(n) > n-10 {
+				head += ".modified"
+			}
+			el := ldiff.Element{
+				Id:   id,
+				Head: head,
+			}
+			els = append(els, el)
+		}
+	}
+	s.diff.Set(els...)
+}
+
+func (s *space) sync(ctx context.Context) error {
+	st := time.Now()
+	peers, err := s.getPeers(ctx)
+	if err != nil {
+		return err
+	}
+	for _, p := range peers {
+		if err := s.syncWithPeer(ctx, p); err != nil {
+			log.Error("can't sync with peer", zap.String("peer", p.Id()), zap.Error(err))
+		}
+	}
+	log.Info("synced", zap.String("spaceId", s.id), zap.Duration("dur", time.Since(st)))
+	return nil
+}
+
+func (s *space) syncWithPeer(ctx context.Context, p peer.Peer) (err error) {
+	cl := spacesyncproto.NewDRPCSpaceClient(p)
+	rdiff := remotediff.NewRemoteDiff(s.id, cl)
+	newIds, changedIds, removedIds, err := s.diff.Diff(ctx, rdiff)
+	if err != nil {
+		return nil
+	}
+	log.Info("sync done:", zap.Int("newIds", len(newIds)), zap.Int("changedIds", len(changedIds)), zap.Int("removedIds", len(removedIds)))
+	return
+}
+
+func (s *space) getPeers(ctx context.Context) (peers []peer.Peer, err error) {
+	if s.nconf.IsResponsible(s.id) {
+		return s.nconf.AllPeers(ctx, s.id)
+	} else {
+		var p peer.Peer
+		p, err = s.nconf.OnePeer(ctx, s.id)
+		if err != nil {
+			return nil, err
+		}
+		return []peer.Peer{p}, nil
+	}
+}
+
+func (s *space) Close() error {
+	s.periodicSync.Close()
+	return nil
+}

common/commonspace/spacesyncproto/protos/spacesync.proto

@@ -0,0 +1,52 @@
+syntax = "proto3";
+package anySpace;
+option go_package = "common/commonspace/spacesyncproto";
+
+enum ErrCodes {
+    Unexpected = 0;
+}
+
+
+service Space {
+    // HeadSync compares all objects and their hashes in a space
+    rpc HeadSync(HeadSyncRequest) returns (HeadSyncResponse);
+    rpc Stream( stream Msg) returns (stream Msg);
+}
+
+// TODO: temporary mock message
+message Msg {
+    string spaceId = 1;
+}
+
+
+
+// HeadSyncRange presenting a request for one range
+message HeadSyncRange {
+    uint64 from = 1;
+    uint64 to = 2;
+    uint32 limit = 3;
+}
+
+// HeadSyncResult presenting a response for one range
+message HeadSyncResult {
+    bytes hash = 1;
+    repeated HeadSyncResultElement elements = 2;
+    uint32 count = 3;
+}
+
+// HeadSyncResultElement presenting state of one object
+message HeadSyncResultElement {
+    string id = 1;
+    string head = 2;
+}
+
+// HeadSyncRequest is a request for HeadSync
+message HeadSyncRequest {
+    string spaceId = 1;
+    repeated HeadSyncRange ranges = 2;
+}
+
+// HeadSyncResponse is a response for HeadSync
+message HeadSyncResponse {
+    repeated HeadSyncResult results = 1;
+}

common/commonspace/spacesyncproto/spacesync_drpc.pb.go

@@ -0,0 +1,188 @@
+// Code generated by protoc-gen-go-drpc. DO NOT EDIT.
+// protoc-gen-go-drpc version: v0.0.32
+// source: common/commonspace/spacesyncproto/protos/spacesync.proto
+
+package spacesyncproto
+
+import (
+	bytes "bytes"
+	context "context"
+	errors "errors"
+	jsonpb "github.com/gogo/protobuf/jsonpb"
+	proto "github.com/gogo/protobuf/proto"
+	drpc "storj.io/drpc"
+	drpcerr "storj.io/drpc/drpcerr"
+)
+
+type drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto struct{}
+
+func (drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto) Marshal(msg drpc.Message) ([]byte, error) {
+	return proto.Marshal(msg.(proto.Message))
+}
+
+func (drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto) Unmarshal(buf []byte, msg drpc.Message) error {
+	return proto.Unmarshal(buf, msg.(proto.Message))
+}
+
+func (drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto) JSONMarshal(msg drpc.Message) ([]byte, error) {
+	var buf bytes.Buffer
+	err := new(jsonpb.Marshaler).Marshal(&buf, msg.(proto.Message))
+	if err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+func (drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto) JSONUnmarshal(buf []byte, msg drpc.Message) error {
+	return jsonpb.Unmarshal(bytes.NewReader(buf), msg.(proto.Message))
+}
+
+type DRPCSpaceClient interface {
+	DRPCConn() drpc.Conn
+
+	HeadSync(ctx context.Context, in *HeadSyncRequest) (*HeadSyncResponse, error)
+	Stream(ctx context.Context) (DRPCSpace_StreamClient, error)
+}
+
+type drpcSpaceClient struct {
+	cc drpc.Conn
+}
+
+func NewDRPCSpaceClient(cc drpc.Conn) DRPCSpaceClient {
+	return &drpcSpaceClient{cc}
+}
+
+func (c *drpcSpaceClient) DRPCConn() drpc.Conn { return c.cc }
+
+func (c *drpcSpaceClient) HeadSync(ctx context.Context, in *HeadSyncRequest) (*HeadSyncResponse, error) {
+	out := new(HeadSyncResponse)
+	err := c.cc.Invoke(ctx, "/anySpace.Space/HeadSync", drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto{}, in, out)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *drpcSpaceClient) Stream(ctx context.Context) (DRPCSpace_StreamClient, error) {
+	stream, err := c.cc.NewStream(ctx, "/anySpace.Space/Stream", drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto{})
+	if err != nil {
+		return nil, err
+	}
+	x := &drpcSpace_StreamClient{stream}
+	return x, nil
+}
+
+type DRPCSpace_StreamClient interface {
+	drpc.Stream
+	Send(*Msg) error
+	Recv() (*Msg, error)
+}
+
+type drpcSpace_StreamClient struct {
+	drpc.Stream
+}
+
+func (x *drpcSpace_StreamClient) Send(m *Msg) error {
+	return x.MsgSend(m, drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto{})
+}
+
+func (x *drpcSpace_StreamClient) Recv() (*Msg, error) {
+	m := new(Msg)
+	if err := x.MsgRecv(m, drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto{}); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func (x *drpcSpace_StreamClient) RecvMsg(m *Msg) error {
+	return x.MsgRecv(m, drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto{})
+}
+
+type DRPCSpaceServer interface {
+	HeadSync(context.Context, *HeadSyncRequest) (*HeadSyncResponse, error)
+	Stream(DRPCSpace_StreamStream) error
+}
+
+type DRPCSpaceUnimplementedServer struct{}
+
+func (s *DRPCSpaceUnimplementedServer) HeadSync(context.Context, *HeadSyncRequest) (*HeadSyncResponse, error) {
+	return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
+}
+
+func (s *DRPCSpaceUnimplementedServer) Stream(DRPCSpace_StreamStream) error {
+	return drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
+}
+
+type DRPCSpaceDescription struct{}
+
+func (DRPCSpaceDescription) NumMethods() int { return 2 }
+
+func (DRPCSpaceDescription) Method(n int) (string, drpc.Encoding, drpc.Receiver, interface{}, bool) {
+	switch n {
+	case 0:
+		return "/anySpace.Space/HeadSync", drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto{},
+			func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
+				return srv.(DRPCSpaceServer).
+					HeadSync(
+						ctx,
+						in1.(*HeadSyncRequest),
+					)
+			}, DRPCSpaceServer.HeadSync, true
+	case 1:
+		return "/anySpace.Space/Stream", drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto{},
+			func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
+				return nil, srv.(DRPCSpaceServer).
+					Stream(
+						&drpcSpace_StreamStream{in1.(drpc.Stream)},
+					)
+			}, DRPCSpaceServer.Stream, true
+	default:
+		return "", nil, nil, nil, false
+	}
+}
+
+func DRPCRegisterSpace(mux drpc.Mux, impl DRPCSpaceServer) error {
+	return mux.Register(impl, DRPCSpaceDescription{})
+}
+
+type DRPCSpace_HeadSyncStream interface {
+	drpc.Stream
+	SendAndClose(*HeadSyncResponse) error
+}
+
+type drpcSpace_HeadSyncStream struct {
+	drpc.Stream
+}
+
+func (x *drpcSpace_HeadSyncStream) SendAndClose(m *HeadSyncResponse) error {
+	if err := x.MsgSend(m, drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto{}); err != nil {
+		return err
+	}
+	return x.CloseSend()
+}
+
+type DRPCSpace_StreamStream interface {
+	drpc.Stream
+	Send(*Msg) error
+	Recv() (*Msg, error)
+}
+
+type drpcSpace_StreamStream struct {
+	drpc.Stream
+}
+
+func (x *drpcSpace_StreamStream) Send(m *Msg) error {
+	return x.MsgSend(m, drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto{})
+}
+
+func (x *drpcSpace_StreamStream) Recv() (*Msg, error) {
+	m := new(Msg)
+	if err := x.MsgRecv(m, drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto{}); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func (x *drpcSpace_StreamStream) RecvMsg(m *Msg) error {
+	return x.MsgRecv(m, drpcEncoding_File_common_commonspace_spacesyncproto_protos_spacesync_proto{})
+}

common/net/dialer/dialer.go

@@ -5,10 +5,9 @@ import (
 	"errors"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/peer"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/secure"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/config"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/peer"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/rpc"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/secure"
 	"github.com/libp2p/go-libp2p-core/sec"
 	"go.uber.org/zap"
 	"net"
@@ -40,7 +39,7 @@ type dialer struct {
 	mu sync.RWMutex
 }
 
-func (d *dialer) Init(ctx context.Context, a *app.App) (err error) {
+func (d *dialer) Init(a *app.App) (err error) {
 	d.transport = a.MustComponent(secure.CName).(secure.Service)
 	nodes := a.MustComponent(config.CName).(*config.Config).Nodes
 	d.peerAddrs = map[string][]string{}
@@ -60,7 +59,7 @@ func (d *dialer) UpdateAddrs(addrs map[string][]string) {
 	d.mu.Unlock()
 }
 
-func (d *dialer) Dial(ctx context.Context, peerId string) (peer peer.Peer, err error) {
+func (d *dialer) Dial(ctx context.Context, peerId string) (p peer.Peer, err error) {
 	d.mu.RLock()
 	defer d.mu.RUnlock()
 	addrs, ok := d.peerAddrs[peerId]
@@ -68,11 +67,11 @@ func (d *dialer) Dial(ctx context.Context, peerId string) (peer peer.Peer, err e
 		return nil, ErrArrdsNotFound
 	}
 	var (
-		stream drpc.Stream
+		conn drpc.Conn
-		sc     sec.SecureConn
+		sc   sec.SecureConn
 	)
 	for _, addr := range addrs {
-		stream, sc, err = d.makeStream(ctx, addr)
+		conn, sc, err = d.handshake(ctx, addr)
 		if err != nil {
 			log.Info("can't connect to host", zap.String("addr", addr))
 		} else {
@@ -83,10 +82,10 @@ func (d *dialer) Dial(ctx context.Context, peerId string) (peer peer.Peer, err e
 	if err != nil {
 		return
 	}
-	return rpc.PeerFromStream(sc, stream, false), nil
+	return peer.NewPeer(sc, conn), nil
 }
 
-func (d *dialer) makeStream(ctx context.Context, addr string) (stream drpc.Stream, sc sec.SecureConn, err error) {
+func (d *dialer) handshake(ctx context.Context, addr string) (conn drpc.Conn, sc sec.SecureConn, err error) {
 	tcpConn, err := net.Dial("tcp", addr)
 	if err != nil {
 		return
@@ -96,9 +95,6 @@ func (d *dialer) makeStream(ctx context.Context, addr string) (stream drpc.Strea
 		return
 	}
 	log.Info("connected with remote host", zap.String("serverPeer", sc.RemotePeer().String()), zap.String("per", sc.LocalPeer().String()))
-	stream, err = drpcconn.New(sc).NewStream(ctx, "", rpc.Encoding)
+	conn = drpcconn.New(sc)
-	if err != nil {
+	return conn, sc, err
-		return
-	}
-	return stream, sc, err
 }

common/net/peer/peer.go

@@ -0,0 +1,59 @@
+package peer
+
+import (
+	"context"
+	"github.com/libp2p/go-libp2p-core/sec"
+	"storj.io/drpc"
+	"sync/atomic"
+	"time"
+)
+
+func NewPeer(sc sec.SecureConn, conn drpc.Conn) Peer {
+	return &peer{
+		id:        sc.RemotePeer().String(),
+		lastUsage: time.Now().Unix(),
+		sc:        sc,
+		Conn:      conn,
+	}
+}
+
+type Peer interface {
+	Id() string
+	LastUsage() time.Time
+	UpdateLastUsage()
+	drpc.Conn
+}
+
+type peer struct {
+	id        string
+	lastUsage int64
+	sc        sec.SecureConn
+	drpc.Conn
+}
+
+func (p *peer) Id() string {
+	return p.id
+}
+
+func (p *peer) LastUsage() time.Time {
+	select {
+	case <-p.Closed():
+		return time.Unix(0, 0)
+	default:
+	}
+	return time.Unix(atomic.LoadInt64(&p.lastUsage), 0)
+}
+
+func (p *peer) Invoke(ctx context.Context, rpc string, enc drpc.Encoding, in, out drpc.Message) error {
+	defer p.UpdateLastUsage()
+	return p.Conn.Invoke(ctx, rpc, enc, in, out)
+}
+
+func (p *peer) NewStream(ctx context.Context, rpc string, enc drpc.Encoding) (drpc.Stream, error) {
+	defer p.UpdateLastUsage()
+	return p.Conn.NewStream(ctx, rpc, enc)
+}
+
+func (p *peer) UpdateLastUsage() {
+	atomic.StoreInt64(&p.lastUsage, time.Now().Unix())
+}

common/net/pool/pool.go

@@ -0,0 +1,107 @@
+package pool
+
+import (
+	"context"
+	"errors"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/dialer"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/peer"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/ocache"
+	"math/rand"
+	"time"
+)
+
+const (
+	CName = "net.pool"
+)
+
+var log = logger.NewNamed(CName)
+
+var (
+	ErrUnableToConnect = errors.New("unable to connect")
+)
+
+func New() Pool {
+	return &pool{}
+}
+
+// Pool creates and caches outgoing connection
+type Pool interface {
+	// Get lookups to peer in existing connections or creates and cache new one
+	Get(ctx context.Context, id string) (peer.Peer, error)
+	// GetOneOf searches at least one existing connection in cache or creates a new one from a randomly selected id from given list
+	GetOneOf(ctx context.Context, peerIds []string) (peer.Peer, error)
+
+	app.ComponentRunnable
+}
+
+type pool struct {
+	cache ocache.OCache
+}
+
+func (p *pool) Init(a *app.App) (err error) {
+	dialer := a.MustComponent(dialer.CName).(dialer.Dialer)
+	p.cache = ocache.New(
+		func(ctx context.Context, id string) (value ocache.Object, err error) {
+			return dialer.Dial(ctx, id)
+		},
+		ocache.WithLogger(log.Sugar()),
+		ocache.WithGCPeriod(time.Minute),
+		ocache.WithTTL(time.Minute*5),
+		ocache.WithRefCounter(false),
+	)
+	return nil
+}
+
+func (p *pool) Name() (name string) {
+	return CName
+}
+
+func (p *pool) Run(ctx context.Context) (err error) {
+	return nil
+}
+
+func (p *pool) Get(ctx context.Context, id string) (peer.Peer, error) {
+	v, err := p.cache.Get(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	pr := v.(peer.Peer)
+	select {
+	case <-pr.Closed():
+	default:
+		return pr, nil
+	}
+	p.cache.Remove(id)
+	return p.Get(ctx, id)
+}
+
+func (p *pool) GetOneOf(ctx context.Context, peerIds []string) (peer.Peer, error) {
+	// finding existing connection
+	for _, peerId := range peerIds {
+		if v, err := p.cache.Pick(ctx, peerId); err == nil {
+			pr := v.(peer.Peer)
+			select {
+			case <-pr.Closed():
+			default:
+				return pr, nil
+			}
+		}
+	}
+	// shuffle ids for better consistency
+	rand.Shuffle(len(peerIds), func(i, j int) {
+		peerIds[i], peerIds[j] = peerIds[j], peerIds[i]
+	})
+	// connecting
+	for _, peerId := range peerIds {
+		if v, err := p.cache.Get(ctx, peerId); err == nil {
+			return v.(peer.Peer), nil
+		}
+	}
+	return nil, ErrUnableToConnect
+}
+
+func (p *pool) Close(ctx context.Context) (err error) {
+	return p.cache.Close()
+}

common/net/pool/pool_test.go

@@ -0,0 +1,213 @@
+package pool
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/dialer"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/peer"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"storj.io/drpc"
+	"testing"
+	"time"
+)
+
+var ctx = context.Background()
+
+func TestPool_Get(t *testing.T) {
+	t.Run("dial error", func(t *testing.T) {
+		fx := newFixture(t)
+		defer fx.Finish()
+		var expErr = errors.New("dial error")
+		fx.Dialer.dial = func(ctx context.Context, peerId string) (peer peer.Peer, err error) {
+			return nil, expErr
+		}
+		p, err := fx.Get(ctx, "1")
+		assert.Nil(t, p)
+		assert.EqualError(t, err, expErr.Error())
+	})
+	t.Run("dial and cached", func(t *testing.T) {
+		fx := newFixture(t)
+		defer fx.Finish()
+		fx.Dialer.dial = func(ctx context.Context, peerId string) (peer peer.Peer, err error) {
+			return newTestPeer("1"), nil
+		}
+		p, err := fx.Get(ctx, "1")
+		assert.NoError(t, err)
+		assert.NotNil(t, p)
+		fx.Dialer.dial = nil
+		p, err = fx.Get(ctx, "1")
+		assert.NoError(t, err)
+		assert.NotNil(t, p)
+	})
+	t.Run("retry for closed", func(t *testing.T) {
+		fx := newFixture(t)
+		defer fx.Finish()
+		tp := newTestPeer("1")
+		fx.Dialer.dial = func(ctx context.Context, peerId string) (peer peer.Peer, err error) {
+			return tp, nil
+		}
+		p, err := fx.Get(ctx, "1")
+		assert.NoError(t, err)
+		assert.NotNil(t, p)
+		p.Close()
+		tp2 := newTestPeer("1")
+		fx.Dialer.dial = func(ctx context.Context, peerId string) (peer peer.Peer, err error) {
+			return tp2, nil
+		}
+		p, err = fx.Get(ctx, "1")
+		assert.NoError(t, err)
+		assert.Equal(t, p, tp2)
+	})
+}
+
+func TestPool_GetOneOf(t *testing.T) {
+	addToCache := func(t *testing.T, fx *fixture, tp *testPeer) {
+		fx.Dialer.dial = func(ctx context.Context, peerId string) (peer peer.Peer, err error) {
+			return tp, nil
+		}
+		gp, err := fx.Get(ctx, tp.Id())
+		require.NoError(t, err)
+		require.Equal(t, gp, tp)
+	}
+
+	t.Run("from cache", func(t *testing.T) {
+		fx := newFixture(t)
+		defer fx.Finish()
+		tp1 := newTestPeer("1")
+		addToCache(t, fx, tp1)
+		p, err := fx.GetOneOf(ctx, []string{"3", "2", "1"})
+		require.NoError(t, err)
+		assert.Equal(t, tp1, p)
+	})
+	t.Run("from cache - skip closed", func(t *testing.T) {
+		fx := newFixture(t)
+		defer fx.Finish()
+		tp2 := newTestPeer("2")
+		addToCache(t, fx, tp2)
+		tp2.Close()
+		tp1 := newTestPeer("1")
+		addToCache(t, fx, tp1)
+		p, err := fx.GetOneOf(ctx, []string{"3", "2", "1"})
+		require.NoError(t, err)
+		assert.Equal(t, tp1, p)
+	})
+	t.Run("dial", func(t *testing.T) {
+		fx := newFixture(t)
+		defer fx.Finish()
+		var called bool
+		fx.Dialer.dial = func(ctx context.Context, peerId string) (peer peer.Peer, err error) {
+			if called {
+				return nil, fmt.Errorf("not expected call")
+			}
+			called = true
+			return newTestPeer(peerId), nil
+		}
+		p, err := fx.GetOneOf(ctx, []string{"3", "2", "1"})
+		require.NoError(t, err)
+		assert.NotNil(t, p)
+	})
+	t.Run("unable to connect", func(t *testing.T) {
+		fx := newFixture(t)
+		defer fx.Finish()
+		fx.Dialer.dial = func(ctx context.Context, peerId string) (peer peer.Peer, err error) {
+			return nil, fmt.Errorf("persistent error")
+		}
+		p, err := fx.GetOneOf(ctx, []string{"3", "2", "1"})
+		assert.Equal(t, ErrUnableToConnect, err)
+		assert.Nil(t, p)
+	})
+}
+
+func newFixture(t *testing.T) *fixture {
+	fx := &fixture{
+		Pool:   New(),
+		Dialer: &dialerMock{},
+	}
+	a := new(app.App)
+	a.Register(fx.Pool)
+	a.Register(fx.Dialer)
+	require.NoError(t, a.Start(context.Background()))
+	fx.a = a
+	fx.t = t
+	return fx
+}
+
+func (fx *fixture) Finish() {
+	require.NoError(fx.t, fx.a.Close(context.Background()))
+}
+
+type fixture struct {
+	Pool
+	Dialer *dialerMock
+	a      *app.App
+	t      *testing.T
+}
+
+var _ dialer.Dialer = (*dialerMock)(nil)
+
+type dialerMock struct {
+	dial func(ctx context.Context, peerId string) (peer peer.Peer, err error)
+}
+
+func (d *dialerMock) Dial(ctx context.Context, peerId string) (peer peer.Peer, err error) {
+	return d.dial(ctx, peerId)
+}
+
+func (d *dialerMock) UpdateAddrs(addrs map[string][]string) {
+	return
+}
+
+func (d *dialerMock) Init(a *app.App) (err error) {
+	return
+}
+
+func (d *dialerMock) Name() (name string) {
+	return dialer.CName
+}
+
+func newTestPeer(id string) *testPeer {
+	return &testPeer{
+		id:     id,
+		closed: make(chan struct{}),
+	}
+}
+
+type testPeer struct {
+	id     string
+	closed chan struct{}
+}
+
+func (t *testPeer) Id() string {
+	return t.id
+}
+
+func (t *testPeer) LastUsage() time.Time {
+	return time.Now()
+}
+
+func (t *testPeer) UpdateLastUsage() {}
+
+func (t *testPeer) Close() error {
+	select {
+	case <-t.closed:
+		return fmt.Errorf("already closed")
+	default:
+		close(t.closed)
+	}
+	return nil
+}
+
+func (t *testPeer) Closed() <-chan struct{} {
+	return t.closed
+}
+
+func (t *testPeer) Invoke(ctx context.Context, rpc string, enc drpc.Encoding, in, out drpc.Message) error {
+	return fmt.Errorf("call Invoke on test peer")
+}
+
+func (t *testPeer) NewStream(ctx context.Context, rpc string, enc drpc.Encoding) (drpc.Stream, error) {
+	return nil, fmt.Errorf("call NewStream on test peer")
+}

common/net/rpc/server/drpcserver.go

@@ -4,15 +4,16 @@ import (
 	"context"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/pool"
+	secure2 "github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/secure"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/config"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/pool"
+	"github.com/zeebo/errs"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/rpc"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/secure"
 	"go.uber.org/zap"
+	"io"
 	"net"
 	"storj.io/drpc"
+	"storj.io/drpc/drpcmux"
 	"storj.io/drpc/drpcserver"
-	"strings"
 	"time"
 )
 
@@ -21,25 +22,27 @@ const CName = "net/drpcserver"
 var log = logger.NewNamed(CName)
 
 func New() DRPCServer {
-	return &drpcServer{}
+	return &drpcServer{Mux: drpcmux.New()}
 }
 
 type DRPCServer interface {
 	app.ComponentRunnable
+	drpc.Mux
 }
 
 type drpcServer struct {
 	config     config.GrpcServer
 	drpcServer *drpcserver.Server
-	transport  secure.Service
+	transport  secure2.Service
-	listeners  []secure.ContextListener
+	listeners  []secure2.ContextListener
 	pool       pool.Pool
 	cancel     func()
+	*drpcmux.Mux
 }
 
-func (s *drpcServer) Init(ctx context.Context, a *app.App) (err error) {
+func (s *drpcServer) Init(a *app.App) (err error) {
 	s.config = a.MustComponent(config.CName).(*config.Config).GrpcServer
-	s.transport = a.MustComponent(secure.CName).(secure.Service)
+	s.transport = a.MustComponent(secure2.CName).(secure2.Service)
 	s.pool = a.MustComponent(pool.CName).(pool.Pool)
 	return nil
 }
@@ -49,7 +52,7 @@ func (s *drpcServer) Name() (name string) {
 }
 
 func (s *drpcServer) Run(ctx context.Context) (err error) {
-	s.drpcServer = drpcserver.New(s)
+	s.drpcServer = drpcserver.New(s.Mux)
 	ctx, s.cancel = context.WithCancel(ctx)
 	for _, addr := range s.config.ListenAddrs {
 		tcpList, err := net.Listen("tcp", addr)
@@ -62,7 +65,7 @@ func (s *drpcServer) Run(ctx context.Context) (err error) {
 	return
 }
 
-func (s *drpcServer) serve(ctx context.Context, lis secure.ContextListener) {
+func (s *drpcServer) serve(ctx context.Context, lis secure2.ContextListener) {
 	l := log.With(zap.String("localAddr", lis.Addr().String()))
 	l.Info("drpc listener started")
 	defer func() {
@@ -86,7 +89,7 @@ func (s *drpcServer) serve(ctx context.Context, lis secure.ContextListener) {
 				}
 				continue
 			}
-			if _, ok := err.(secure.HandshakeError); ok {
+			if _, ok := err.(secure2.HandshakeError); ok {
 				l.Warn("listener handshake error", zap.Error(err))
 				continue
 			}
@@ -101,7 +104,7 @@ func (s *drpcServer) serveConn(ctx context.Context, conn net.Conn) {
 	l := log.With(zap.String("remoteAddr", conn.RemoteAddr().String())).With(zap.String("localAddr", conn.LocalAddr().String()))
 	l.Debug("connection opened")
 	if err := s.drpcServer.ServeOne(ctx, conn); err != nil {
-		if err == context.Canceled || strings.Contains(err.Error(), "EOF") {
+		if errs.Is(err, context.Canceled) || errs.Is(err, io.EOF) {
 			l.Debug("connection closed")
 		} else {
 			l.Warn("serve connection error", zap.Error(err))
@@ -109,16 +112,6 @@ func (s *drpcServer) serveConn(ctx context.Context, conn net.Conn) {
 	}
 }
 
-func (s *drpcServer) HandleRPC(stream drpc.Stream, _ string) (err error) {
-	ctx := stream.Context()
-	sc, err := secure.CtxSecureConn(ctx)
-	if err != nil {
-		return
-	}
-	log.With(zap.String("peer", sc.RemotePeer().String())).Debug("stream opened")
-	return s.pool.AddAndReadPeer(rpc.PeerFromStream(sc, stream, true))
-}
-
 func (s *drpcServer) Close(ctx context.Context) (err error) {
 	if s.cancel != nil {
 		s.cancel()

common/net/secure/service.go

@@ -35,7 +35,7 @@ type service struct {
 	key crypto.PrivKey
 }
 
-func (s *service) Init(ctx context.Context, a *app.App) (err error) {
+func (s *service) Init(a *app.App) (err error) {
 	account := a.MustComponent(config.CName).(*config.Config).Account
 	decoder := signingkey.NewEDPrivKeyDecoder()
 	pkb, err := decoder.DecodeFromStringIntoBytes(account.SigningKey)

common/nodeconf/configuration.go

@@ -1,10 +1,10 @@
-package configuration
+package nodeconf
 
 import (
 	"context"
 	"fmt"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/peer"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/peer"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/pool"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/pool"
 	"github.com/anytypeio/go-chash"
 )
 
@@ -13,7 +13,7 @@ func New() Service {
 }
 
 type Configuration interface {
-	// Id returns current configuration id
+	// Id returns current nodeconf id
 	Id() string
 	// AllPeers returns all peers by spaceId except current account
 	AllPeers(ctx context.Context, spaceId string) (peers []peer.Peer, err error)
@@ -40,7 +40,7 @@ func (c *configuration) AllPeers(ctx context.Context, spaceId string) (peers []p
 	nodeIds := c.NodeIds(spaceId)
 	peers = make([]peer.Peer, 0, len(nodeIds))
 	for _, id := range nodeIds {
-		p, e := c.pool.DialAndAddPeer(ctx, id)
+		p, e := c.pool.Get(ctx, id)
 		if e == nil {
 			peers = append(peers, p)
 		}
@@ -53,7 +53,7 @@ func (c *configuration) AllPeers(ctx context.Context, spaceId string) (peers []p
 
 func (c *configuration) OnePeer(ctx context.Context, spaceId string) (p peer.Peer, err error) {
 	nodeIds := c.NodeIds(spaceId)
-	return c.pool.GetOrDialOneOf(ctx, nodeIds)
+	return c.pool.GetOneOf(ctx, nodeIds)
 }
 
 func (c *configuration) NodeIds(spaceId string) []string {

common/nodeconf/service.go

@@ -1,16 +1,15 @@
-package configuration
+package nodeconf
 
 import (
-	"context"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/pool"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/config"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/pool"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/node"
 	"github.com/anytypeio/go-chash"
 )
 
-const CName = "configuration"
+const CName = "common.nodeconf"
 
 const (
 	partitionCount    = 3000
@@ -32,7 +31,7 @@ type service struct {
 	last Configuration
 }
 
-func (s *service) Init(ctx context.Context, a *app.App) (err error) {
+func (s *service) Init(a *app.App) (err error) {
 	conf := a.MustComponent(config.CName).(*config.Config)
 	s.accountId = conf.Account.PeerId
 	s.pool = a.MustComponent(pool.CName).(pool.Pool)

config/config.go

@@ -1,7 +1,6 @@
 package config
 
 import (
-	"context"
 	"fmt"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
@@ -32,8 +31,8 @@ type Config struct {
 	Space      Space      `yaml:"space"`
 }
 
-func (c *Config) Init(ctx context.Context, a *app.App) (err error) {
+func (c *Config) Init(a *app.App) (err error) {
-	logger.NewNamed("config").Info(fmt.Sprint(*c))
+	logger.NewNamed("config").Info(fmt.Sprint(c.Space))
 	return
 }
 

config/space.go

@@ -1,6 +1,6 @@
 package config
 
 type Space struct {
-	GCTTL      int `json:"gcTTL"`
+	GCTTL      int `yaml:"gcTTL"`
-	SyncPeriod int `json:"syncPeriod"`
+	SyncPeriod int `yaml:"syncPeriod"`
 }

etc/config.yml

@@ -11,6 +11,9 @@ account:
   encryptionKey: JgG4CcCbae1qEpe7mKpBzsHjZhXUmDSNVNX2B1gxFZsJyMX4V6kBQUott9zRWyeXaW1ZmpzuxDXnwSQpAnNurhXyGa9iQaAPqzY9A9VWBPD33Yy1eW7TRuVemzToh8jJQKQKnZNbF8ucTWV9qahusKzyvN8uyhrqoW2tAPfA9S3E3ognCuqbLSW6yjE2rBKayvyS1BVwzjSd6FZK4DDyjfU3pbEVjut3wytGEAn9af6sNMmyCnf2MX5vLovWs9rU8av61wD4z7HTsXyGFx4K75N4Go249Hpe9SKAT6HxhRc3yvj63krPLiQV5yMuH2UeMUXBDekUQyNmBEdn9wrur7mLqB67Bc6tcc2PP8XApBCdWJHvHjN4FktSpaG5vbCqoZbLD1oCbk36q2x9s6XM8pydVqD1J9P3nTbfgMb5pJCTFjNtgKeuKv6wjfJeA9jF1VhcJQisfsahgv9MvZ9M8FJpZTq1zKUhYDCRnZxUkraoMS5yNNVdDzaUckKEDthqik7BMWCWT79vq7uVgMwEvGwGi76gtoMg1159bbPMLZ4bdPVfhH2S9QjPrzQfwZSrzB2YeVPjWpaXDeLDity5H8n1NK2oniAQR6gE71n81neSptsuhV6o6QpQ89AU8y57XmEsou4VEryn8vUxBHhULLxrLNUouxyWamCeFiDjk5cSN6koQsf9BYKSNTPFTrwjTKForDokMhcPdMtFktKwjv7u9UEGcY4MKvNzZZkc77gHiP8bqVtdNNoLpTFUC5SZ9i7bKdHvK12HpSy7yzzPeMXJ9UwhLxkok1g81ngTbN1yxRhvYXyHZFtguCR9kvGojDjka91MTBtk551qDw9eCn2xZT9U8jqzBCjdpvSg3mRWKMPnYAGB7m7u1ye165wyGFvzcHAx3vtXjxAqLUeKYZCjv2m6V9D2Y4qH1TQNddWqH14T1JVMis971UCH9Ddpj6a3387oUnufD1P6HZN2ieJCvptrmbGVvxJYYSvmVf1dkwbtqurDRNWD7TJ7gf6iqSP549C9bxP4GpLt3ygjHmMtcuUzstBuztvunJUnQhfnJxqU6LjRdsFzm53wGWgXNxab7ZvQcPyLwsevn1b98FGPnVpS5iY4LjmqW4ugrC6HgrbsjrXiKzR1yZKhLQkCbLzPoaHb8iB5iBnCr7d4yf5CtfpFRqgoqMFdK5LNZYmDX4HzUKN6A7wC3gGiSRFTLcgGZeSMkB5Pa61CZBU7WCQgFxykycE9HRA7PiQa496GWDCV15teToCpFRsAa6jDmR1MGXPeLRqQgve49VXnQN5FL7c1VuEv5SWjeTuCnMB47DJKBaP7eKJNKgLwETALzSCMF3nRiRgeb15kfoS4BbrJ5yupjrvwmbmvNg1AYFFS5sYNWft7K8v87wQvBakRtGP71Kp8NX77XFtu6xdB7sR6jpfC6qJPyB9akWNXgCrWy9kE4ih42gwAZdUugNZ9YtEsgRM3pwb6qJhkAPyEJtrxrja859PCAgqPSQiPQN33PaMkgQ6HJknu8CrjKRiXAycZ16KLUkHV64TNhEjPTcX1a7rqpD131AYMWX8d7CCdc9Ys7RUb6BwguuNSh8rJK3x4AkMDSUsaE8ynKvpC7RXZpJ9Nxfhd
 apiServer:
   port: "8080"
+space:
+  gcTTL: 60
+  syncPeriod: 10
 nodes:
   - peerId: 12D3KooWMHuhZgK2skkLrvL51QQTXaXQKYy2QqfvPNBFnzR2ubA1
     address: 127.0.0.1:4430

etc/configs/12D3KooWMHuhZgK2skkLrvL51QQTXaXQKYy2QqfvPNBFnzR2ubA1.yml

@@ -11,6 +11,9 @@ account:
     encryptionKey: JgG4CcCbae1qEpe7mKpBzsHjZhXUmDSNVNX2B1gxFZsJyMX4V6kBQUott9zRWyeXaW1ZmpzuxDXnwSQpAnNurhXyGa9iQaAPqzY9A9VWBPD33Yy1eW7TRuVemzToh8jJQKQKnZNbF8ucTWV9qahusKzyvN8uyhrqoW2tAPfA9S3E3ognCuqbLSW6yjE2rBKayvyS1BVwzjSd6FZK4DDyjfU3pbEVjut3wytGEAn9af6sNMmyCnf2MX5vLovWs9rU8av61wD4z7HTsXyGFx4K75N4Go249Hpe9SKAT6HxhRc3yvj63krPLiQV5yMuH2UeMUXBDekUQyNmBEdn9wrur7mLqB67Bc6tcc2PP8XApBCdWJHvHjN4FktSpaG5vbCqoZbLD1oCbk36q2x9s6XM8pydVqD1J9P3nTbfgMb5pJCTFjNtgKeuKv6wjfJeA9jF1VhcJQisfsahgv9MvZ9M8FJpZTq1zKUhYDCRnZxUkraoMS5yNNVdDzaUckKEDthqik7BMWCWT79vq7uVgMwEvGwGi76gtoMg1159bbPMLZ4bdPVfhH2S9QjPrzQfwZSrzB2YeVPjWpaXDeLDity5H8n1NK2oniAQR6gE71n81neSptsuhV6o6QpQ89AU8y57XmEsou4VEryn8vUxBHhULLxrLNUouxyWamCeFiDjk5cSN6koQsf9BYKSNTPFTrwjTKForDokMhcPdMtFktKwjv7u9UEGcY4MKvNzZZkc77gHiP8bqVtdNNoLpTFUC5SZ9i7bKdHvK12HpSy7yzzPeMXJ9UwhLxkok1g81ngTbN1yxRhvYXyHZFtguCR9kvGojDjka91MTBtk551qDw9eCn2xZT9U8jqzBCjdpvSg3mRWKMPnYAGB7m7u1ye165wyGFvzcHAx3vtXjxAqLUeKYZCjv2m6V9D2Y4qH1TQNddWqH14T1JVMis971UCH9Ddpj6a3387oUnufD1P6HZN2ieJCvptrmbGVvxJYYSvmVf1dkwbtqurDRNWD7TJ7gf6iqSP549C9bxP4GpLt3ygjHmMtcuUzstBuztvunJUnQhfnJxqU6LjRdsFzm53wGWgXNxab7ZvQcPyLwsevn1b98FGPnVpS5iY4LjmqW4ugrC6HgrbsjrXiKzR1yZKhLQkCbLzPoaHb8iB5iBnCr7d4yf5CtfpFRqgoqMFdK5LNZYmDX4HzUKN6A7wC3gGiSRFTLcgGZeSMkB5Pa61CZBU7WCQgFxykycE9HRA7PiQa496GWDCV15teToCpFRsAa6jDmR1MGXPeLRqQgve49VXnQN5FL7c1VuEv5SWjeTuCnMB47DJKBaP7eKJNKgLwETALzSCMF3nRiRgeb15kfoS4BbrJ5yupjrvwmbmvNg1AYFFS5sYNWft7K8v87wQvBakRtGP71Kp8NX77XFtu6xdB7sR6jpfC6qJPyB9akWNXgCrWy9kE4ih42gwAZdUugNZ9YtEsgRM3pwb6qJhkAPyEJtrxrja859PCAgqPSQiPQN33PaMkgQ6HJknu8CrjKRiXAycZ16KLUkHV64TNhEjPTcX1a7rqpD131AYMWX8d7CCdc9Ys7RUb6BwguuNSh8rJK3x4AkMDSUsaE8ynKvpC7RXZpJ9Nxfhd
 apiServer:
     port: "8080"
+space:
+  gcTTL: 60
+  syncPeriod: 10
 nodes:
     - peerId: 12D3KooWMHuhZgK2skkLrvL51QQTXaXQKYy2QqfvPNBFnzR2ubA1
       address: 127.0.0.1:4430

etc/configs/12D3KooWT3c7Y5zvWhhjSxd5Ve3GKZi6WCsG6JHxcxgXixRFdBbw.yml

@@ -11,6 +11,9 @@ account:
     encryptionKey: JgG4CcCbae1qEpe7mKXzp7m5hNc56SSyZd9DwUaEStKJrq7RToAC2Vgd3i6hKRwa58zCWeN6Wjc3o6qrdKPEPRvcyEPysamajVo5mdQiUgWAmr97pGEsyjuRjQoC2GY2LvLiEQxEgwFgJxKGMHMiaWMtDfxCDUaDEm4bu5RdMhqRZekAWho6c3WoEeruSr14iX1TrocFNfBkBY7CjEw8kcywXCTNgtvhb2Qiwgj5AxEF4wyw4bzaNA9ctXb1hoHPFVMu6C51pkFY7jUD9zwyH3ukgnAewkGAcPNbKmaTAtMosKRVaAN97mAwXh2VRt1hWmRvVk7r76EjnVKhD4vbsKZc56RVcHTVWRVdhU7FGyPsiE5rSQAz1JQGYzxnZpX7EG77CyrmUGyfueVfRHhwY2oq8A4uQCRaQxSaJHYLowjXSxh8DQ2V6MTqyzti32C27utBYdHzLVCJSGkmdzGwrFcHqsq7nLDxmvJVErPvyReixEe8kFmqopJ3e6LLm8WdYw9K6JYBjXnEfwPzm7Von9sf3dcaGDUHYfttMyeke7fAXJkvPRje69hYVyzdQGAauuojzGkkvQWCSMK1KCMNMznRaPDCNvofrQhYrub24WhmwpKhorufdfW8Cb4T6reBDCtaWVsbuinjtL6F6Sui5aYHJFLJ6e4pPewr1P4EuZYRbMBZwN5KvDLhTGLBuBnaTqUUdF6bj2U22NoRYMogiHiftqKqiexKNDXX1Zg9RQEvxgjuVo6SBW42mVEA8agrLhruRqCmiduJxVrfqLNGeYXHXrcmMEgW7uosJbPXvTcfRvdFWS1ov7oSALvj6vhDQ28Yi9D2ETNdNsfVWAFQuwvPpW7CHQGXTitprVbqH8JYxNZuGygcLmr5efbB22Vzu4ntd1HoraQpG12qeDEUA7tXYUpoYyuSdWwKPjSAMtaQcCSfVrhKQHQuKJargrVrez8vjWuwLfvSucV7ZHe7gjqvYgULdE1ubRCRSd7DuLjEN2Vd6obzV2c3MRet7ZSf4Sp88WM5AuTyW7BjArBc4S3gUQ8rYaiZ8Tu7NCxkEzbFwWRaemZkwfvcsX3XxqjyF37tFSGkEqE5kuBvpZW72675LkDffj7kH1zA8yE6dVujJjWsNYVFJWndUtz5Vy2KCdZAbBgq19q4AtsxWPodU2N3yZXzFAFAzTrxS6V4P7Scpdau1avgRvHLcBQPunA37xaYMy8YMifJwtmRY25mnAQwZAk3eANk7tXwZd58SDnciLNvARJvwKzTQBXcshkwyy52SX8XmXDJsPnRLaHmiYBJ63Yzr5XpZuuAtxb9qrWG2NHCNxfomHokWacV1hjZPPd6ZxT1FuRozB6Qt2NLcyqY7bnTcQJb1jPUaTAGXXCR8WVmmmYo2fDQe8CdBmgyPvbzNTEJUyScBz4RdycB5PZap4SurJCWtHbuMyQbQUB6jJgURDstfXS5Akfe4oruNq9rnYcNtnsDJPtrhXHBqzDizmf1BDxR5FB2RCxzCgeAfg8WQ1Ug9PVAGTzob6ZqCrGXzWXEUniZnf1vjr7QhGKBYXEX9SWDoSMUpP4FreVDTnx15ijRZTV3p8xG5fE9e36TnugRVvTyq7XzmyPBjW2r66f1bior
 apiServer:
     port: "8081"
+space:
+  gcTTL: 60
+  syncPeriod: 10
 nodes:
     - peerId: 12D3KooWMHuhZgK2skkLrvL51QQTXaXQKYy2QqfvPNBFnzR2ubA1
       address: 127.0.0.1:4430

go.mod

@@ -32,6 +32,7 @@ require (
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
 	github.com/fogleman/gg v1.3.0 // indirect
 	github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/klauspost/cpuid/v2 v2.0.12 // indirect
 	github.com/libp2p/go-buffer-pool v0.0.2 // indirect
 	github.com/libp2p/go-openssl v0.0.7 // indirect
@@ -46,12 +47,13 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
-	github.com/zeebo/errs v1.2.2 // indirect
+	github.com/zeebo/errs v1.3.0 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
 	golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4 // indirect
 	golang.org/x/image v0.0.0-20200119044424-58c23975cae1 // indirect
 	golang.org/x/sys v0.0.0-20220422013727-9388b58f7150 // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	lukechampine.com/blake3 v1.1.6 // indirect

go.sum

@@ -32,6 +32,10 @@ github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 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/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw=
+github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/huandu/go-assert v1.1.5 h1:fjemmA7sSfYHJD7CUqs9qTwwfdNAx7/j2/ZlHXzNB3c=
 github.com/huandu/go-assert v1.1.5/go.mod h1:yOLvuqZwmcHIC5rIzrBhT7D3Q9c3GFnd0JrPVhn/06U=
 github.com/huandu/skiplist v1.2.0 h1:gox56QD77HzSC0w+Ws3MH3iie755GBJU1OER3h5VsYw=
@@ -115,6 +119,8 @@ github.com/zeebo/blake3 v0.2.3 h1:TFoLXsjeXqRNFxSbk35Dk4YtszE/MQQGK10BH4ptoTg=
 github.com/zeebo/blake3 v0.2.3/go.mod h1:mjJjZpnsyIVtVgTOSpJ9vmRE4wgDeyt2HU3qXvvKCaQ=
 github.com/zeebo/errs v1.2.2 h1:5NFypMTuSdoySVTqlNs1dEoU21QVamMQJxW/Fii5O7g=
 github.com/zeebo/errs v1.2.2/go.mod h1:sgbWHsvVuTPHcqJJGQ1WhI5KbWlHYz+2+2C/LSEtCw4=
+github.com/zeebo/errs v1.3.0 h1:hmiaKqgYZzcVgRL1Vkc1Mn2914BbzB0IBxs+ebeutGs=
+github.com/zeebo/errs v1.3.0/go.mod h1:sgbWHsvVuTPHcqJJGQ1WhI5KbWlHYz+2+2C/LSEtCw4=
 github.com/zeebo/pcg v1.0.1 h1:lyqfGeWiv4ahac6ttHs+I5hwtH/+1mrhlCtVNQM2kHo=
 github.com/zeebo/pcg v1.0.1/go.mod h1:09F0S9iiKrwn9rlI5yjLkmrug154/YRW6KnnXVDM/l4=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
@@ -178,6 +184,10 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+google.golang.org/protobuf v1.28.1 h1:d0NfwRgPtno5B1Wa6L2DAG+KivqkdutMf1UhdNx175w=
+google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=

node/nodespace/rpchandler.go

@@ -0,0 +1,30 @@
+package nodespace
+
+import (
+	"context"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace/spacesyncproto"
+)
+
+type rpcHandler struct {
+	s *service
+}
+
+func (r *rpcHandler) HeadSync(ctx context.Context, req *spacesyncproto.HeadSyncRequest) (*spacesyncproto.HeadSyncResponse, error) {
+	sp, err := r.s.GetSpace(ctx, req.SpaceId)
+	if err != nil {
+		return nil, err
+	}
+	return sp.SpaceSyncRpc().HeadSync(ctx, req)
+}
+
+func (r *rpcHandler) Stream(stream spacesyncproto.DRPCSpace_StreamStream) error {
+	msg, err := stream.Recv()
+	if err != nil {
+		return err
+	}
+	sp, err := r.s.GetSpace(stream.Context(), msg.SpaceId)
+	if err != nil {
+		return err
+	}
+	return sp.SpaceSyncRpc().Stream(stream)
+}

node/nodespace/service.go

@@ -0,0 +1,71 @@
+package nodespace
+
+import (
+	"context"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace/spacesyncproto"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/rpc/server"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/config"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/ocache"
+	"time"
+)
+
+const CName = "node.nodespace"
+
+var log = logger.NewNamed(CName)
+
+func New() Service {
+	return &service{}
+}
+
+type Service interface {
+	GetSpace(ctx context.Context, id string) (commonspace.Space, error)
+	app.ComponentRunnable
+}
+
+type service struct {
+	conf        config.Space
+	cache       ocache.OCache
+	commonSpace commonspace.Service
+}
+
+func (s *service) Init(a *app.App) (err error) {
+	s.conf = a.MustComponent(config.CName).(*config.Config).Space
+	s.commonSpace = a.MustComponent(commonspace.CName).(commonspace.Service)
+	s.cache = ocache.New(
+		func(ctx context.Context, id string) (value ocache.Object, err error) {
+			return s.commonSpace.CreateSpace(ctx, id)
+		},
+		ocache.WithLogger(log.Sugar()),
+		ocache.WithGCPeriod(time.Minute),
+		ocache.WithTTL(time.Duration(s.conf.GCTTL)*time.Second),
+		ocache.WithRefCounter(false),
+	)
+	return spacesyncproto.DRPCRegisterSpace(a.MustComponent(server.CName).(server.DRPCServer), &rpcHandler{s})
+}
+
+func (s *service) Name() (name string) {
+	return CName
+}
+
+func (s *service) Run(ctx context.Context) (err error) {
+	go func() {
+		time.Sleep(time.Second * 5)
+		_, _ = s.GetSpace(ctx, "testDSpace")
+	}()
+	return
+}
+
+func (s *service) GetSpace(ctx context.Context, id string) (commonspace.Space, error) {
+	v, err := s.cache.Get(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	return v.(commonspace.Space), nil
+}
+
+func (s *service) Close(ctx context.Context) (err error) {
+	return s.cache.Close()
+}

pkg/ocache/ocache.go

@@ -27,9 +27,9 @@ type LoadFunc func(ctx context.Context, id string) (value Object, err error)
 
 type Option func(*oCache)
 
-var WithLogServiceName = func(name string) Option {
+var WithLogger = func(l *zap.SugaredLogger) Option {
 	return func(cache *oCache) {
-		cache.log = cache.log.With("service_name", name)
+		cache.log = l
 	}
 }
 
@@ -45,6 +45,12 @@ var WithGCPeriod = func(gcPeriod time.Duration) Option {
 	}
 }
 
+var WithRefCounter = func(enable bool) Option {
+	return func(cache *oCache) {
+		cache.noRefCounter = !enable
+	}
+}
+
 func New(loadFunc LoadFunc, opts ...Option) OCache {
 	c := &oCache{
 		data:     make(map[string]*entry),
@@ -69,10 +75,13 @@ type Object interface {
 }
 
 type ObjectLocker interface {
-	Object
 	Locked() bool
 }
 
+type ObjectLastUsage interface {
+	LastUsage() time.Time
+}
+
 type entry struct {
 	id        string
 	lastUsage time.Time
@@ -99,7 +108,7 @@ type OCache interface {
 	// When 'loadFunc' returns a non-nil error, an object will not be stored to cache
 	Get(ctx context.Context, id string) (value Object, err error)
 	// Pick returns value if it's presents in cache (will not call loadFunc)
-	Pick(id string) (value Object, err error)
+	Pick(ctx context.Context, id string) (value Object, err error)
 	// Add adds new object to cache
 	// Returns error when object exists
 	Add(id string, value Object) (err error)
@@ -121,15 +130,16 @@ type OCache interface {
 }
 
 type oCache struct {
-	mu       sync.Mutex
+	mu           sync.Mutex
-	data     map[string]*entry
+	data         map[string]*entry
-	loadFunc LoadFunc
+	loadFunc     LoadFunc
-	timeNow  func() time.Time
+	timeNow      func() time.Time
-	ttl      time.Duration
+	ttl          time.Duration
-	gc       time.Duration
+	gc           time.Duration
-	closed   bool
+	closed       bool
-	closeCh  chan struct{}
+	closeCh      chan struct{}
-	log      *zap.SugaredLogger
+	log          *zap.SugaredLogger
+	noRefCounter bool
 }
 
 func (c *oCache) Get(ctx context.Context, id string) (value Object, err error) {
@@ -152,7 +162,9 @@ func (c *oCache) Get(ctx context.Context, id string) (value Object, err error) {
 		c.data[id] = e
 	}
 	e.lastUsage = c.timeNow()
-	e.refCount++
+	if !c.noRefCounter {
+		e.refCount++
+	}
 	c.mu.Unlock()
 
 	if load {
@@ -166,13 +178,18 @@ func (c *oCache) Get(ctx context.Context, id string) (value Object, err error) {
 	return e.value, e.loadErr
 }
 
-func (c *oCache) Pick(id string) (value Object, err error) {
+func (c *oCache) Pick(ctx context.Context, id string) (value Object, err error) {
 	c.mu.Lock()
 	val, ok := c.data[id]
 	c.mu.Unlock()
 	if !ok {
 		return nil, ErrNotExists
 	}
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-val.load:
+	}
 	<-val.load
 	return val.value, val.loadErr
 }
@@ -198,7 +215,7 @@ func (c *oCache) Release(id string) bool {
 		return false
 	}
 	if e, ok := c.data[id]; ok {
-		if e.refCount > 0 {
+		if !c.noRefCounter && e.refCount > 0 {
 			e.refCount--
 			return true
 		}
@@ -307,7 +324,11 @@ func (c *oCache) GC() {
 	deadline := c.timeNow().Add(-c.ttl)
 	var toClose []*entry
 	for k, e := range c.data {
-		if !e.locked() && e.refCount <= 0 && e.lastUsage.Before(deadline) {
+		lu := e.lastUsage
+		if lug, ok := e.value.(ObjectLastUsage); ok {
+			lu = lug.LastUsage()
+		}
+		if !e.locked() && e.refCount <= 0 && lu.Before(deadline) {
 			delete(c.data, k)
 			toClose = append(toClose, e)
 		}

service/api/service.go

@@ -30,7 +30,7 @@ type service struct {
 	cfg             *config.Config
 }
 
-func (s *service) Init(ctx context.Context, a *app.App) (err error) {
+func (s *service) Init(a *app.App) (err error) {
 	s.treeCache = a.MustComponent(treecache.CName).(treecache.Service)
 	s.documentService = a.MustComponent(document.CName).(document.Service)
 	s.cfg = a.MustComponent(config.CName).(*config.Config)

service/document/service.go

@@ -5,11 +5,11 @@ import (
 	"fmt"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/account"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/aclchanges/aclpb"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/list"
 	testchanges "github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/testutils/testchanges/proto"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/tree"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/account"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/node"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/storage"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/sync/message"
@@ -41,7 +41,7 @@ func New() app.Component {
 	return &service{}
 }
 
-func (s *service) Init(ctx context.Context, a *app.App) (err error) {
+func (s *service) Init(a *app.App) (err error) {
 	s.account = a.MustComponent(account.CName).(account.Service)
 	s.messageService = a.MustComponent(message.CName).(message.Service)
 	s.treeCache = a.MustComponent(treecache.CName).(treecache.Service)

service/net/peer/peer.go

@@ -1,35 +0,0 @@
-package peer
-
-import (
-	"context"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/syncproto"
-	"time"
-)
-
-type Dir uint
-
-const (
-	// DirInbound indicates peer created connection
-	DirInbound Dir = iota
-	// DirOutbound indicates that our host created  connection
-	DirOutbound
-)
-
-type Info struct {
-	Id             string
-	Dir            Dir
-	LastActiveUnix int64
-}
-
-func (i Info) LastActive() time.Time {
-	return time.Unix(i.LastActiveUnix, 0)
-}
-
-type Peer interface {
-	Id() string
-	Info() Info
-	Recv() (*syncproto.Message, error)
-	Send(msg *syncproto.Message) (err error)
-	Context() context.Context
-	Close() error
-}

service/net/pool/handler/reply.go

@@ -1,32 +0,0 @@
-package handler
-
-import (
-	"context"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/pool"
-	"github.com/gogo/protobuf/proto"
-	"go.uber.org/zap"
-)
-
-var log = logger.NewNamed("replyHandler")
-
-type ReplyHandler interface {
-	Handle(ctx context.Context, req []byte) (rep proto.Marshaler, err error)
-}
-
-type Reply struct {
-	ReplyHandler
-}
-
-func (r Reply) Handle(ctx context.Context, msg *pool.Message) error {
-	rep, e := r.ReplyHandler.Handle(ctx, msg.GetData())
-	if msg.GetHeader().RequestId == 0 {
-		if e != nil {
-			log.Error("handler returned error", zap.Error(e))
-		} else if rep != nil {
-			log.Debug("sender didn't expect a reply, but the handler made")
-		}
-		return nil
-	}
-	return msg.ReplyType(msg.GetHeader().GetType(), rep)
-}

service/net/pool/message.go

@@ -1,136 +0,0 @@
-package pool
-
-import (
-	"fmt"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/peer"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/syncproto"
-	"github.com/gogo/protobuf/proto"
-	"go.uber.org/zap"
-	"gopkg.in/mgo.v2/bson"
-)
-
-type Message struct {
-	*syncproto.Message
-	peer peer.Peer
-}
-
-func (m *Message) Peer() peer.Peer {
-	return m.peer
-}
-
-func (m *Message) Reply(data []byte) (err error) {
-	rep := &syncproto.Message{
-		Header: &syncproto.Header{
-			TraceId: m.GetHeader().TraceId,
-			ReplyId: m.GetHeader().RequestId,
-			Type:    syncproto.MessageType_MessageTypeSync,
-		},
-		Data: data,
-	}
-	return m.peer.Send(rep)
-}
-
-func (m *Message) ReplyType(tp syncproto.MessageType, data proto.Marshaler) (err error) {
-	dataBytes, err := data.Marshal()
-	if err != nil {
-		return err
-	}
-	rep := &syncproto.Message{
-		Header: &syncproto.Header{
-			TraceId: m.GetHeader().TraceId,
-			ReplyId: m.GetHeader().RequestId,
-			Type:    tp,
-		},
-		Data: dataBytes,
-	}
-	return m.peer.Send(rep)
-}
-
-func (m *Message) Ack() (err error) {
-	ack := &syncproto.System{
-		Ack: &syncproto.SystemAck{},
-	}
-	data, err := ack.Marshal()
-	if err != nil {
-		return
-	}
-	rep := &syncproto.Message{
-		Header: &syncproto.Header{
-			TraceId:   m.GetHeader().TraceId,
-			ReplyId:   m.GetHeader().RequestId,
-			Type:      syncproto.MessageType_MessageTypeSystem,
-			DebugInfo: "Ack",
-		},
-		Data: data,
-	}
-	err = m.peer.Send(rep)
-	if err != nil {
-		log.With(
-			zap.String("peerId", m.peer.Id()),
-			zap.String("header", rep.GetHeader().String())).
-			Error("failed sending ack to peer", zap.Error(err))
-	} else {
-		log.With(
-			zap.String("peerId", m.peer.Id()),
-			zap.String("header", rep.GetHeader().String())).
-			Debug("sent ack to peer")
-	}
-	return
-}
-
-func (m *Message) AckError(code syncproto.SystemErrorCode, description string) (err error) {
-	ack := &syncproto.System{
-		Ack: &syncproto.SystemAck{
-			Error: &syncproto.SystemError{
-				Code:        code,
-				Description: description,
-			},
-		},
-	}
-	data, err := ack.Marshal()
-	if err != nil {
-		return
-	}
-	rep := &syncproto.Message{
-		Header: &syncproto.Header{
-			TraceId:   []byte(bson.NewObjectId()),
-			ReplyId:   m.GetHeader().RequestId,
-			Type:      syncproto.MessageType_MessageTypeSystem,
-			DebugInfo: "AckError",
-		},
-		Data: data,
-	}
-	if err != nil {
-		log.With(
-			zap.String("peerId", m.peer.Id()),
-			zap.String("header", rep.GetHeader().String())).
-			Error("failed sending ackError to peer", zap.Error(err))
-	} else {
-		log.With(
-			zap.String("peerId", m.peer.Id()),
-			zap.String("header", rep.GetHeader().String())).
-			Debug("sent ackError to peer")
-	}
-	return m.peer.Send(rep)
-}
-
-func (m *Message) IsAck() (err error) {
-	if tp := m.GetHeader().GetType(); tp != syncproto.MessageType_MessageTypeSystem {
-		return fmt.Errorf("unexpected message type in response: %v, want System", tp)
-	}
-	sys := &syncproto.System{}
-	if err = sys.Unmarshal(m.GetData()); err != nil {
-		return
-	}
-	if ack := sys.Ack; ack != nil {
-		if ack.Error != nil {
-			return fmt.Errorf("response error: code=%d; descriptipon=%s", ack.Error.Code, ack.Error.Description)
-		}
-		return nil
-	}
-	return fmt.Errorf("received not ack response")
-}
-
-func (m *Message) UnmarshalData(msg proto.Unmarshaler) error {
-	return msg.Unmarshal(m.Data)
-}

service/net/pool/peer.go

@@ -1,28 +0,0 @@
-package pool
-
-import (
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/peer"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/util/slice"
-)
-
-type peerEntry struct {
-	peer     peer.Peer
-	groupIds []string
-	ready    chan struct{}
-}
-
-func (pe *peerEntry) addGroup(groupId string) (ok bool) {
-	if slice.FindPos(pe.groupIds, groupId) != -1 {
-		return false
-	}
-	pe.groupIds = append(pe.groupIds, groupId)
-	return true
-}
-
-func (pe *peerEntry) removeGroup(groupId string) (ok bool) {
-	if slice.FindPos(pe.groupIds, groupId) == -1 {
-		return false
-	}
-	pe.groupIds = slice.Remove(pe.groupIds, groupId)
-	return true
-}

service/net/pool/pool.go

@@ -1,397 +0,0 @@
-package pool
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/dialer"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/peer"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/syncproto"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/util/slice"
-	"go.uber.org/zap"
-	"math/rand"
-	"sync"
-	"sync/atomic"
-)
-
-const (
-	CName                              = "sync/peerPool"
-	maxSimultaneousOperationsPerStream = 10
-)
-
-var log = logger.NewNamed("peerPool")
-
-var (
-	ErrPoolClosed   = errors.New("peer pool is closed")
-	ErrPeerNotFound = errors.New("peer not found")
-)
-
-func NewPool() Pool {
-	return &pool{closed: true}
-}
-
-type Handler func(ctx context.Context, msg *Message) (err error)
-
-type Pool interface {
-	AddAndReadPeer(peer peer.Peer) (err error)
-	AddHandler(msgType syncproto.MessageType, h Handler)
-	AddPeerIdToGroup(peerId, groupId string) (err error)
-	RemovePeerIdFromGroup(peerId, groupId string) (err error)
-	DialAndAddPeer(ctx context.Context, id string) (peer.Peer, error)
-	GetOrDialOneOf(ctx context.Context, peerIds []string) (peer.Peer, error)
-
-	SendAndWait(ctx context.Context, peerId string, msg *syncproto.Message) (err error)
-	SendAndWaitResponse(ctx context.Context, id string, s *syncproto.Message) (resp *Message, err error)
-	Broadcast(ctx context.Context, groupId string, msg *syncproto.Message) (err error)
-
-	app.ComponentRunnable
-}
-
-type pool struct {
-	peersById       map[string]*peerEntry
-	waiters         *waiters
-	handlers        map[syncproto.MessageType][]Handler
-	peersIdsByGroup map[string][]string
-
-	dialer dialer.Dialer
-
-	closed bool
-	mu     sync.RWMutex
-	wg     *sync.WaitGroup
-}
-
-func (p *pool) Init(ctx context.Context, a *app.App) (err error) {
-	p.peersById = map[string]*peerEntry{}
-	p.handlers = map[syncproto.MessageType][]Handler{}
-	p.peersIdsByGroup = map[string][]string{}
-	p.waiters = &waiters{waiters: map[uint64]*waiter{}}
-	p.dialer = a.MustComponent(dialer.CName).(dialer.Dialer)
-	p.wg = &sync.WaitGroup{}
-	return nil
-}
-
-func (p *pool) Name() (name string) {
-	return CName
-}
-
-func (p *pool) Run(ctx context.Context) (err error) {
-	p.closed = false
-	return nil
-}
-
-func (p *pool) AddHandler(msgType syncproto.MessageType, h Handler) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	if !p.closed {
-		// unable to add handler after Run
-		return
-	}
-	p.handlers[msgType] = append(p.handlers[msgType], h)
-}
-
-func (p *pool) DialAndAddPeer(ctx context.Context, peerId string) (peer.Peer, error) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	if p.closed {
-		return nil, ErrPoolClosed
-	}
-	return p.dialAndAdd(ctx, peerId)
-}
-
-func (p *pool) dialAndAdd(ctx context.Context, peerId string) (peer.Peer, error) {
-	if peer, ok := p.peersById[peerId]; ok {
-		return peer.peer, nil
-	}
-	peer, err := p.dialer.Dial(ctx, peerId)
-	if err != nil {
-		return nil, err
-	}
-	p.peersById[peer.Id()] = &peerEntry{
-		peer: peer,
-	}
-	p.wg.Add(1)
-	go p.readPeerLoop(peer)
-	return peer, nil
-}
-
-func (p *pool) AddAndReadPeer(peer peer.Peer) (err error) {
-	p.mu.Lock()
-	if p.closed {
-		p.mu.Unlock()
-		return ErrPoolClosed
-	}
-	p.peersById[peer.Id()] = &peerEntry{
-		peer: peer,
-	}
-	p.wg.Add(1)
-	p.mu.Unlock()
-	return p.readPeerLoop(peer)
-}
-
-func (p *pool) AddPeerIdToGroup(peerId, groupId string) (err error) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	peer, ok := p.peersById[peerId]
-	if !ok {
-		return ErrPeerNotFound
-	}
-	if slice.FindPos(peer.groupIds, groupId) != -1 {
-		return nil
-	}
-	peer.addGroup(groupId)
-	p.peersIdsByGroup[groupId] = append(p.peersIdsByGroup[groupId], peerId)
-	return
-}
-
-func (p *pool) RemovePeerIdFromGroup(peerId, groupId string) (err error) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	peer, ok := p.peersById[peerId]
-	if !ok {
-		return ErrPeerNotFound
-	}
-	if slice.FindPos(peer.groupIds, groupId) == -1 {
-		return nil
-	}
-	peer.removeGroup(groupId)
-	p.peersIdsByGroup[groupId] = slice.Remove(p.peersIdsByGroup[groupId], peerId)
-	return
-}
-
-func (p *pool) SendAndWait(ctx context.Context, peerId string, msg *syncproto.Message) (err error) {
-	resp, err := p.SendAndWaitResponse(ctx, peerId, msg)
-	if err != nil {
-		return
-	}
-	return resp.IsAck()
-}
-
-func (p *pool) SendAndWaitResponse(ctx context.Context, peerId string, msg *syncproto.Message) (resp *Message, err error) {
-	defer func() {
-		if err != nil {
-			log.With(
-				zap.String("peerId", peerId),
-				zap.String("header", msg.GetHeader().String())).
-				Error("failed sending message to peer", zap.Error(err))
-		} else {
-			log.With(
-				zap.String("peerId", peerId),
-				zap.String("header", msg.GetHeader().String())).
-				Debug("sent message to peer")
-		}
-	}()
-
-	p.mu.RLock()
-	peer := p.peersById[peerId]
-	p.mu.RUnlock()
-	if peer == nil {
-		err = ErrPeerNotFound
-		return
-	}
-
-	repId := p.waiters.NewReplyId()
-	msg.GetHeader().RequestId = repId
-	ch := make(chan Reply, 1)
-
-	log.With(zap.Uint64("reply id", repId)).Debug("adding waiter for reply id")
-	p.waiters.Add(repId, &waiter{ch: ch})
-	defer p.waiters.Remove(repId)
-
-	if err = peer.peer.Send(msg); err != nil {
-		return
-	}
-	select {
-	case rep := <-ch:
-		if rep.Error != nil {
-			err = rep.Error
-			return
-		}
-		resp = rep.Message
-		return
-	case <-ctx.Done():
-		log.Debug("context done in SendAndWait")
-		err = ctx.Err()
-	}
-	return
-}
-
-func (p *pool) GetOrDialOneOf(ctx context.Context, peerIds []string) (peer.Peer, error) {
-	p.mu.RLock()
-	if p.closed {
-		p.mu.RUnlock()
-		return nil, ErrPoolClosed
-	}
-	for _, peerId := range peerIds {
-		peer, ok := p.peersById[peerId]
-		if ok {
-			p.mu.RUnlock()
-			return peer.peer, nil
-		}
-	}
-	p.mu.RUnlock()
-	rand.Shuffle(len(peerIds), func(i, j int) {
-		peerIds[i], peerIds[j] = peerIds[j], peerIds[i]
-	})
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	var lastErr error
-	for _, peerId := range peerIds {
-		peer, err := p.dialAndAdd(ctx, peerId)
-		if err != nil {
-			lastErr = err
-			continue
-		} else {
-			return peer, nil
-		}
-	}
-	return nil, lastErr
-}
-
-func (p *pool) Broadcast(ctx context.Context, groupId string, msg *syncproto.Message) (err error) {
-	//TODO implement me
-	panic("implement me")
-}
-
-func (p *pool) readPeerLoop(peer peer.Peer) (err error) {
-	defer p.wg.Done()
-
-	limiter := make(chan struct{}, maxSimultaneousOperationsPerStream)
-	for i := 0; i < maxSimultaneousOperationsPerStream; i++ {
-		limiter <- struct{}{}
-	}
-Loop:
-	for {
-		msg, err := peer.Recv()
-		if err != nil {
-			log.Debug("peer receive error", zap.Error(err), zap.String("peerId", peer.Id()))
-			break
-		}
-		select {
-		case <-limiter:
-		case <-peer.Context().Done():
-			break Loop
-		}
-		go func() {
-			defer func() {
-				limiter <- struct{}{}
-			}()
-			p.handleMessage(peer, msg)
-		}()
-	}
-	if err = p.removePeer(peer.Id()); err != nil {
-		log.Error("remove peer error", zap.String("peerId", peer.Id()), zap.Error(err))
-	}
-	return
-}
-
-func (p *pool) removePeer(peerId string) (err error) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	_, ok := p.peersById[peerId]
-	if !ok {
-		return ErrPeerNotFound
-	}
-	delete(p.peersById, peerId)
-	return
-}
-
-func (p *pool) handleMessage(peer peer.Peer, msg *syncproto.Message) {
-	log.With(zap.String("peerId", peer.Id()), zap.String("header", msg.GetHeader().String())).
-		Debug("received message from peer")
-	replyId := msg.GetHeader().GetReplyId()
-	if replyId != 0 {
-		if !p.waiters.Send(replyId, Reply{
-			PeerInfo: peer.Info(),
-			Message: &Message{
-				Message: msg,
-				peer:    peer,
-			},
-		}) {
-			log.Debug("received reply with unknown (or expired) replyId", zap.Uint64("replyId", replyId), zap.String("header", msg.GetHeader().String()))
-		}
-		return
-	}
-	handlers := p.handlers[msg.GetHeader().GetType()]
-	if len(handlers) == 0 {
-		log.With(zap.String("peerId", peer.Id())).Debug("no handlers for such message")
-		return
-	}
-
-	message := &Message{Message: msg, peer: peer}
-
-	for _, h := range handlers {
-		if err := h(peer.Context(), message); err != nil {
-			log.Error("handle message error", zap.Error(err))
-		}
-	}
-}
-
-func (p *pool) Close(ctx context.Context) (err error) {
-	p.mu.Lock()
-	for _, peer := range p.peersById {
-		peer.peer.Close()
-	}
-	wg := p.wg
-	p.mu.Unlock()
-	if wg != nil {
-		wg.Wait()
-	}
-	return nil
-}
-
-type waiter struct {
-	sent int
-	ch   chan<- Reply
-}
-
-type waiters struct {
-	waiters  map[uint64]*waiter
-	replySeq uint64
-	mu       sync.Mutex
-}
-
-func (w *waiters) Send(replyId uint64, r Reply) (ok bool) {
-	w.mu.Lock()
-	wait := w.waiters[replyId]
-	if wait == nil {
-		w.mu.Unlock()
-		return false
-	}
-	wait.sent++
-	var lastMessage = wait.sent == cap(wait.ch)
-	if lastMessage {
-		delete(w.waiters, replyId)
-	}
-	w.mu.Unlock()
-	wait.ch <- r
-	if lastMessage {
-		close(wait.ch)
-	}
-	return true
-}
-
-func (w *waiters) Add(replyId uint64, wait *waiter) {
-	w.mu.Lock()
-	w.waiters[replyId] = wait
-	w.mu.Unlock()
-}
-
-func (w *waiters) Remove(id uint64) error {
-	w.mu.Lock()
-	defer w.mu.Unlock()
-	if _, ok := w.waiters[id]; ok {
-		delete(w.waiters, id)
-		return nil
-	}
-	return fmt.Errorf("waiter not found")
-}
-
-func (w *waiters) NewReplyId() uint64 {
-	res := atomic.AddUint64(&w.replySeq, 1)
-	if res == 0 {
-		return w.NewReplyId()
-	}
-	return res
-}

service/net/pool/request.go

@@ -1,45 +0,0 @@
-package pool
-
-import "context"
-
-// 1. message for one peerId with ack
-// 		pool.SendAndWait(ctx context,.C
-// 2. message for many peers without ack (or group)
-
-type Request struct {
-	groupId   string
-	oneOf     []string
-	all       []string
-	tryDial   bool
-	needReply bool
-	pool      *pool
-}
-
-func (r *Request) GroupId(groupId string) *Request {
-	r.groupId = groupId
-	return r
-}
-
-func (r *Request) All(peerIds ...string) *Request {
-	r.all = peerIds
-	return r
-}
-
-func (r *Request) OneOf(peerIds ...string) *Request {
-	r.oneOf = peerIds
-	return r
-}
-
-func (r *Request) TryDial(is bool) *Request {
-	r.tryDial = is
-	return r
-}
-
-func (r *Request) NeedReply(is bool) *Request {
-	r.needReply = is
-	return r
-}
-
-func (r *Request) Exec(ctx context.Context, msg *Message) *Results {
-	return nil
-}

service/net/pool/result.go

@@ -1,53 +0,0 @@
-package pool
-
-import (
-	"context"
-	"fmt"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/peer"
-)
-
-// Results of request collects replies and errors
-// Must be closed after usage r.Close()
-type Results struct {
-	ctx      context.Context
-	cancel   func()
-	waiterId uint64
-	ch       chan Reply
-	pool     *pool
-}
-
-// Iterate iterates over replies
-// if callback will return a non-nil error then iteration stops
-func (r *Results) Iterate(callback func(r Reply) (err error)) (err error) {
-	if r.ctx == nil || r.ch == nil {
-		return fmt.Errorf("results not initialized")
-	}
-	for {
-		select {
-		case <-r.ctx.Done():
-			return r.ctx.Err()
-		case m, ok := <-r.ch:
-			if ok {
-				if err = callback(m); err != nil {
-					return err
-				}
-			} else {
-				return
-			}
-		}
-	}
-}
-
-// Close cancels iteration and unregister reply handler in the pool
-// Required to call to avoid memory leaks
-func (r *Results) Close() (err error) {
-	r.cancel()
-	return r.pool.waiters.Remove(r.waiterId)
-}
-
-// Reply presents the result of request executing can be error or result message
-type Reply struct {
-	PeerInfo peer.Info
-	Error    error
-	Message  *Message
-}

service/net/rpc/encoding.go

@@ -1,18 +0,0 @@
-package rpc
-
-import (
-	"github.com/gogo/protobuf/proto"
-	"storj.io/drpc"
-)
-
-var Encoding = enc{}
-
-type enc struct{}
-
-func (e enc) Marshal(msg drpc.Message) ([]byte, error) {
-	return msg.(proto.Marshaler).Marshal()
-}
-
-func (e enc) Unmarshal(buf []byte, msg drpc.Message) error {
-	return msg.(proto.Unmarshaler).Unmarshal(buf)
-}

service/net/rpc/stream.go

@@ -1,55 +0,0 @@
-package rpc
-
-import (
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/peer"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/syncproto"
-	"github.com/libp2p/go-libp2p-core/sec"
-	"storj.io/drpc"
-	"sync/atomic"
-	"time"
-)
-
-func PeerFromStream(sc sec.SecureConn, stream drpc.Stream, incoming bool) peer.Peer {
-	dp := &drpcPeer{
-		sc:     sc,
-		Stream: stream,
-	}
-	dp.info.Id = sc.RemotePeer().String()
-	if incoming {
-		dp.info.Dir = peer.DirInbound
-	} else {
-		dp.info.Dir = peer.DirOutbound
-	}
-	return dp
-}
-
-type drpcPeer struct {
-	sc   sec.SecureConn
-	info peer.Info
-	drpc.Stream
-}
-
-func (d *drpcPeer) Id() string {
-	return d.info.Id
-}
-
-func (d *drpcPeer) Info() peer.Info {
-	return d.info
-}
-
-func (d *drpcPeer) Recv() (msg *syncproto.Message, err error) {
-	msg = &syncproto.Message{}
-	if err = d.Stream.MsgRecv(msg, Encoding); err != nil {
-		return
-	}
-	atomic.StoreInt64(&d.info.LastActiveUnix, time.Now().Unix())
-	return
-}
-
-func (d *drpcPeer) Send(msg *syncproto.Message) (err error) {
-	if err = d.Stream.MsgSend(msg, Encoding); err != nil {
-		return
-	}
-	atomic.StoreInt64(&d.info.LastActiveUnix, time.Now().Unix())
-	return
-}

service/node/service.go

@@ -44,7 +44,7 @@ type service struct {
 	nodes []*Node
 }
 
-func (s *service) Init(ctx context.Context, a *app.App) (err error) {
+func (s *service) Init(a *app.App) (err error) {
 	cfg := a.MustComponent(config.CName).(*config.Config)
 	signDecoder := signingkey.NewEDPrivKeyDecoder()
 	rsaDecoder := encryptionkey.NewRSAPrivKeyDecoder()

service/space/remotediff/remotediff.go

@@ -1,125 +0,0 @@
-package remotediff
-
-import (
-	"context"
-	"fmt"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/ldiff"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/pool"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/space/spacesync"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/syncproto"
-)
-
-func NewRemoteDiff(p pool.Pool, peerId, spaceId string) ldiff.Remote {
-	return remote{
-		pool:    p,
-		peerId:  peerId,
-		spaceId: spaceId,
-	}
-}
-
-type remote struct {
-	pool    pool.Pool
-	peerId  string
-	spaceId string
-}
-
-func (r remote) Ranges(ctx context.Context, ranges []ldiff.Range, resBuf []ldiff.RangeResult) (results []ldiff.RangeResult, err error) {
-	results = resBuf[:0]
-	pbRanges := make([]*spacesync.DiffRangeRequestRange, 0, len(ranges))
-	for _, rg := range ranges {
-		pbRanges = append(pbRanges, &spacesync.DiffRangeRequestRange{
-			From:  rg.From,
-			To:    rg.To,
-			Limit: uint32(rg.Limit),
-		})
-	}
-	req := &spacesync.Space{
-		SpaceId: r.spaceId,
-		Message: &spacesync.SpaceContent{
-			Value: &spacesync.SpaceContentValueOfDiffRange{
-				DiffRange: &spacesync.DiffRange{
-					Request: &spacesync.DiffRangeRequest{
-						Ranges: pbRanges,
-					},
-				},
-			},
-		},
-	}
-	msg, err := req.Marshal()
-	if err != nil {
-		return
-	}
-	resp, err := r.pool.SendAndWaitResponse(ctx, r.peerId, &syncproto.Message{
-		Header: &syncproto.Header{
-			Type: syncproto.MessageType_MessageTypeSpace,
-		},
-		Data: msg,
-	})
-	if err != nil {
-		return
-	}
-	var spaceResp = &spacesync.Space{}
-	if err = resp.UnmarshalData(spaceResp); err != nil {
-		return
-	}
-	rangeResp := spaceResp.GetMessage().GetDiffRange().GetResponse()
-	if rangeResp != nil {
-		return nil, fmt.Errorf("got nil response")
-	}
-	for _, rr := range rangeResp.Results {
-		var elms []ldiff.Element
-		if len(rr.Elements) > 0 {
-			elms = make([]ldiff.Element, 0, len(rr.Elements))
-		}
-		results = append(results, ldiff.RangeResult{
-			Hash:     rr.Hash,
-			Elements: elms,
-			Count:    int(rr.Count),
-		})
-	}
-	return
-}
-
-func HandlerRangeRequest(ctx context.Context, d ldiff.Diff, diffRange *spacesync.DiffRange) (resp *spacesync.DiffRange, err error) {
-	req := diffRange.GetRequest()
-	if req != nil {
-		return nil, fmt.Errorf("received nil request")
-	}
-
-	ranges := make([]ldiff.Range, 0, len(req.Ranges))
-	for _, reqRange := range req.Ranges {
-		ranges = append(ranges, ldiff.Range{
-			From:  reqRange.From,
-			To:    reqRange.To,
-			Limit: int(reqRange.Limit),
-		})
-	}
-	res, err := d.Ranges(ctx, ranges, nil)
-	if err != nil {
-		return
-	}
-
-	var rangeResp = &spacesync.DiffRangeResponse{
-		Results: make([]*spacesync.DiffRangeResponseResult, len(res)),
-	}
-	for _, rangeRes := range res {
-		var elements []*spacesync.DiffRangeResponseResultElement
-		if len(rangeRes.Elements) > 0 {
-			elements = make([]*spacesync.DiffRangeResponseResultElement, 0, len(rangeRes.Elements))
-			for _, el := range rangeRes.Elements {
-				elements = append(elements, &spacesync.DiffRangeResponseResultElement{
-					Id:   el.Id,
-					Head: el.Head,
-				})
-			}
-		}
-		rangeResp.Results = append(rangeResp.Results, &spacesync.DiffRangeResponseResult{
-			Hash:     rangeRes.Hash,
-			Elements: elements,
-			Count:    uint32(rangeRes.Count),
-		})
-	}
-	return &spacesync.DiffRange{
-		Response: rangeResp,
-	}, nil
-}

service/space/service.go

@@ -1,93 +0,0 @@
-package space
-
-import (
-	"context"
-	"fmt"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/config"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/ocache"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/configuration"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/pool"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/pool/handler"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/space/spacesync"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/syncproto"
-	"github.com/gogo/protobuf/proto"
-	"time"
-)
-
-const CName = "space"
-
-var log = logger.NewNamed(CName)
-
-func New() Service {
-	return new(service)
-}
-
-type Service interface {
-	handler.ReplyHandler
-	app.ComponentRunnable
-}
-
-type service struct {
-	conf        config.Space
-	cache       ocache.OCache
-	pool        pool.Pool
-	confService configuration.Service
-}
-
-func (s *service) Init(ctx context.Context, a *app.App) (err error) {
-	s.conf = a.MustComponent(config.CName).(*config.Config).Space
-	s.pool = a.MustComponent(pool.CName).(pool.Pool)
-	s.confService = a.MustComponent(configuration.CName).(configuration.Service)
-	ttlSec := time.Second * time.Duration(s.conf.GCTTL)
-	s.cache = ocache.New(s.loadSpace, ocache.WithTTL(ttlSec), ocache.WithGCPeriod(time.Minute))
-	s.pool.AddHandler(syncproto.MessageType_MessageTypeSpace, handler.Reply{ReplyHandler: s}.Handle)
-
-	return nil
-}
-
-func (s *service) Name() (name string) {
-	return CName
-}
-
-func (s *service) Run(ctx context.Context) (err error) {
-	return
-}
-
-func (s *service) loadSpace(ctx context.Context, id string) (value ocache.Object, err error) {
-	// TODO: load from database here
-	sp := &space{s: s, id: id, conf: s.confService.GetLast()}
-	if err = sp.Run(ctx); err != nil {
-		return nil, err
-	}
-	return sp, nil
-}
-
-func (s *service) get(ctx context.Context, id string) (Space, error) {
-	obj, err := s.cache.Get(ctx, id)
-	if err != nil {
-		return nil, err
-	}
-	return obj.(Space), nil
-}
-
-func (s *service) Handle(ctx context.Context, data []byte) (resp proto.Marshaler, err error) {
-	var spaceReq = &spacesync.Space{}
-	if err = spaceReq.Unmarshal(data); err != nil {
-		return
-	}
-	if spaceReq.SpaceId != "" {
-		var sp Space
-		sp, err = s.get(ctx, spaceReq.SpaceId)
-		if err != nil {
-			return
-		}
-		return sp.Handle(ctx, spaceReq)
-	}
-	return nil, fmt.Errorf("unexpected space message")
-}
-
-func (s *service) Close(ctx context.Context) (err error) {
-	return s.cache.Close()
-}

service/space/space.go

@@ -1,152 +0,0 @@
-package space
-
-import (
-	"context"
-	"fmt"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/ldiff"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/configuration"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/space/remotediff"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/space/spacesync"
-	"go.uber.org/zap"
-	"math/rand"
-	"sync"
-	"time"
-)
-
-type Space interface {
-	Id() string
-	Handle(ctx context.Context, msg *spacesync.Space) (repl *spacesync.Space, err error)
-	Close() error
-}
-
-type space struct {
-	id           string
-	conf         configuration.Configuration
-	diff         ldiff.Diff
-	diffHandler  func()
-	syncCtx      context.Context
-	syncCancel   func()
-	syncLoopDone chan struct{}
-	s            *service
-	mu           sync.RWMutex
-}
-
-func (s *space) Id() string {
-	return s.id
-}
-
-func (s *space) Run(ctx context.Context) error {
-	s.diff = ldiff.New(16, 16)
-	s.syncCtx, s.syncCancel = context.WithCancel(context.Background())
-	s.syncLoopDone = make(chan struct{})
-	s.testFill()
-	go s.syncLoop()
-	return nil
-}
-
-func (s *space) testFill() {
-	var n = 1000
-	var els = make([]ldiff.Element, 0, n)
-	rand.Seed(time.Now().UnixNano())
-	for i := 0; i < n; i++ {
-		if rand.Intn(n) > 2 {
-			id := fmt.Sprintf("%s.%d", s.id, n)
-			head := "head." + id
-			if rand.Intn(n) > 2 {
-				head += ".modified"
-			}
-			els = append(els, ldiff.Element{
-				Id:   id,
-				Head: head,
-			})
-		}
-	}
-	s.diff.Set(els...)
-}
-
-func (s *space) Handle(ctx context.Context, msg *spacesync.Space) (repl *spacesync.Space, err error) {
-	if diffRange := msg.GetMessage().GetDiffRange(); diffRange != nil {
-		resp, er := remotediff.HandlerRangeRequest(ctx, s.diff, diffRange)
-		if er != nil {
-			return nil, er
-		}
-		return &spacesync.Space{SpaceId: s.id, Message: &spacesync.SpaceContent{
-			Value: &spacesync.SpaceContentValueOfDiffRange{
-				DiffRange: resp,
-			},
-		}}, nil
-	}
-
-	return nil, fmt.Errorf("unexpected request")
-}
-
-func (s *space) syncLoop() {
-	defer close(s.syncLoopDone)
-	doSync := func() {
-		ctx, cancel := context.WithTimeout(s.syncCtx, time.Minute)
-		defer cancel()
-		if err := s.sync(ctx); err != nil {
-			log.Error("periodic sync error", zap.Error(err), zap.String("spaceId", s.id))
-		}
-	}
-	doSync()
-	if s.s.conf.SyncPeriod > 0 {
-		ticker := time.NewTicker(time.Second * time.Duration(s.s.conf.SyncPeriod))
-		defer ticker.Stop()
-		for {
-			select {
-			case <-s.syncCtx.Done():
-			case <-ticker.C:
-				doSync()
-			}
-		}
-	}
-}
-
-func (s *space) sync(ctx context.Context) error {
-	peerIds, err := s.peerIds(ctx)
-	if err != nil {
-		return err
-	}
-	for _, peerId := range peerIds {
-		if err := s.syncWithPeer(ctx, peerId); err != nil {
-			log.Error("can't sync with peer", zap.String("peer", peerId), zap.Error(err))
-		}
-	}
-	return nil
-}
-
-func (s *space) syncWithPeer(ctx context.Context, peerId string) (err error) {
-	rdiff := remotediff.NewRemoteDiff(s.s.pool, peerId, s.id)
-	newIds, changedIds, removedIds, err := s.diff.Diff(ctx, rdiff)
-	if err != nil {
-		return nil
-	}
-	log.Info("sync done:", zap.Strings("newIds", newIds), zap.Strings("changedIds", changedIds), zap.Strings("removedIds", removedIds))
-	return
-}
-
-func (s *space) peerIds(ctx context.Context) (peerIds []string, err error) {
-	if s.conf.IsResponsible(s.id) {
-		peers, err := s.conf.AllPeers(ctx, s.id)
-		if err != nil {
-			return nil, err
-		}
-		for _, p := range peers {
-			peerIds = append(peerIds, p.Id())
-		}
-	} else {
-		peer, err := s.conf.OnePeer(ctx, s.id)
-		if err != nil {
-			return nil, err
-		}
-		peerIds = append(peerIds, peer.Id())
-	}
-	return
-}
-
-func (s *space) Close() error {
-	s.syncCancel()
-	<-s.syncLoopDone
-	return nil
-}

service/space/spacesync/protos/spacesync.proto

@@ -1,41 +0,0 @@
-syntax = "proto3";
-package anytype;
-option go_package = "service/space/spacesync";
-
-message Space {
-    string spaceId = 1;
-
-    Content message = 2;
-
-    message Content {
-        oneof value {
-            DiffRange diffRange = 1;
-        }
-    }
-}
-
-message DiffRange {
-    Request request = 1;
-    Response response = 2;
-
-    message Request {
-        repeated Range ranges = 1;
-        message Range {
-            uint64 from = 1;
-            uint64 to = 2;
-            uint32 limit = 3;
-        }
-    }
-    message Response {
-        repeated Result results = 1;
-        message Result {
-            bytes hash = 1;
-            repeated Element elements = 2;
-            uint32 count = 3;
-            message Element {
-                string id = 1;
-                string head = 2;
-            }
-        }
-    }
-}

service/storage/service.go

@@ -2,16 +2,10 @@ package storage
 
 import (
 	"context"
-	"fmt"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/etc"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/aclchanges/aclpb"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/list"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/storage"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/testutils/acllistbuilder"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/account"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/node"
 )
 
 var CName = "storage"
@@ -38,10 +32,10 @@ type service struct {
 	importedACLSyncData ImportedACLSyncData
 }
 
-func (s *service) Init(ctx context.Context, a *app.App) (err error) {
+func (s *service) Init(a *app.App) (err error) {
 	s.storageProvider = storage.NewInMemoryTreeStorageProvider()
 	// importing hardcoded acl list, check that the keys there are correct
-	return s.importACLList(a)
+	return nil
 }
 
 func (s *service) Storage(treeId string) (storage.Storage, error) {
@@ -75,62 +69,3 @@ func (s *service) Run(ctx context.Context) (err error) {
 func (s service) Close(ctx context.Context) (err error) {
 	return nil
 }
-
-func (s *service) importACLList(a *app.App) (err error) {
-	path := fmt.Sprintf("%s/%s", etc.Path(), "acl.yml")
-	st, err := acllistbuilder.NewACLListStorageBuilderFromFile(path)
-	if err != nil {
-		return err
-	}
-
-	id, err := st.ID()
-	if err != nil {
-		return err
-	}
-
-	header, err := st.Header()
-	if err != nil {
-		return err
-	}
-
-	// checking that acl list contains all the needed permissions for all our nodes
-	err = s.checkActualNodesPermissions(st, a)
-	if err != nil {
-		return err
-	}
-
-	s.importedACLSyncData = ImportedACLSyncData{
-		Id:      id,
-		Header:  header,
-		Records: st.GetRawRecords(),
-	}
-
-	log.Infof("imported ACLList with id %s", id)
-	return s.storageProvider.AddStorage(id, st)
-}
-
-func (s *service) checkActualNodesPermissions(st *acllistbuilder.ACLListStorageBuilder, a *app.App) error {
-	nodes := a.MustComponent(node.CName).(node.Service)
-	acc := a.MustComponent(account.CName).(account.Service)
-
-	aclList, err := list.BuildACLListWithIdentity(acc.Account(), st)
-	if err != nil {
-		return err
-	}
-
-	state := aclList.ACLState()
-
-	// checking own state
-	if state.GetUserStates()[acc.Account().Identity].Permissions != aclpb.ACLChange_Admin {
-		return fmt.Errorf("own node with signing key %s should be admin", acc.Account().Identity)
-	}
-
-	// checking other nodes' states
-	for _, n := range nodes.Nodes() {
-		if state.GetUserStates()[n.SigningKeyString].Permissions != aclpb.ACLChange_Admin {
-			return fmt.Errorf("other node with signing key %s should be admin", n.SigningKeyString)
-		}
-	}
-
-	return nil
-}

service/sync/message/service.go

@@ -1,124 +1,128 @@
 package message
 
-import (
+import "github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
-	"context"
+
-	"fmt"
+//import (
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
+//	"context"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
+//	"fmt"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/pool"
+//	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/node"
+//	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/sync/requesthandler"
+//	pool2 "github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/pool"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/syncproto"
+//	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/net/pool"
-	"github.com/gogo/protobuf/proto"
+//	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/node"
-	"sync"
+//	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/sync/requesthandler"
-	"time"
+//	"github.com/anytypeio/go-anytype-infrastructure-experiments/syncproto"
-)
+//	"github.com/gogo/protobuf/proto"
+//	"sync"
+//	"time"
+//)
 
 var log = logger.NewNamed("messageservice")
 
 const CName = "MessageService"
 
-type service struct {
+//
-	nodes          []*node.Node
+//type service struct {
-	requestHandler requesthandler.RequestHandler
+//	nodes          []*node.Node
-	pool           pool.Pool
+//	requestHandler requesthandler.RequestHandler
-	sync.RWMutex
+//	pool           pool2.Pool
-}
+//	sync.RWMutex
-
+//}
-func New() app.Component {
+//
-	return &service{}
+//func New() app.Component {
-}
+//	return &service{}
-
+//}
-type Service interface {
+//
-	SendMessageAsync(peerId string, msg *syncproto.Sync) error
+//type Service interface {
-	SendToSpaceAsync(spaceId string, msg *syncproto.Sync) error
+//	SendMessageAsync(peerId string, msg *syncproto.Sync) error
-}
+//	SendToSpaceAsync(spaceId string, msg *syncproto.Sync) error
-
+//}
-func (s *service) Init(ctx context.Context, a *app.App) (err error) {
+//
-	s.requestHandler = a.MustComponent(requesthandler.CName).(requesthandler.RequestHandler)
+//func (s *service) Init(a *app.App) (err error) {
-	s.nodes = a.MustComponent(node.CName).(node.Service).Nodes()
+//	s.requestHandler = a.MustComponent(requesthandler.CName).(requesthandler.RequestHandler)
-	s.pool = a.MustComponent(pool.CName).(pool.Pool)
+//	s.nodes = a.MustComponent(node.CName).(node.Service).Nodes()
-	s.pool.AddHandler(syncproto.MessageType_MessageTypeSync, s.HandleMessage)
+//	s.pool = a.MustComponent(pool2.CName).(pool2.Pool)
-	return nil
+//	s.pool.AddHandler(syncproto.MessageType_MessageTypeSync, s.HandleMessage)
-}
+//	return nil
-
+//}
-func (s *service) Name() (name string) {
+//
-	return CName
+//func (s *service) Name() (name string) {
-}
+//	return CName
-
+//}
-func (s *service) Run(ctx context.Context) (err error) {
+//
-	return nil
+//func (s *service) Run(ctx context.Context) (err error) {
-}
+//	return nil
-
+//}
-func (s *service) Close(ctx context.Context) (err error) {
+//
-	return nil
+//func (s *service) Close(ctx context.Context) (err error) {
-}
+//	return nil
-
+//}
-func (s *service) HandleMessage(ctx context.Context, msg *pool.Message) (err error) {
+//
-	defer func() {
+//func (s *service) HandleMessage(ctx context.Context, msg *pool.Message) (err error) {
-		if err != nil {
+//	defer func() {
-			msg.AckError(syncproto.SystemError_UNKNOWN, err.Error())
+//		if err != nil {
-		} else {
+//			msg.AckError(syncproto.System_Error_UNKNOWN, err.Error())
-			msg.Ack()
+//		} else {
-		}
+//			msg.Ack()
-	}()
+//		}
-
+//	}()
-	syncMsg := &syncproto.Sync{}
+//
-	err = proto.Unmarshal(msg.Data, syncMsg)
+//	syncMsg := &syncproto.Sync{}
-	if err != nil {
+//	err = proto.Unmarshal(msg.Data, syncMsg)
-		return
+//	if err != nil {
-	}
+//		return
-
+//	}
-	timeoutCtx, cancel := context.WithTimeout(ctx, time.Second*30)
+//
-	defer cancel()
+//	timeoutCtx, cancel := context.WithTimeout(ctx, time.Second*30)
-	err = s.requestHandler.HandleSyncMessage(timeoutCtx, msg.Peer().Id(), syncMsg)
+//	defer cancel()
-	return
+//	err = s.requestHandler.HandleSyncMessage(timeoutCtx, msg.Peer().Id(), syncMsg)
-}
+//	return
-
+//}
-func (s *service) SendMessageAsync(peerId string, msg *syncproto.Sync) (err error) {
+//
-	_, err = s.pool.DialAndAddPeer(context.Background(), peerId)
+//func (s *service) SendMessageAsync(peerId string, msg *syncproto.Sync) (err error) {
-	if err != nil {
+//	_, err = s.pool.DialAndAddPeer(context.Background(), peerId)
-		return
+//	if err != nil {
-	}
+//		return
-
+//	}
-	marshalled, err := proto.Marshal(msg)
+//
-	if err != nil {
+//	marshalled, err := proto.Marshal(msg)
-		return
+//	if err != nil {
-	}
+//		return
-
+//	}
-	go s.sendAsync(peerId, msgInfo(msg), marshalled)
+//
-	return
+//	go s.sendAsync(peerId, msgInfo(msg), marshalled)
-}
+//	return
-
+//}
-func (s *service) SendToSpaceAsync(spaceId string, msg *syncproto.Sync) error {
+//
-	for _, rp := range s.nodes {
+//func (s *service) SendToSpaceAsync(spaceId string, msg *syncproto.Sync) error {
-		s.SendMessageAsync(rp.PeerId, msg)
+//	for _, rp := range s.nodes {
-	}
+//		s.SendMessageAsync(rp.PeerId, msg)
-	return nil
+//	}
-}
+//	return nil
-
+//}
-func (s *service) sendAsync(peerId string, msgTypeStr string, marshalled []byte) error {
+//
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
+//func (s *service) sendAsync(peerId string, msgTypeStr string, marshalled []byte) error {
-	defer cancel()
+//	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
-	return s.pool.SendAndWait(ctx, peerId, &syncproto.Message{
+//	defer cancel()
-		Header: &syncproto.Header{
+//	return s.pool.SendAndWait(ctx, peerId, &syncproto.Message{
-			Type:      syncproto.MessageType_MessageTypeSync,
+//		Header: &syncproto.Header{
-			DebugInfo: msgTypeStr,
+//			Type:      syncproto.MessageType_MessageTypeSync,
-		},
+//			DebugInfo: msgTypeStr,
-		Data: marshalled,
+//		},
-	})
+//		Data: marshalled,
-}
+//	})
-
+//}
-func msgInfo(content *syncproto.Sync) (syncMethod string) {
+//
-	msg := content.GetMessage()
+//func msgInfo(content *syncproto.Sync) (syncMethod string) {
-	switch {
+//	msg := content.GetMessage()
-	case msg.GetFullSyncRequest() != nil:
+//	switch {
-		syncMethod = "FullSyncRequest"
+//	case msg.GetFullSyncRequest() != nil:
-	case msg.GetFullSyncResponse() != nil:
+//		syncMethod = "FullSyncRequest"
-		syncMethod = "FullSyncResponse"
+//	case msg.GetFullSyncResponse() != nil:
-	case msg.GetHeadUpdate() != nil:
+//		syncMethod = "FullSyncResponse"
-		syncMethod = "HeadUpdate"
+//	case msg.GetHeadUpdate() != nil:
-	}
+//		syncMethod = "HeadUpdate"
-	syncMethod = fmt.Sprintf("method: %s, treeType: %s", syncMethod, content.TreeHeader.DocType.String())
+//	}
-	return
+//	syncMethod = fmt.Sprintf("method: %s, treeType: %s", syncMethod, content.TreeHeader.DocType.String())
-}
+//	return
+//}

service/sync/requesthandler/requesthandler.go

@@ -4,10 +4,10 @@ import (
 	"context"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/account"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/aclchanges/aclpb"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/storage"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/tree"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/account"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/treecache"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/syncproto"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/util/slice"
@@ -37,7 +37,7 @@ type MessageSender interface {
 
 const CName = "SyncRequestHandler"
 
-func (r *requestHandler) Init(ctx context.Context, a *app.App) (err error) {
+func (r *requestHandler) Init(a *app.App) (err error) {
 	r.treeCache = a.MustComponent(treecache.CName).(treecache.Service)
 	r.account = a.MustComponent(account.CName).(account.Service)
 	r.messageService = a.MustComponent("MessageService").(MessageSender)

service/treecache/service.go

@@ -5,12 +5,12 @@ import (
 	"fmt"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/account"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/aclchanges/aclpb"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/list"
 	aclstorage "github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/storage"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/acl/tree"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/ocache"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/account"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/storage"
 	"go.uber.org/zap"
 )
@@ -74,7 +74,7 @@ func (s *service) Add(ctx context.Context, treeId string, payload any) error {
 	return nil
 }
 
-func (s *service) Init(ctx context.Context, a *app.App) (err error) {
+func (s *service) Init(a *app.App) (err error) {
 	s.cache = ocache.New(s.loadTree)
 	s.account = a.MustComponent(account.CName).(account.Service)
 	s.storage = a.MustComponent(storage.CName).(storage.Service)