Add auxiliary methods to acl for getting records, export peer manager

commonspace/object/acl/list/list.go

@@ -49,6 +49,7 @@ type AclList interface {
 	Head() *AclRecord
 
 	RecordsAfter(ctx context.Context, id string) (records []*consensusproto.RawRecordWithId, err error)
+	RecordsBefore(ctx context.Context, headId string) (records []*consensusproto.RawRecordWithId, err error)
 	Get(id string) (*AclRecord, error)
 	GetIndex(idx int) (*AclRecord, error)
 	Iterate(iterFunc IterFunc)
@@ -305,6 +306,24 @@ func (a *aclList) RecordsAfter(ctx context.Context, id string) (records []*conse
 	return
 }
 
+func (a *aclList) RecordsBefore(ctx context.Context, headId string) (records []*consensusproto.RawRecordWithId, err error) {
+	if headId == "" {
+		headId = a.Head().Id
+	}
+	recIdx, ok := a.indexes[headId]
+	if !ok {
+		return nil, ErrNoSuchRecord
+	}
+	for i := 0; i <= recIdx; i++ {
+		rawRec, err := a.storage.GetRawRecord(ctx, a.records[i].Id)
+		if err != nil {
+			return nil, err
+		}
+		records = append(records, rawRec)
+	}
+	return
+}
+
 func (a *aclList) IterateFrom(startId string, iterFunc IterFunc) {
 	recIdx, ok := a.indexes[startId]
 	if !ok {

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

@@ -297,6 +297,21 @@ func (mr *MockAclListMockRecorder) RecordsAfter(arg0, arg1 interface{}) *gomock.
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecordsAfter", reflect.TypeOf((*MockAclList)(nil).RecordsAfter), arg0, arg1)
 }
 
+// RecordsBefore mocks base method.
+func (m *MockAclList) RecordsBefore(arg0 context.Context, arg1 string) ([]*consensusproto.RawRecordWithId, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "RecordsBefore", arg0, arg1)
+	ret0, _ := ret[0].([]*consensusproto.RawRecordWithId)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// RecordsBefore indicates an expected call of RecordsBefore.
+func (mr *MockAclListMockRecorder) RecordsBefore(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecordsBefore", reflect.TypeOf((*MockAclList)(nil).RecordsBefore), arg0, arg1)
+}
+
 // Root mocks base method.
 func (m *MockAclList) Root() *consensusproto.RawRecordWithId {
 	m.ctrl.T.Helper()

commonspace/object/acl/syncacl/mock_syncacl/mock_syncacl.go

@@ -357,6 +357,21 @@ func (mr *MockSyncAclMockRecorder) RecordsAfter(arg0, arg1 interface{}) *gomock.
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecordsAfter", reflect.TypeOf((*MockSyncAcl)(nil).RecordsAfter), arg0, arg1)
 }
 
+// RecordsBefore mocks base method.
+func (m *MockSyncAcl) RecordsBefore(arg0 context.Context, arg1 string) ([]*consensusproto.RawRecordWithId, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "RecordsBefore", arg0, arg1)
+	ret0, _ := ret[0].([]*consensusproto.RawRecordWithId)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// RecordsBefore indicates an expected call of RecordsBefore.
+func (mr *MockSyncAclMockRecorder) RecordsBefore(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecordsBefore", reflect.TypeOf((*MockSyncAcl)(nil).RecordsBefore), arg0, arg1)
+}
+
 // Root mocks base method.
 func (m *MockSyncAcl) Root() *consensusproto.RawRecordWithId {
 	m.ctrl.T.Helper()

commonspace/object/tree/synctree/utils_test.go

@@ -157,6 +157,10 @@ func (r *requestPeerManager) GetResponsiblePeers(ctx context.Context) (peers []p
 	return nil, nil
 }
 
+func (r *requestPeerManager) GetNodePeers(ctx context.Context) (peers []peer.Peer, err error) {
+	return nil, nil
+}
+
 // testSyncHandler is the wrapper around individual tree to test sync protocol
 type testSyncHandler struct {
 	synchandler.SyncHandler

commonspace/peermanager/mock_peermanager/mock_peermanager.go

@@ -51,6 +51,21 @@ func (mr *MockPeerManagerMockRecorder) Broadcast(arg0, arg1 interface{}) *gomock
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Broadcast", reflect.TypeOf((*MockPeerManager)(nil).Broadcast), arg0, arg1)
 }
 
+// GetNodePeers mocks base method.
+func (m *MockPeerManager) GetNodePeers(arg0 context.Context) ([]peer.Peer, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetNodePeers", arg0)
+	ret0, _ := ret[0].([]peer.Peer)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetNodePeers indicates an expected call of GetNodePeers.
+func (mr *MockPeerManagerMockRecorder) GetNodePeers(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNodePeers", reflect.TypeOf((*MockPeerManager)(nil).GetNodePeers), arg0)
+}
+
 // GetResponsiblePeers mocks base method.
 func (m *MockPeerManager) GetResponsiblePeers(arg0 context.Context) ([]peer.Peer, error) {
 	m.ctrl.T.Helper()

commonspace/peermanager/peermanager.go

@@ -18,8 +18,10 @@ type PeerManager interface {
 	SendPeer(ctx context.Context, peerId string, msg *spacesyncproto.ObjectSyncMessage) (err error)
 	// Broadcast sends a message to all subscribed peers
 	Broadcast(ctx context.Context, msg *spacesyncproto.ObjectSyncMessage) (err error)
-	// GetResponsiblePeers dials or gets from cache responsible peers to unary operations
+	// GetResponsiblePeers dials or gets from cache responsible peers
 	GetResponsiblePeers(ctx context.Context) (peers []peer.Peer, err error)
+	// GetNodePeers dials or gets from cache node peers
+	GetNodePeers(ctx context.Context) (peers []peer.Peer, err error)
 }
 
 type PeerManagerProvider interface {

commonspace/space.go

@@ -9,11 +9,13 @@ import (
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	"github.com/anyproto/any-sync/commonspace/objectsync"
 	"github.com/anyproto/any-sync/commonspace/objecttreebuilder"
+	"github.com/anyproto/any-sync/commonspace/peermanager"
 	"github.com/anyproto/any-sync/commonspace/settings"
 	"github.com/anyproto/any-sync/commonspace/spacestate"
 	"github.com/anyproto/any-sync/commonspace/spacestorage"
 	"github.com/anyproto/any-sync/commonspace/spacesyncproto"
 	"github.com/anyproto/any-sync/commonspace/syncstatus"
+	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/util/crypto"
 	"go.uber.org/zap"
 	"strconv"
@@ -73,6 +75,8 @@ type Space interface {
 	SpaceDeleteRawChange(ctx context.Context) (raw *treechangeproto.RawTreeChangeWithId, err error)
 	DeleteSpace(ctx context.Context, deleteChange *treechangeproto.RawTreeChangeWithId) (err error)
 
+	GetNodePeers(ctx context.Context) (peer []peer.Peer, err error)
+
 	HandleMessage(ctx context.Context, msg objectsync.HandleMessage) (err error)
 	HandleSyncRequest(ctx context.Context, req *spacesyncproto.ObjectSyncMessage) (resp *spacesyncproto.ObjectSyncMessage, err error)
 	HandleRangeRequest(ctx context.Context, req *spacesyncproto.HeadSyncRequest) (resp *spacesyncproto.HeadSyncResponse, err error)
@@ -89,6 +93,7 @@ type space struct {
 	app   *app.App
 
 	treeBuilder objecttreebuilder.TreeBuilderComponent
+	peerManager peermanager.PeerManager
 	headSync    headsync.HeadSync
 	objectSync  objectsync.ObjectSync
 	syncStatus  syncstatus.StatusService
@@ -154,6 +159,10 @@ func (s *space) TreeBuilder() objecttreebuilder.TreeBuilder {
 	return s.treeBuilder
 }
 
+func (s *space) GetNodePeers(ctx context.Context) (peer []peer.Peer, err error) {
+	return s.peerManager.GetNodePeers(ctx)
+}
+
 func (s *space) Acl() list.AclList {
 	return s.aclList
 }
@@ -173,6 +182,7 @@ func (s *space) Init(ctx context.Context) (err error) {
 	s.settings = s.app.MustComponent(settings.CName).(settings.Settings)
 	s.objectSync = s.app.MustComponent(objectsync.CName).(objectsync.ObjectSync)
 	s.storage = s.app.MustComponent(spacestorage.CName).(spacestorage.SpaceStorage)
+	s.peerManager = s.app.MustComponent(peermanager.CName).(peermanager.PeerManager)
 	s.aclList = s.app.MustComponent(syncacl.CName).(list.AclList)
 	s.header, err = s.storage.SpaceHeader()
 	return

commonspace/spacesyncproto/mock_spacesyncproto/mock_spacesyncproto.go

@@ -36,6 +36,36 @@ func (m *MockDRPCSpaceSyncClient) EXPECT() *MockDRPCSpaceSyncClientMockRecorder
 	return m.recorder
 }
 
+// AclAddRecord mocks base method.
+func (m *MockDRPCSpaceSyncClient) AclAddRecord(arg0 context.Context, arg1 *spacesyncproto.AclAddRecordRequest) (*spacesyncproto.AclAddRecordResponse, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "AclAddRecord", arg0, arg1)
+	ret0, _ := ret[0].(*spacesyncproto.AclAddRecordResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// AclAddRecord indicates an expected call of AclAddRecord.
+func (mr *MockDRPCSpaceSyncClientMockRecorder) AclAddRecord(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AclAddRecord", reflect.TypeOf((*MockDRPCSpaceSyncClient)(nil).AclAddRecord), arg0, arg1)
+}
+
+// AclGetRecords mocks base method.
+func (m *MockDRPCSpaceSyncClient) AclGetRecords(arg0 context.Context, arg1 *spacesyncproto.AclGetRecordsRequest) (*spacesyncproto.AclGetRecordsResponse, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "AclGetRecords", arg0, arg1)
+	ret0, _ := ret[0].(*spacesyncproto.AclGetRecordsResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// AclGetRecords indicates an expected call of AclGetRecords.
+func (mr *MockDRPCSpaceSyncClientMockRecorder) AclGetRecords(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AclGetRecords", reflect.TypeOf((*MockDRPCSpaceSyncClient)(nil).AclGetRecords), arg0, arg1)
+}
+
 // DRPCConn mocks base method.
 func (m *MockDRPCSpaceSyncClient) DRPCConn() drpc.Conn {
 	m.ctrl.T.Helper()

commonspace/spacesyncproto/protos/spacesync.proto

@@ -11,6 +11,7 @@ enum ErrCodes {
     SpaceIsDeleted = 4;
     PeerIsNotResponsible = 5;
     ReceiptIsInvalid = 6;
+    InvalidPayload = 7;
     ErrorOffset = 100;
 }
 
@@ -25,6 +26,10 @@ service SpaceSync {
     rpc ObjectSyncStream(stream ObjectSyncMessage) returns (stream ObjectSyncMessage);
     // ObjectSync sends object sync message and synchronously gets response message
     rpc ObjectSync(ObjectSyncMessage) returns (ObjectSyncMessage);
+    // AclAddRecord adds a new record to acl log. Works only with any-sync-node
+    rpc AclAddRecord(AclAddRecordRequest) returns (AclAddRecordResponse);
+    // AclGetRecords gets acl records
+    rpc AclGetRecords(AclGetRecordsRequest) returns (AclGetRecordsResponse);
 }
 
 // HeadSyncRange presenting a request for one range
@@ -157,3 +162,27 @@ message SpaceSubscription {
     repeated string spaceIds = 1;
     SpaceSubscriptionAction action = 2;
 }
+
+// AclAddRecordRequest contains marshaled consensusproto.RawRecord
+message AclAddRecordRequest {
+    string spaceId = 1;
+    bytes payload = 2;
+}
+
+// AclAddRecordResponse contains created record id and marshaled consensusproto.RawRecord
+message AclAddRecordResponse {
+    string recordId = 1;
+    bytes payload = 2;
+}
+
+// AclGetRecordsRequest can optionally contain the last known aclHeal, the server will return only new records or an empty list if there are no new records.
+// If aclHead is not provided the whole list will be returned.
+message AclGetRecordsRequest {
+    string spaceId = 1;
+    string aclHead = 2;
+}
+
+// AclGetRecordsResponse contains list of marshaled consensusproto.RawRecordWithId
+message AclGetRecordsResponse {
+    repeated bytes records = 1;
+}

commonspace/spacesyncproto/spacesync_drpc.pb.go

@@ -45,6 +45,8 @@ type DRPCSpaceSyncClient interface {
 	SpacePull(ctx context.Context, in *SpacePullRequest) (*SpacePullResponse, error)
 	ObjectSyncStream(ctx context.Context) (DRPCSpaceSync_ObjectSyncStreamClient, error)
 	ObjectSync(ctx context.Context, in *ObjectSyncMessage) (*ObjectSyncMessage, error)
+	AclAddRecord(ctx context.Context, in *AclAddRecordRequest) (*AclAddRecordResponse, error)
+	AclGetRecords(ctx context.Context, in *AclGetRecordsRequest) (*AclGetRecordsResponse, error)
 }
 
 type drpcSpaceSyncClient struct {
@@ -132,12 +134,32 @@ func (c *drpcSpaceSyncClient) ObjectSync(ctx context.Context, in *ObjectSyncMess
 	return out, nil
 }
 
+func (c *drpcSpaceSyncClient) AclAddRecord(ctx context.Context, in *AclAddRecordRequest) (*AclAddRecordResponse, error) {
+	out := new(AclAddRecordResponse)
+	err := c.cc.Invoke(ctx, "/spacesync.SpaceSync/AclAddRecord", drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}, in, out)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *drpcSpaceSyncClient) AclGetRecords(ctx context.Context, in *AclGetRecordsRequest) (*AclGetRecordsResponse, error) {
+	out := new(AclGetRecordsResponse)
+	err := c.cc.Invoke(ctx, "/spacesync.SpaceSync/AclGetRecords", drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}, in, out)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 type DRPCSpaceSyncServer interface {
 	HeadSync(context.Context, *HeadSyncRequest) (*HeadSyncResponse, error)
 	SpacePush(context.Context, *SpacePushRequest) (*SpacePushResponse, error)
 	SpacePull(context.Context, *SpacePullRequest) (*SpacePullResponse, error)
 	ObjectSyncStream(DRPCSpaceSync_ObjectSyncStreamStream) error
 	ObjectSync(context.Context, *ObjectSyncMessage) (*ObjectSyncMessage, error)
+	AclAddRecord(context.Context, *AclAddRecordRequest) (*AclAddRecordResponse, error)
+	AclGetRecords(context.Context, *AclGetRecordsRequest) (*AclGetRecordsResponse, error)
 }
 
 type DRPCSpaceSyncUnimplementedServer struct{}
@@ -162,9 +184,17 @@ func (s *DRPCSpaceSyncUnimplementedServer) ObjectSync(context.Context, *ObjectSy
 	return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
 }
 
+func (s *DRPCSpaceSyncUnimplementedServer) AclAddRecord(context.Context, *AclAddRecordRequest) (*AclAddRecordResponse, error) {
+	return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
+}
+
+func (s *DRPCSpaceSyncUnimplementedServer) AclGetRecords(context.Context, *AclGetRecordsRequest) (*AclGetRecordsResponse, error) {
+	return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
+}
+
 type DRPCSpaceSyncDescription struct{}
 
-func (DRPCSpaceSyncDescription) NumMethods() int { return 5 }
+func (DRPCSpaceSyncDescription) NumMethods() int { return 7 }
 
 func (DRPCSpaceSyncDescription) Method(n int) (string, drpc.Encoding, drpc.Receiver, interface{}, bool) {
 	switch n {
@@ -212,6 +242,24 @@ func (DRPCSpaceSyncDescription) Method(n int) (string, drpc.Encoding, drpc.Recei
 						in1.(*ObjectSyncMessage),
 					)
 			}, DRPCSpaceSyncServer.ObjectSync, true
+	case 5:
+		return "/spacesync.SpaceSync/AclAddRecord", drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{},
+			func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
+				return srv.(DRPCSpaceSyncServer).
+					AclAddRecord(
+						ctx,
+						in1.(*AclAddRecordRequest),
+					)
+			}, DRPCSpaceSyncServer.AclAddRecord, true
+	case 6:
+		return "/spacesync.SpaceSync/AclGetRecords", drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{},
+			func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
+				return srv.(DRPCSpaceSyncServer).
+					AclGetRecords(
+						ctx,
+						in1.(*AclGetRecordsRequest),
+					)
+			}, DRPCSpaceSyncServer.AclGetRecords, true
 	default:
 		return "", nil, nil, nil, false
 	}
@@ -310,3 +358,35 @@ func (x *drpcSpaceSync_ObjectSyncStream) SendAndClose(m *ObjectSyncMessage) erro
 	}
 	return x.CloseSend()
 }
+
+type DRPCSpaceSync_AclAddRecordStream interface {
+	drpc.Stream
+	SendAndClose(*AclAddRecordResponse) error
+}
+
+type drpcSpaceSync_AclAddRecordStream struct {
+	drpc.Stream
+}
+
+func (x *drpcSpaceSync_AclAddRecordStream) SendAndClose(m *AclAddRecordResponse) error {
+	if err := x.MsgSend(m, drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}); err != nil {
+		return err
+	}
+	return x.CloseSend()
+}
+
+type DRPCSpaceSync_AclGetRecordsStream interface {
+	drpc.Stream
+	SendAndClose(*AclGetRecordsResponse) error
+}
+
+type drpcSpaceSync_AclGetRecordsStream struct {
+	drpc.Stream
+}
+
+func (x *drpcSpaceSync_AclGetRecordsStream) SendAndClose(m *AclGetRecordsResponse) error {
+	if err := x.MsgSend(m, drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}); err != nil {
+		return err
+	}
+	return x.CloseSend()
+}

commonspace/spaceutils_test.go

@@ -151,6 +151,10 @@ func (p *mockPeerManager) GetResponsiblePeers(ctx context.Context) (peers []peer
 	return nil, nil
 }
 
+func (p *mockPeerManager) GetNodePeers(ctx context.Context) (peers []peer.Peer, err error) {
+	return nil, nil
+}
+
 //
 // Mock PeerManagerProvider
 //