consensus node

consensus/config/account.go

@@ -1,7 +0,0 @@
-package config
-
-type Account struct {
-	PeerId        string `yaml:"peerId"`
-	SigningKey    string `yaml:"signingKey"`
-	EncryptionKey string `yaml:"encryptionKey"`
-}

consensus/config/config.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/config"
 	"gopkg.in/yaml.v3"
 	"io/ioutil"
 )
@@ -21,9 +22,9 @@ func NewFromFile(path string) (c *Config, err error) {
 }
 
 type Config struct {
-	GrpcServer GrpcServer `yaml:"grpcServer"`
-	Account    Account    `yaml:"account"`
-	Mongo      Mongo      `yaml:"mongo"`
+	GrpcServer config.GrpcServer `yaml:"grpcServer"`
+	Account    config.Account    `yaml:"account"`
+	Mongo      Mongo             `yaml:"mongo"`
 }
 
 func (c *Config) Init(a *app.App) (err error) {
@@ -33,3 +34,15 @@ func (c *Config) Init(a *app.App) (err error) {
 func (c Config) Name() (name string) {
 	return CName
 }
+
+func (c Config) GetMongo() Mongo {
+	return c.Mongo
+}
+
+func (c Config) GetGRPCServer() config.GrpcServer {
+	return c.GrpcServer
+}
+
+func (c Config) GetAccount() config.Account {
+	return c.Account
+}

consensus/config/grpc.go

@@ -1,6 +0,0 @@
-package config
-
-type GrpcServer struct {
-	ListenAddrs []string `yaml:"listenAddrs"`
-	TLS         bool     `yaml:"tls"`
-}

consensus/consensusrpc/consensrpc.go

@@ -0,0 +1,114 @@
+package consensusrpc
+
+import (
+	"context"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/rpc/server"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/consensusproto"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/db"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/stream"
+	"time"
+)
+
+const CName = "consensus.consensusrpc"
+
+func New() app.Component {
+	return &consensusRpc{}
+}
+
+type consensusRpc struct {
+	db     db.Service
+	stream stream.Service
+}
+
+func (c *consensusRpc) Init(a *app.App) (err error) {
+	c.db = a.MustComponent(db.CName).(db.Service)
+	c.stream = a.MustComponent(stream.CName).(stream.Service)
+	return consensusproto.DRPCRegisterConsensus(a.MustComponent(server.CName).(server.DRPCServer), c)
+}
+
+func (c *consensusRpc) Name() (name string) {
+	return CName
+}
+
+func (c *consensusRpc) AddLog(ctx context.Context, req *consensusproto.AddLogRequest) (*consensusproto.Ok, error) {
+	if err := c.db.AddLog(ctx, logFromProto(req.Log)); err != nil {
+		return nil, err
+	}
+	return &consensusproto.Ok{}, nil
+}
+
+func (c *consensusRpc) AddRecord(ctx context.Context, req *consensusproto.AddRecordRequest) (*consensusproto.Ok, error) {
+	if err := c.db.AddRecord(ctx, req.LogId, recordFromProto(req.Record)); err != nil {
+		return nil, err
+	}
+	return &consensusproto.Ok{}, nil
+}
+
+func (c *consensusRpc) WatchLog(req *consensusproto.WatchLogRequest, rpcStream consensusproto.DRPCConsensus_WatchLogStream) error {
+	stream, err := c.stream.Subscribe(rpcStream.Context(), req.LogId)
+	if err != nil {
+		return err
+	}
+	defer stream.Close()
+	var logSent bool
+	for {
+		if !logSent {
+			if err = rpcStream.Send(&consensusproto.WatchLogEvent{
+				LogId:   req.LogId,
+				Records: recordsToProto(stream.Records()),
+			}); err != nil {
+				return err
+			}
+		} else {
+			recs := stream.WaitRecords()
+			if len(recs) == 0 {
+				return rpcStream.Close()
+			}
+			if err = rpcStream.Send(&consensusproto.WatchLogEvent{
+				LogId:   req.LogId,
+				Records: recordsToProto(recs),
+			}); err != nil {
+				return err
+			}
+		}
+	}
+}
+
+func logFromProto(log *consensusproto.Log) consensus.Log {
+	return consensus.Log{
+		Id:      log.Id,
+		Records: recordsFromProto(log.Records),
+	}
+}
+
+func recordsFromProto(recs []*consensusproto.Record) []consensus.Record {
+	res := make([]consensus.Record, len(recs))
+	for i, rec := range recs {
+		res[i] = recordFromProto(rec)
+	}
+	return res
+}
+
+func recordFromProto(rec *consensusproto.Record) consensus.Record {
+	return consensus.Record{
+		Id:      rec.Id,
+		PrevId:  rec.PrevId,
+		Payload: rec.Payload,
+		Created: time.Unix(int64(rec.CreatedUnix), 0),
+	}
+}
+
+func recordsToProto(recs []consensus.Record) []*consensusproto.Record {
+	res := make([]*consensusproto.Record, len(recs))
+	for i, rec := range recs {
+		res[i] = &consensusproto.Record{
+			Id:          rec.Id,
+			PrevId:      rec.PrevId,
+			Payload:     rec.Payload,
+			CreatedUnix: uint64(rec.Created.Unix()),
+		}
+	}
+	return res
+}

consensus/db/db.go

@@ -153,7 +153,7 @@ type streamResult struct {
 	UpdateDescription struct {
 		UpdateFields struct {
 			Records []consensus.Record `bson:"records"`
-		} `bson:"updatedFields""`
+		} `bson:"updatedFields"`
 	} `bson:"updateDescription"`
 }
 

consensus/stream/object.go

@@ -0,0 +1,68 @@
+package stream
+
+import (
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus"
+	"github.com/cheggaaa/mb/v2"
+	"sync"
+)
+
+type object struct {
+	logId   []byte
+	records []consensus.Record
+
+	streams map[uint32]*stream
+
+	lastStreamId uint32
+
+	mu sync.Mutex
+}
+
+func (o *object) AddRecords(recs []consensus.Record) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	if len(recs) <= len(o.records) {
+		return
+	}
+	diff := recs[0 : len(recs)-len(o.records)]
+	o.records = recs
+	for _, st := range o.streams {
+		st.AddRecords(diff)
+	}
+}
+
+func (o *object) Records() []consensus.Record {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return o.records
+}
+
+func (o *object) NewStream() Stream {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.lastStreamId++
+	st := &stream{
+		id:      o.lastStreamId,
+		obj:     o,
+		records: o.records,
+		mb:      mb.New(consensus.Record{}, 100),
+	}
+	o.streams[st.id] = st
+	return st
+}
+
+func (o *object) Locked() bool {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return len(o.streams) > 0
+}
+
+func (o *object) removeStream(id uint32) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	delete(o.streams, id)
+}
+
+func (o *object) Close() (err error) {
+	return nil
+}

