Add synctest and simple sync test infra + upgrade drpc to v0.0.34

Makefile

@@ -25,6 +25,7 @@ proto:
 	protoc --gogofaster_out=:. net/secureservice/handshake/handshakeproto/protos/*.proto
 	protoc --gogofaster_out=$(PKGMAP):. --go-drpc_out=protolib=github.com/gogo/protobuf:. coordinator/coordinatorproto/protos/*.proto
 	protoc --gogofaster_out=:. --go-drpc_out=protolib=github.com/gogo/protobuf:. consensus/consensusproto/protos/*.proto
+	protoc --gogofaster_out=:. --go-drpc_out=protolib=github.com/gogo/protobuf:. commonspace/sync/synctestproto/protos/*.proto
 	protoc --gogofaster_out=:. --go-drpc_out=protolib=github.com/gogo/protobuf:. identityrepo/identityrepoproto/protos/*.proto
 	protoc --gogofaster_out=:. --go-drpc_out=protolib=github.com/gogo/protobuf:. nameservice/nameserviceproto/protos/*.proto
 	protoc --gogofaster_out=:. --go-drpc_out=protolib=github.com/gogo/protobuf:. paymentservice/paymentserviceproto/protos/*.proto

commonfile/fileproto/file_drpc.pb.go

@@ -1,5 +1,5 @@
 // Code generated by protoc-gen-go-drpc. DO NOT EDIT.
-// protoc-gen-go-drpc version: v0.0.33
+// protoc-gen-go-drpc version: v0.0.34
 // source: commonfile/fileproto/protos/file.proto
 
 package fileproto

commonspace/spacesyncproto/spacesync_drpc.pb.go

@@ -1,5 +1,5 @@
 // Code generated by protoc-gen-go-drpc. DO NOT EDIT.
-// protoc-gen-go-drpc version: v0.0.33
+// protoc-gen-go-drpc version: v0.0.34
 // source: commonspace/spacesyncproto/protos/spacesync.proto
 
 package spacesyncproto

commonspace/sync/sync_test.go

@@ -0,0 +1,51 @@
+package sync
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"storj.io/drpc"
+
+	"github.com/anyproto/any-sync/app"
+	"github.com/anyproto/any-sync/commonspace/sync/synctest"
+	"github.com/anyproto/any-sync/commonspace/sync/synctestproto"
+	"github.com/anyproto/any-sync/net/rpc/rpctest"
+)
+
+var ctx = context.Background()
+
+func TestNewSyncService(t *testing.T) {
+	connProvider := synctest.NewConnProvider()
+	var (
+		firstApp  = &app.App{}
+		secondApp = &app.App{}
+	)
+	firstApp.Register(connProvider).
+		Register(rpctest.NewTestServer()).
+		Register(synctest.NewRpcServer()).
+		Register(synctest.NewPeerProvider("first"))
+	secondApp.Register(connProvider).
+		Register(rpctest.NewTestServer()).
+		Register(synctest.NewRpcServer()).
+		Register(synctest.NewPeerProvider("second"))
+	require.NoError(t, firstApp.Start(ctx))
+	require.NoError(t, secondApp.Start(ctx))
+	pr1, err := firstApp.Component(synctest.PeerName).(*synctest.PeerProvider).GetPeer("second")
+	require.NoError(t, err)
+	err = pr1.DoDrpc(ctx, func(conn drpc.Conn) error {
+		cl := synctestproto.NewDRPCCounterSyncClient(conn)
+		_, err := cl.CounterStreamRequest(ctx, &synctestproto.CounterRequest{ObjectId: "test1"})
+		require.NoError(t, err)
+		return nil
+	})
+	pr2, err := secondApp.Component(synctest.PeerName).(*synctest.PeerProvider).GetPeer("first")
+	require.NoError(t, err)
+	err = pr2.DoDrpc(ctx, func(conn drpc.Conn) error {
+		cl := synctestproto.NewDRPCCounterSyncClient(conn)
+		_, err := cl.CounterStreamRequest(ctx, &synctestproto.CounterRequest{ObjectId: "test2"})
+		require.NoError(t, err)
+		return nil
+	})
+	require.NoError(t, err)
+}

commonspace/sync/synctest/connprovider.go

@@ -0,0 +1,59 @@
+package synctest
+
+import (
+	"sync"
+
+	"github.com/anyproto/any-sync/app"
+	"github.com/anyproto/any-sync/net/rpc/rpctest"
+	"github.com/anyproto/any-sync/net/transport"
+)
+
+const ConnName = "connprovider"
+
+type ConnProvider struct {
+	sync.Mutex
+	multiConns map[string]transport.MultiConn
+	providers  map[string]*PeerProvider
+}
+
+func (c *ConnProvider) Init(a *app.App) (err error) {
+	return
+}
+
+func (c *ConnProvider) Name() (name string) {
+	return ConnName
+}
+
+func (c *ConnProvider) Observe(provider *PeerProvider, peerId string) {
+	c.Lock()
+	defer c.Unlock()
+	c.providers[peerId] = provider
+}
+
+func (c *ConnProvider) GetConn(firstId, secondId string) (conn transport.MultiConn) {
+	c.Lock()
+	defer c.Unlock()
+	id := mapId(firstId, secondId)
+	if conn, ok := c.multiConns[id]; ok {
+		return conn
+	}
+	first, second := rpctest.MultiConnPair(firstId, secondId)
+	c.multiConns[id] = first
+	c.multiConns[mapId(secondId, firstId)] = second
+	err := c.providers[secondId].StartPeer(secondId, second)
+	if err != nil {
+		panic(err)
+	}
+	return first
+}
+
+func NewConnProvider() *ConnProvider {
+	return &ConnProvider{
+		multiConns: make(map[string]transport.MultiConn),
+		providers:  make(map[string]*PeerProvider),
+	}
+}
+
+func mapId(firstId, secondId string) string {
+	return firstId + "-" + secondId
+}

commonspace/sync/synctest/peerprovider.go

@@ -0,0 +1,57 @@
+package synctest
+
+import (
+	"sync"
+
+	"github.com/anyproto/any-sync/app"
+	"github.com/anyproto/any-sync/net/peer"
+	"github.com/anyproto/any-sync/net/rpc/rpctest"
+	"github.com/anyproto/any-sync/net/rpc/server"
+	"github.com/anyproto/any-sync/net/transport"
+)
+
+const PeerName = "peerprovider"
+
+type PeerProvider struct {
+	sync.Mutex
+	myPeer       string
+	peers        map[string]peer.Peer
+	connProvider *ConnProvider
+	server       *rpctest.TestServer
+}
+
+func (c *PeerProvider) Init(a *app.App) (err error) {
+	c.connProvider = a.MustComponent(ConnName).(*ConnProvider)
+	c.server = a.MustComponent(server.CName).(*rpctest.TestServer)
+	c.connProvider.Observe(c, c.myPeer)
+	return
+}
+
+func (c *PeerProvider) Name() (name string) {
+	return PeerName
+}
+
+func (c *PeerProvider) StartPeer(peerId string, conn transport.MultiConn) (err error) {
+	c.Lock()
+	defer c.Unlock()
+	c.peers[peerId], err = peer.NewPeer(conn, c.server)
+	return err
+}
+
+func (c *PeerProvider) GetPeer(peerId string) (pr peer.Peer, err error) {
+	c.Lock()
+	defer c.Unlock()
+	if pr, ok := c.peers[peerId]; ok {
+		return pr, nil
+	}
+	conn := c.connProvider.GetConn(c.myPeer, peerId)
+	c.peers[peerId], err = peer.NewPeer(conn, c.server)
+	if err != nil {
+		return nil, err
+	}
+	return c.peers[peerId], nil
+}
+
+func NewPeerProvider(myPeer string) *PeerProvider {
+	return &PeerProvider{myPeer: myPeer, peers: make(map[string]peer.Peer)}
+}

commonspace/sync/synctest/rpcserver.go

@@ -0,0 +1,38 @@
+package synctest
+
+import (
+	"fmt"
+
+	"github.com/anyproto/any-sync/net/rpc/rpctest"
+	"github.com/anyproto/any-sync/net/rpc/server"
+
+	"github.com/anyproto/any-sync/app"
+	"github.com/anyproto/any-sync/commonspace/sync/synctestproto"
+)
+
+const RpcName = "rpcserver"
+
+type RpcServer struct {
+}
+
+func NewRpcServer() *RpcServer {
+	return &RpcServer{}
+}
+
+func (r *RpcServer) CounterStreamRequest(request *synctestproto.CounterRequest, stream synctestproto.DRPCCounterSync_CounterStreamRequestStream) error {
+	fmt.Println(request.ObjectId)
+	return nil
+}
+
+func (r *RpcServer) CounterStream(request *synctestproto.CounterRequest, stream synctestproto.DRPCCounterSync_CounterStreamStream) error {
+	return nil
+}
+
+func (r *RpcServer) Init(a *app.App) (err error) {
+	serv := a.MustComponent(server.CName).(*rpctest.TestServer)
+	return synctestproto.DRPCRegisterCounterSync(serv, r)
+}
+
+func (r *RpcServer) Name() (name string) {
+	return RpcName
+}

commonspace/sync/synctest/syncapp.go

@@ -0,0 +1,7 @@
+package synctest
+
+import "github.com/anyproto/any-sync/app"
+
+type SyncApp struct {
+	app *app.App
+}

commonspace/sync/synctestproto/protos/synctest.proto

@@ -0,0 +1,18 @@
+syntax = "proto3";
+package synctest;
+option go_package = "commonspace/sync/synctestproto";
+
+message CounterIncrease {
+    int32 value = 1;
+    string objectId = 2;
+}
+
+message CounterRequest {
+    repeated int32 existingValues = 1;
+    string objectId = 2;
+}
+
+service CounterSync {
+    rpc CounterStreamRequest(CounterRequest) returns (stream CounterIncrease);
+    rpc CounterStream(CounterRequest) returns (stream CounterIncrease);
+}

commonspace/sync/synctestproto/synctest.pb.go

@@ -0,0 +1,637 @@
+// Code generated by protoc-gen-gogo. DO NOT EDIT.
+// source: commonspace/sync/synctestproto/protos/synctest.proto
+
+package synctestproto
+
+import (
+	fmt "fmt"
+	proto "github.com/gogo/protobuf/proto"
+	io "io"
+	math "math"
+	math_bits "math/bits"
+)
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.GoGoProtoPackageIsVersion3 // please upgrade the proto package
+
+type CounterIncrease struct {
+	Value    int32  `protobuf:"varint,1,opt,name=value,proto3" json:"value,omitempty"`
+	ObjectId string `protobuf:"bytes,2,opt,name=objectId,proto3" json:"objectId,omitempty"`
+}
+
+func (m *CounterIncrease) Reset()         { *m = CounterIncrease{} }
+func (m *CounterIncrease) String() string { return proto.CompactTextString(m) }
+func (*CounterIncrease) ProtoMessage()    {}
+func (*CounterIncrease) Descriptor() ([]byte, []int) {
+	return fileDescriptor_dd5c22b15d7f69e4, []int{0}
+}
+func (m *CounterIncrease) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *CounterIncrease) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_CounterIncrease.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *CounterIncrease) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_CounterIncrease.Merge(m, src)
+}
+func (m *CounterIncrease) XXX_Size() int {
+	return m.Size()
+}
+func (m *CounterIncrease) XXX_DiscardUnknown() {
+	xxx_messageInfo_CounterIncrease.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_CounterIncrease proto.InternalMessageInfo
+
+func (m *CounterIncrease) GetValue() int32 {
+	if m != nil {
+		return m.Value
+	}
+	return 0
+}
+
+func (m *CounterIncrease) GetObjectId() string {
+	if m != nil {
+		return m.ObjectId
+	}
+	return ""
+}
+
+type CounterRequest struct {
+	ExistingValues []int32 `protobuf:"varint,1,rep,packed,name=existingValues,proto3" json:"existingValues,omitempty"`
+	ObjectId       string  `protobuf:"bytes,2,opt,name=objectId,proto3" json:"objectId,omitempty"`
+}
+
+func (m *CounterRequest) Reset()         { *m = CounterRequest{} }
+func (m *CounterRequest) String() string { return proto.CompactTextString(m) }
+func (*CounterRequest) ProtoMessage()    {}
+func (*CounterRequest) Descriptor() ([]byte, []int) {
+	return fileDescriptor_dd5c22b15d7f69e4, []int{1}
+}
+func (m *CounterRequest) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *CounterRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_CounterRequest.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *CounterRequest) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_CounterRequest.Merge(m, src)
+}
+func (m *CounterRequest) XXX_Size() int {
+	return m.Size()
+}
+func (m *CounterRequest) XXX_DiscardUnknown() {
+	xxx_messageInfo_CounterRequest.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_CounterRequest proto.InternalMessageInfo
+
+func (m *CounterRequest) GetExistingValues() []int32 {
+	if m != nil {
+		return m.ExistingValues
+	}
+	return nil
+}
+
+func (m *CounterRequest) GetObjectId() string {
+	if m != nil {
+		return m.ObjectId
+	}
+	return ""
+}
+
+func init() {
+	proto.RegisterType((*CounterIncrease)(nil), "synctest.CounterIncrease")
+	proto.RegisterType((*CounterRequest)(nil), "synctest.CounterRequest")
+}
+
+func init() {
+	proto.RegisterFile("commonspace/sync/synctestproto/protos/synctest.proto", fileDescriptor_dd5c22b15d7f69e4)
+}
+
+var fileDescriptor_dd5c22b15d7f69e4 = []byte{
+	// 247 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0x32, 0x49, 0xce, 0xcf, 0xcd,
+	0xcd, 0xcf, 0x2b, 0x2e, 0x48, 0x4c, 0x4e, 0xd5, 0x2f, 0xae, 0xcc, 0x4b, 0x06, 0x13, 0x25, 0xa9,
+	0xc5, 0x25, 0x05, 0x45, 0xf9, 0x25, 0xf9, 0xfa, 0x60, 0xb2, 0x18, 0x2e, 0xa8, 0x07, 0xe6, 0x0b,
+	0x71, 0xc0, 0xf8, 0x4a, 0xce, 0x5c, 0xfc, 0xce, 0xf9, 0xa5, 0x79, 0x25, 0xa9, 0x45, 0x9e, 0x79,
+	0xc9, 0x45, 0xa9, 0x89, 0xc5, 0xa9, 0x42, 0x22, 0x5c, 0xac, 0x65, 0x89, 0x39, 0xa5, 0xa9, 0x12,
+	0x8c, 0x0a, 0x8c, 0x1a, 0xac, 0x41, 0x10, 0x8e, 0x90, 0x14, 0x17, 0x47, 0x7e, 0x52, 0x56, 0x6a,
+	0x72, 0x89, 0x67, 0x8a, 0x04, 0x93, 0x02, 0xa3, 0x06, 0x67, 0x10, 0x9c, 0xaf, 0x14, 0xc2, 0xc5,
+	0x07, 0x35, 0x24, 0x28, 0xb5, 0xb0, 0x34, 0xb5, 0xb8, 0x44, 0x48, 0x8d, 0x8b, 0x2f, 0xb5, 0x22,
+	0xb3, 0xb8, 0x24, 0x33, 0x2f, 0x3d, 0x0c, 0xa4, 0xbd, 0x58, 0x82, 0x51, 0x81, 0x59, 0x83, 0x35,
+	0x08, 0x4d, 0x14, 0x9f, 0xa9, 0x46, 0x4b, 0x18, 0xb9, 0xb8, 0xa1, 0xc6, 0x06, 0x57, 0xe6, 0x25,
+	0x0b, 0xf9, 0x72, 0x89, 0xc0, 0xb8, 0x25, 0x45, 0xa9, 0x89, 0xb9, 0x30, 0xbb, 0x24, 0xf4, 0xe0,
+	0xbe, 0x43, 0x75, 0x85, 0x94, 0x24, 0x86, 0x0c, 0xcc, 0x93, 0x06, 0x8c, 0x42, 0x6e, 0x5c, 0xbc,
+	0x28, 0xc6, 0x91, 0x69, 0x8e, 0x93, 0xc5, 0x89, 0x47, 0x72, 0x8c, 0x17, 0x1e, 0xc9, 0x31, 0x3e,
+	0x78, 0x24, 0xc7, 0x38, 0xe1, 0xb1, 0x1c, 0xc3, 0x85, 0xc7, 0x72, 0x0c, 0x37, 0x1e, 0xcb, 0x31,
+	0x44, 0xc9, 0xe1, 0x8f, 0x9b, 0x24, 0x36, 0x30, 0x65, 0x0c, 0x08, 0x00, 0x00, 0xff, 0xff, 0x3b,
+	0x05, 0x13, 0xdb, 0xc4, 0x01, 0x00, 0x00,
+}
+
+func (m *CounterIncrease) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *CounterIncrease) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *CounterIncrease) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if len(m.ObjectId) > 0 {
+		i -= len(m.ObjectId)
+		copy(dAtA[i:], m.ObjectId)
+		i = encodeVarintSynctest(dAtA, i, uint64(len(m.ObjectId)))
+		i--
+		dAtA[i] = 0x12
+	}
+	if m.Value != 0 {
+		i = encodeVarintSynctest(dAtA, i, uint64(m.Value))
+		i--
+		dAtA[i] = 0x8
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *CounterRequest) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *CounterRequest) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *CounterRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if len(m.ObjectId) > 0 {
+		i -= len(m.ObjectId)
+		copy(dAtA[i:], m.ObjectId)
+		i = encodeVarintSynctest(dAtA, i, uint64(len(m.ObjectId)))
+		i--
+		dAtA[i] = 0x12
+	}
+	if len(m.ExistingValues) > 0 {
+		dAtA2 := make([]byte, len(m.ExistingValues)*10)
+		var j1 int
+		for _, num1 := range m.ExistingValues {
+			num := uint64(num1)
+			for num >= 1<<7 {
+				dAtA2[j1] = uint8(uint64(num)&0x7f | 0x80)
+				num >>= 7
+				j1++
+			}
+			dAtA2[j1] = uint8(num)
+			j1++
+		}
+		i -= j1
+		copy(dAtA[i:], dAtA2[:j1])
+		i = encodeVarintSynctest(dAtA, i, uint64(j1))
+		i--
+		dAtA[i] = 0xa
+	}
+	return len(dAtA) - i, nil
+}
+
+func encodeVarintSynctest(dAtA []byte, offset int, v uint64) int {
+	offset -= sovSynctest(v)
+	base := offset
+	for v >= 1<<7 {
+		dAtA[offset] = uint8(v&0x7f | 0x80)
+		v >>= 7
+		offset++
+	}
+	dAtA[offset] = uint8(v)
+	return base
+}
+func (m *CounterIncrease) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.Value != 0 {
+		n += 1 + sovSynctest(uint64(m.Value))
+	}
+	l = len(m.ObjectId)
+	if l > 0 {
+		n += 1 + l + sovSynctest(uint64(l))
+	}
+	return n
+}
+
+func (m *CounterRequest) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if len(m.ExistingValues) > 0 {
+		l = 0
+		for _, e := range m.ExistingValues {
+			l += sovSynctest(uint64(e))
+		}
+		n += 1 + sovSynctest(uint64(l)) + l
+	}
+	l = len(m.ObjectId)
+	if l > 0 {
+		n += 1 + l + sovSynctest(uint64(l))
+	}
+	return n
+}
+
+func sovSynctest(x uint64) (n int) {
+	return (math_bits.Len64(x|1) + 6) / 7
+}
+func sozSynctest(x uint64) (n int) {
+	return sovSynctest(uint64((x << 1) ^ uint64((int64(x) >> 63))))
+}
+func (m *CounterIncrease) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowSynctest
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: CounterIncrease: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: CounterIncrease: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Value", wireType)
+			}
+			m.Value = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowSynctest
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Value |= int32(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ObjectId", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowSynctest
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthSynctest
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthSynctest
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.ObjectId = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipSynctest(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthSynctest
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *CounterRequest) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowSynctest
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: CounterRequest: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: CounterRequest: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType == 0 {
+				var v int32
+				for shift := uint(0); ; shift += 7 {
+					if shift >= 64 {
+						return ErrIntOverflowSynctest
+					}
+					if iNdEx >= l {
+						return io.ErrUnexpectedEOF
+					}
+					b := dAtA[iNdEx]
+					iNdEx++
+					v |= int32(b&0x7F) << shift
+					if b < 0x80 {
+						break
+					}
+				}
+				m.ExistingValues = append(m.ExistingValues, v)
+			} else if wireType == 2 {
+				var packedLen int
+				for shift := uint(0); ; shift += 7 {
+					if shift >= 64 {
+						return ErrIntOverflowSynctest
+					}
+					if iNdEx >= l {
+						return io.ErrUnexpectedEOF
+					}
+					b := dAtA[iNdEx]
+					iNdEx++
+					packedLen |= int(b&0x7F) << shift
+					if b < 0x80 {
+						break
+					}
+				}
+				if packedLen < 0 {
+					return ErrInvalidLengthSynctest
+				}
+				postIndex := iNdEx + packedLen
+				if postIndex < 0 {
+					return ErrInvalidLengthSynctest
+				}
+				if postIndex > l {
+					return io.ErrUnexpectedEOF
+				}
+				var elementCount int
+				var count int
+				for _, integer := range dAtA[iNdEx:postIndex] {
+					if integer < 128 {
+						count++
+					}
+				}
+				elementCount = count
+				if elementCount != 0 && len(m.ExistingValues) == 0 {
+					m.ExistingValues = make([]int32, 0, elementCount)
+				}
+				for iNdEx < postIndex {
+					var v int32
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowSynctest
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						v |= int32(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					m.ExistingValues = append(m.ExistingValues, v)
+				}
+			} else {
+				return fmt.Errorf("proto: wrong wireType = %d for field ExistingValues", wireType)
+			}
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ObjectId", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowSynctest
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthSynctest
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthSynctest
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.ObjectId = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipSynctest(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthSynctest
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func skipSynctest(dAtA []byte) (n int, err error) {
+	l := len(dAtA)
+	iNdEx := 0
+	depth := 0
+	for iNdEx < l {
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return 0, ErrIntOverflowSynctest
+			}
+			if iNdEx >= l {
+				return 0, io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= (uint64(b) & 0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		wireType := int(wire & 0x7)
+		switch wireType {
+		case 0:
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return 0, ErrIntOverflowSynctest
+				}
+				if iNdEx >= l {
+					return 0, io.ErrUnexpectedEOF
+				}
+				iNdEx++
+				if dAtA[iNdEx-1] < 0x80 {
+					break
+				}
+			}
+		case 1:
+			iNdEx += 8
+		case 2:
+			var length int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return 0, ErrIntOverflowSynctest
+				}
+				if iNdEx >= l {
+					return 0, io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				length |= (int(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if length < 0 {
+				return 0, ErrInvalidLengthSynctest
+			}
+			iNdEx += length
+		case 3:
+			depth++
+		case 4:
+			if depth == 0 {
+				return 0, ErrUnexpectedEndOfGroupSynctest
+			}
+			depth--
+		case 5:
+			iNdEx += 4
+		default:
+			return 0, fmt.Errorf("proto: illegal wireType %d", wireType)
+		}
+		if iNdEx < 0 {
+			return 0, ErrInvalidLengthSynctest
+		}
+		if depth == 0 {
+			return iNdEx, nil
+		}
+	}
+	return 0, io.ErrUnexpectedEOF
+}
+
+var (
+	ErrInvalidLengthSynctest        = fmt.Errorf("proto: negative length found during unmarshaling")
+	ErrIntOverflowSynctest          = fmt.Errorf("proto: integer overflow")
+	ErrUnexpectedEndOfGroupSynctest = fmt.Errorf("proto: unexpected end of group")
+)

commonspace/sync/synctestproto/synctest_drpc.pb.go

@@ -0,0 +1,209 @@
+// Code generated by protoc-gen-go-drpc. DO NOT EDIT.
+// protoc-gen-go-drpc version: v0.0.34
+// source: commonspace/sync/synctestproto/protos/synctest.proto
+
+package synctestproto
+
+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_commonspace_sync_synctestproto_protos_synctest_proto struct{}
+
+func (drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto) Marshal(msg drpc.Message) ([]byte, error) {
+	return proto.Marshal(msg.(proto.Message))
+}
+
+func (drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto) Unmarshal(buf []byte, msg drpc.Message) error {
+	return proto.Unmarshal(buf, msg.(proto.Message))
+}
+
+func (drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_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_commonspace_sync_synctestproto_protos_synctest_proto) JSONUnmarshal(buf []byte, msg drpc.Message) error {
+	return jsonpb.Unmarshal(bytes.NewReader(buf), msg.(proto.Message))
+}
+
+type DRPCCounterSyncClient interface {
+	DRPCConn() drpc.Conn
+
+	CounterStreamRequest(ctx context.Context, in *CounterRequest) (DRPCCounterSync_CounterStreamRequestClient, error)
+	CounterStream(ctx context.Context, in *CounterRequest) (DRPCCounterSync_CounterStreamClient, error)
+}
+
+type drpcCounterSyncClient struct {
+	cc drpc.Conn
+}
+
+func NewDRPCCounterSyncClient(cc drpc.Conn) DRPCCounterSyncClient {
+	return &drpcCounterSyncClient{cc}
+}
+
+func (c *drpcCounterSyncClient) DRPCConn() drpc.Conn { return c.cc }
+
+func (c *drpcCounterSyncClient) CounterStreamRequest(ctx context.Context, in *CounterRequest) (DRPCCounterSync_CounterStreamRequestClient, error) {
+	stream, err := c.cc.NewStream(ctx, "/synctest.CounterSync/CounterStreamRequest", drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto{})
+	if err != nil {
+		return nil, err
+	}
+	x := &drpcCounterSync_CounterStreamRequestClient{stream}
+	if err := x.MsgSend(in, drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto{}); err != nil {
+		return nil, err
+	}
+	if err := x.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+type DRPCCounterSync_CounterStreamRequestClient interface {
+	drpc.Stream
+	Recv() (*CounterIncrease, error)
+}
+
+type drpcCounterSync_CounterStreamRequestClient struct {
+	drpc.Stream
+}
+
+func (x *drpcCounterSync_CounterStreamRequestClient) GetStream() drpc.Stream {
+	return x.Stream
+}
+
+func (x *drpcCounterSync_CounterStreamRequestClient) Recv() (*CounterIncrease, error) {
+	m := new(CounterIncrease)
+	if err := x.MsgRecv(m, drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto{}); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func (x *drpcCounterSync_CounterStreamRequestClient) RecvMsg(m *CounterIncrease) error {
+	return x.MsgRecv(m, drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto{})
+}
+
+func (c *drpcCounterSyncClient) CounterStream(ctx context.Context, in *CounterRequest) (DRPCCounterSync_CounterStreamClient, error) {
+	stream, err := c.cc.NewStream(ctx, "/synctest.CounterSync/CounterStream", drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto{})
+	if err != nil {
+		return nil, err
+	}
+	x := &drpcCounterSync_CounterStreamClient{stream}
+	if err := x.MsgSend(in, drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto{}); err != nil {
+		return nil, err
+	}
+	if err := x.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+type DRPCCounterSync_CounterStreamClient interface {
+	drpc.Stream
+	Recv() (*CounterIncrease, error)
+}
+
+type drpcCounterSync_CounterStreamClient struct {
+	drpc.Stream
+}
+
+func (x *drpcCounterSync_CounterStreamClient) GetStream() drpc.Stream {
+	return x.Stream
+}
+
+func (x *drpcCounterSync_CounterStreamClient) Recv() (*CounterIncrease, error) {
+	m := new(CounterIncrease)
+	if err := x.MsgRecv(m, drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto{}); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func (x *drpcCounterSync_CounterStreamClient) RecvMsg(m *CounterIncrease) error {
+	return x.MsgRecv(m, drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto{})
+}
+
+type DRPCCounterSyncServer interface {
+	CounterStreamRequest(*CounterRequest, DRPCCounterSync_CounterStreamRequestStream) error
+	CounterStream(*CounterRequest, DRPCCounterSync_CounterStreamStream) error
+}
+
+type DRPCCounterSyncUnimplementedServer struct{}
+
+func (s *DRPCCounterSyncUnimplementedServer) CounterStreamRequest(*CounterRequest, DRPCCounterSync_CounterStreamRequestStream) error {
+	return drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
+}
+
+func (s *DRPCCounterSyncUnimplementedServer) CounterStream(*CounterRequest, DRPCCounterSync_CounterStreamStream) error {
+	return drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
+}
+
+type DRPCCounterSyncDescription struct{}
+
+func (DRPCCounterSyncDescription) NumMethods() int { return 2 }
+
+func (DRPCCounterSyncDescription) Method(n int) (string, drpc.Encoding, drpc.Receiver, interface{}, bool) {
+	switch n {
+	case 0:
+		return "/synctest.CounterSync/CounterStreamRequest", drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto{},
+			func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
+				return nil, srv.(DRPCCounterSyncServer).
+					CounterStreamRequest(
+						in1.(*CounterRequest),
+						&drpcCounterSync_CounterStreamRequestStream{in2.(drpc.Stream)},
+					)
+			}, DRPCCounterSyncServer.CounterStreamRequest, true
+	case 1:
+		return "/synctest.CounterSync/CounterStream", drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto{},
+			func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
+				return nil, srv.(DRPCCounterSyncServer).
+					CounterStream(
+						in1.(*CounterRequest),
+						&drpcCounterSync_CounterStreamStream{in2.(drpc.Stream)},
+					)
+			}, DRPCCounterSyncServer.CounterStream, true
+	default:
+		return "", nil, nil, nil, false
+	}
+}
+
+func DRPCRegisterCounterSync(mux drpc.Mux, impl DRPCCounterSyncServer) error {
+	return mux.Register(impl, DRPCCounterSyncDescription{})
+}
+
+type DRPCCounterSync_CounterStreamRequestStream interface {
+	drpc.Stream
+	Send(*CounterIncrease) error
+}
+
+type drpcCounterSync_CounterStreamRequestStream struct {
+	drpc.Stream
+}
+
+func (x *drpcCounterSync_CounterStreamRequestStream) Send(m *CounterIncrease) error {
+	return x.MsgSend(m, drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto{})
+}
+
+type DRPCCounterSync_CounterStreamStream interface {
+	drpc.Stream
+	Send(*CounterIncrease) error
+}
+
+type drpcCounterSync_CounterStreamStream struct {
+	drpc.Stream
+}
+
+func (x *drpcCounterSync_CounterStreamStream) Send(m *CounterIncrease) error {
+	return x.MsgSend(m, drpcEncoding_File_commonspace_sync_synctestproto_protos_synctest_proto{})
+}

consensus/consensusproto/consensus_drpc.pb.go

@@ -1,5 +1,5 @@
 // Code generated by protoc-gen-go-drpc. DO NOT EDIT.
-// protoc-gen-go-drpc version: v0.0.33
+// protoc-gen-go-drpc version: v0.0.34
 // source: consensus/consensusproto/protos/consensus.proto
 
 package consensusproto

coordinator/coordinatorproto/coordinator_drpc.pb.go

@@ -1,5 +1,5 @@
 // Code generated by protoc-gen-go-drpc. DO NOT EDIT.
-// protoc-gen-go-drpc version: v0.0.33
+// protoc-gen-go-drpc version: v0.0.34
 // source: coordinator/coordinatorproto/protos/coordinator.proto
 
 package coordinatorproto

identityrepo/identityrepoproto/identityrepo_drpc.pb.go

@@ -1,5 +1,5 @@
 // Code generated by protoc-gen-go-drpc. DO NOT EDIT.
-// protoc-gen-go-drpc version: v0.0.33
+// protoc-gen-go-drpc version: v0.0.34
 // source: identityrepo/identityrepoproto/protos/identityrepo.proto
 
 package identityrepoproto

nameservice/nameserviceproto/nameservice_aa_drpc.pb.go

@@ -1,5 +1,5 @@
 // Code generated by protoc-gen-go-drpc. DO NOT EDIT.
-// protoc-gen-go-drpc version: v0.0.33
+// protoc-gen-go-drpc version: v0.0.34
 // source: nameservice/nameserviceproto/protos/nameservice_aa.proto
 
 package nameserviceproto

nameservice/nameserviceproto/nameservice_drpc.pb.go

@@ -1,5 +1,5 @@
 // Code generated by protoc-gen-go-drpc. DO NOT EDIT.
-// protoc-gen-go-drpc version: v0.0.33
+// protoc-gen-go-drpc version: v0.0.34
 // source: nameservice/nameserviceproto/protos/nameservice.proto
 
 package nameserviceproto

net/streampool/testservice/testservice_drpc.pb.go

@@ -1,5 +1,5 @@
 // Code generated by protoc-gen-go-drpc. DO NOT EDIT.
-// protoc-gen-go-drpc version: v0.0.33
+// protoc-gen-go-drpc version: v0.0.34
 // source: net/streampool/testservice/protos/testservice.proto
 
 package testservice

paymentservice/paymentserviceproto/paymentservice_drpc.pb.go

@@ -1,5 +1,5 @@
 // Code generated by protoc-gen-go-drpc. DO NOT EDIT.
-// protoc-gen-go-drpc version: v0.0.33
+// protoc-gen-go-drpc version: v0.0.34
 // source: paymentservice/paymentserviceproto/protos/paymentservice.proto
 
 package paymentserviceproto