consensus/stream/service.go

@@ -0,0 +1,127 @@
+package stream
+
+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/consensus"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/db"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/pkg/ocache"
+	"github.com/mr-tron/base58"
+	"go.uber.org/zap"
+	"time"
+)
+
+const CName = "consensus.stream"
+
+var log = logger.NewNamed(CName)
+
+var (
+	cacheTTL = time.Minute
+)
+
+type ctxLog uint
+
+const (
+	ctxLogKey ctxLog = 1
+)
+
+func New() Service {
+	return &service{}
+}
+
+type Stream interface {
+	LogId() []byte
+	Records() []consensus.Record
+	WaitRecords() []consensus.Record
+	Close()
+}
+
+type Service interface {
+	Subscribe(ctx context.Context, logId []byte) (stream Stream, err error)
+	app.ComponentRunnable
+}
+
+type service struct {
+	db    db.Service
+	cache ocache.OCache
+}
+
+func (s *service) Init(a *app.App) (err error) {
+	s.db = a.MustComponent(db.CName).(db.Service)
+	s.cache = ocache.New(s.loadLog,
+		ocache.WithTTL(cacheTTL),
+		ocache.WithRefCounter(false),
+		ocache.WithLogger(log.Named("cache").Sugar()),
+	)
+
+	return s.db.SetChangeReceiver(s.receiveChange)
+}
+
+func (s *service) Subscribe(ctx context.Context, logId []byte) (Stream, error) {
+	obj, err := s.getObject(ctx, logId)
+	if err != nil {
+		return nil, err
+	}
+	return obj.NewStream(), nil
+}
+
+func (s *service) Name() (name string) {
+	return CName
+}
+
+func (s *service) Run(ctx context.Context) (err error) {
+	return nil
+}
+
+func (s *service) loadLog(ctx context.Context, id string) (value ocache.Object, err error) {
+	if ctxLog := ctx.Value(ctxLogKey); ctxLog != nil {
+		return &object{
+			logId:   ctxLog.(consensus.Log).Id,
+			records: ctxLog.(consensus.Log).Records,
+			streams: make(map[uint32]*stream),
+		}, nil
+	}
+	logId := logIdFromString(id)
+	dbLog, err := s.db.FetchLog(ctx, logId)
+	if err != nil {
+		return nil, err
+	}
+	return &object{
+		logId:   dbLog.Id,
+		records: dbLog.Records,
+		streams: make(map[uint32]*stream),
+	}, nil
+}
+
+func (s *service) receiveChange(logId []byte, records []consensus.Record) {
+	ctx := context.WithValue(context.Background(), ctxLogKey, consensus.Log{Id: logId, Records: records})
+	obj, err := s.getObject(ctx, logId)
+	if err != nil {
+		log.Error("failed load object from cache", zap.Error(err))
+		return
+	}
+	obj.AddRecords(records)
+}
+
+func (s *service) getObject(ctx context.Context, logId []byte) (*object, error) {
+	id := logIdToString(logId)
+	cacheObj, err := s.cache.Get(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	return cacheObj.(*object), nil
+}
+
+func (s *service) Close(ctx context.Context) (err error) {
+	return s.cache.Close()
+}
+
+func logIdToString(logId []byte) string {
+	return base58.Encode(logId)
+}
+
+func logIdFromString(s string) []byte {
+	logId, _ := base58.Decode(s)
+	return logId
+}

consensus/stream/service_test.go

@@ -0,0 +1,169 @@
+package stream
+
+import (
+	"context"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/db"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"testing"
+	"time"
+)
+
+var ctx = context.Background()
+
+func TestService_Subscribe(t *testing.T) {
+	fx := newFixture(t)
+	defer fx.Finish(t)
+
+	var expLogId = []byte("logId")
+
+	fx.mockDB.fetchLog = func(ctx context.Context, logId []byte) (log consensus.Log, err error) {
+		require.Equal(t, expLogId, logId)
+		return consensus.Log{
+			Id: logId,
+			Records: []consensus.Record{
+				{
+					Id: []byte{'1'},
+				},
+			},
+		}, nil
+	}
+
+	st1, err := fx.Subscribe(ctx, expLogId)
+	require.NoError(t, err)
+	require.Equal(t, expLogId, st1.LogId())
+	sr1 := readStream(st1)
+	assert.Equal(t, uint32(1), sr1.id)
+
+	st2, err := fx.Subscribe(ctx, expLogId)
+	require.NoError(t, err)
+	require.Equal(t, expLogId, st2.LogId())
+	sr2 := readStream(st2)
+	assert.Equal(t, uint32(2), sr2.id)
+
+	fx.mockDB.receiver(expLogId, []consensus.Record{
+		{
+			Id: []byte{'1'},
+		},
+	})
+	fx.mockDB.receiver([]byte("other id"), []consensus.Record{
+		{
+			Id:     []byte{'2'},
+			PrevId: []byte{'1'},
+		},
+		{
+			Id: []byte{'1'},
+		},
+	})
+	fx.mockDB.receiver(expLogId, []consensus.Record{
+		{
+			Id:     []byte{'2'},
+			PrevId: []byte{'1'},
+		},
+		{
+			Id: []byte{'1'},
+		},
+	})
+	st1.Close()
+	st2.Close()
+
+	for _, sr := range []*streamReader{sr1, sr2} {
+		select {
+		case <-time.After(time.Second / 3):
+			require.False(t, true, "timeout")
+		case <-sr.finished:
+		}
+	}
+
+	require.Equal(t, sr1.records, sr2.records)
+	require.Len(t, sr1.records, 1)
+	assert.Equal(t, []byte{'2'}, sr1.records[0].Id)
+}
+
+func newFixture(t *testing.T) *fixture {
+	fx := &fixture{
+		Service: New(),
+		mockDB:  &mockDB{},
+		a:       new(app.App),
+	}
+
+	fx.a.Register(fx.Service).Register(fx.mockDB)
+	require.NoError(t, fx.a.Start(ctx))
+	return fx
+}
+
+type fixture struct {
+	Service
+	mockDB *mockDB
+	a      *app.App
+}
+
+func (fx *fixture) Finish(t *testing.T) {
+	require.NoError(t, fx.a.Close(ctx))
+}
+
+func readStream(st Stream) *streamReader {
+	sr := &streamReader{
+		id:       st.(*stream).id,
+		stream:   st,
+		finished: make(chan struct{}),
+	}
+	go sr.read()
+	return sr
+}
+
+type streamReader struct {
+	id     uint32
+	stream Stream
+
+	records  []consensus.Record
+	finished chan struct{}
+}
+
+func (sr *streamReader) read() {
+	defer close(sr.finished)
+	for {
+		records := sr.stream.WaitRecords()
+		if len(records) == 0 {
+			return
+		}
+		sr.records = append(sr.records, records...)
+	}
+}
+
+type mockDB struct {
+	receiver db.ChangeReceiver
+	fetchLog func(ctx context.Context, logId []byte) (log consensus.Log, err error)
+}
+
+func (m *mockDB) AddLog(ctx context.Context, log consensus.Log) (err error) { return nil }
+func (m *mockDB) AddRecord(ctx context.Context, logId []byte, record consensus.Record) (err error) {
+	return nil
+}
+
+func (m *mockDB) FetchLog(ctx context.Context, logId []byte) (log consensus.Log, err error) {
+	return m.fetchLog(ctx, logId)
+}
+
+func (m *mockDB) SetChangeReceiver(receiver db.ChangeReceiver) (err error) {
+	m.receiver = receiver
+	return nil
+}
+
+func (m *mockDB) Init(a *app.App) (err error) {
+	return nil
+}
+
+func (m *mockDB) Name() (name string) {
+	return db.CName
+}
+
+func (m *mockDB) Run(ctx context.Context) (err error) {
+	return
+}
+
+func (m *mockDB) Close(ctx context.Context) (err error) {
+	return
+}

consensus/stream/stream.go

@@ -0,0 +1,34 @@
+package stream
+
+import (
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus"
+	"github.com/cheggaaa/mb/v2"
+)
+
+type stream struct {
+	id      uint32
+	obj     *object
+	records []consensus.Record
+	mb      *mb.MB[consensus.Record]
+}
+
+func (s *stream) LogId() []byte {
+	return s.obj.logId
+}
+
+func (s *stream) AddRecords(records []consensus.Record) {
+	_ = s.mb.Add(records...)
+}
+
+func (s *stream) Records() []consensus.Record {
+	return s.records
+}
+
+func (s *stream) WaitRecords() []consensus.Record {
+	return s.mb.Wait()
+}
+
+func (s *stream) Close() {
+	_ = s.mb.Close()
+	s.obj.removeStream(s.id)
+}