consistent node db component

Makefile

@@ -29,6 +29,8 @@ proto:
 	$(GOGO_START) protoc --gogofaster_out=:. $(P_TEST_CHANGES_PATH_PB)/proto/*.proto
 	$(eval PKGMAP := $$(P_TREE_CHANGES),$$(P_ACL_RECORDS))
 	$(GOGO_START) protoc --gogofaster_out=$(PKGMAP):. --go-drpc_out=protolib=github.com/gogo/protobuf:. common/commonspace/spacesyncproto/protos/*.proto
+	$(GOGO_START) protoc --gogofaster_out=:. --go-drpc_out=protolib=github.com/gogo/protobuf:. consensus/consensusproto/protos/*.proto
+
 
 
 build:

consensus/config/account.go

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

consensus/config/config.go

@@ -0,0 +1,35 @@
+package config
+
+import (
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
+	"gopkg.in/yaml.v3"
+	"io/ioutil"
+)
+
+const CName = "config"
+
+func NewFromFile(path string) (c *Config, err error) {
+	c = &Config{}
+	data, err := ioutil.ReadFile(path)
+	if err != nil {
+		return nil, err
+	}
+	if err = yaml.Unmarshal(data, c); err != nil {
+		return nil, err
+	}
+	return
+}
+
+type Config struct {
+	GrpcServer GrpcServer `yaml:"grpcServer"`
+	Account    Account    `yaml:"account"`
+	Mongo      Mongo      `yaml:"mongo"`
+}
+
+func (c *Config) Init(a *app.App) (err error) {
+	return
+}
+
+func (c Config) Name() (name string) {
+	return CName
+}

consensus/config/grpc.go

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

consensus/config/mongo.go

@@ -0,0 +1,7 @@
+package config
+
+type Mongo struct {
+	Connect       string `yaml:"connect"`
+	Database      string `yaml:"database"`
+	LogCollection string `yaml:"logCollection"`
+}

consensus/consensus.go

@@ -0,0 +1,22 @@
+package consensus
+
+import "time"
+
+type Log struct {
+	Id      []byte   `bson:"_id"`
+	Records []Record `bson:"records"`
+}
+
+type Record struct {
+	Id      []byte    `bson:"id"`
+	PrevId  []byte    `bson:"prevId"`
+	Payload []byte    `bson:"payload"`
+	Created time.Time `bson:"created"'`
+}
+
+func (l Log) CopyRecords() Log {
+	l2 := l
+	l2.Records = make([]Record, len(l.Records))
+	copy(l2.Records, l.Records)
+	return l2
+}

consensus/consensusproto/consensus_drpc.pb.go

@@ -0,0 +1,217 @@
+// Code generated by protoc-gen-go-drpc. DO NOT EDIT.
+// protoc-gen-go-drpc version: v0.0.32
+// source: consensus/consensusproto/protos/consensus.proto
+
+package consensusproto
+
+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_consensus_consensusproto_protos_consensus_proto struct{}
+
+func (drpcEncoding_File_consensus_consensusproto_protos_consensus_proto) Marshal(msg drpc.Message) ([]byte, error) {
+	return proto.Marshal(msg.(proto.Message))
+}
+
+func (drpcEncoding_File_consensus_consensusproto_protos_consensus_proto) Unmarshal(buf []byte, msg drpc.Message) error {
+	return proto.Unmarshal(buf, msg.(proto.Message))
+}
+
+func (drpcEncoding_File_consensus_consensusproto_protos_consensus_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_consensus_consensusproto_protos_consensus_proto) JSONUnmarshal(buf []byte, msg drpc.Message) error {
+	return jsonpb.Unmarshal(bytes.NewReader(buf), msg.(proto.Message))
+}
+
+type DRPCConsensusClient interface {
+	DRPCConn() drpc.Conn
+
+	AddLog(ctx context.Context, in *AddLogRequest) (*Ok, error)
+	AddRecord(ctx context.Context, in *AddRecordRequest) (*Ok, error)
+	WatchLog(ctx context.Context, in *WatchLogRequest) (DRPCConsensus_WatchLogClient, error)
+}
+
+type drpcConsensusClient struct {
+	cc drpc.Conn
+}
+
+func NewDRPCConsensusClient(cc drpc.Conn) DRPCConsensusClient {
+	return &drpcConsensusClient{cc}
+}
+
+func (c *drpcConsensusClient) DRPCConn() drpc.Conn { return c.cc }
+
+func (c *drpcConsensusClient) AddLog(ctx context.Context, in *AddLogRequest) (*Ok, error) {
+	out := new(Ok)
+	err := c.cc.Invoke(ctx, "/anyConsensus.Consensus/AddLog", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}, in, out)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *drpcConsensusClient) AddRecord(ctx context.Context, in *AddRecordRequest) (*Ok, error) {
+	out := new(Ok)
+	err := c.cc.Invoke(ctx, "/anyConsensus.Consensus/AddRecord", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}, in, out)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *drpcConsensusClient) WatchLog(ctx context.Context, in *WatchLogRequest) (DRPCConsensus_WatchLogClient, error) {
+	stream, err := c.cc.NewStream(ctx, "/anyConsensus.Consensus/WatchLog", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{})
+	if err != nil {
+		return nil, err
+	}
+	x := &drpcConsensus_WatchLogClient{stream}
+	if err := x.MsgSend(in, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}); err != nil {
+		return nil, err
+	}
+	if err := x.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+type DRPCConsensus_WatchLogClient interface {
+	drpc.Stream
+	Recv() (*WatchLogEvent, error)
+}
+
+type drpcConsensus_WatchLogClient struct {
+	drpc.Stream
+}
+
+func (x *drpcConsensus_WatchLogClient) Recv() (*WatchLogEvent, error) {
+	m := new(WatchLogEvent)
+	if err := x.MsgRecv(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func (x *drpcConsensus_WatchLogClient) RecvMsg(m *WatchLogEvent) error {
+	return x.MsgRecv(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{})
+}
+
+type DRPCConsensusServer interface {
+	AddLog(context.Context, *AddLogRequest) (*Ok, error)
+	AddRecord(context.Context, *AddRecordRequest) (*Ok, error)
+	WatchLog(*WatchLogRequest, DRPCConsensus_WatchLogStream) error
+}
+
+type DRPCConsensusUnimplementedServer struct{}
+
+func (s *DRPCConsensusUnimplementedServer) AddLog(context.Context, *AddLogRequest) (*Ok, error) {
+	return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
+}
+
+func (s *DRPCConsensusUnimplementedServer) AddRecord(context.Context, *AddRecordRequest) (*Ok, error) {
+	return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
+}
+
+func (s *DRPCConsensusUnimplementedServer) WatchLog(*WatchLogRequest, DRPCConsensus_WatchLogStream) error {
+	return drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
+}
+
+type DRPCConsensusDescription struct{}
+
+func (DRPCConsensusDescription) NumMethods() int { return 3 }
+
+func (DRPCConsensusDescription) Method(n int) (string, drpc.Encoding, drpc.Receiver, interface{}, bool) {
+	switch n {
+	case 0:
+		return "/anyConsensus.Consensus/AddLog", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{},
+			func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
+				return srv.(DRPCConsensusServer).
+					AddLog(
+						ctx,
+						in1.(*AddLogRequest),
+					)
+			}, DRPCConsensusServer.AddLog, true
+	case 1:
+		return "/anyConsensus.Consensus/AddRecord", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{},
+			func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
+				return srv.(DRPCConsensusServer).
+					AddRecord(
+						ctx,
+						in1.(*AddRecordRequest),
+					)
+			}, DRPCConsensusServer.AddRecord, true
+	case 2:
+		return "/anyConsensus.Consensus/WatchLog", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{},
+			func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
+				return nil, srv.(DRPCConsensusServer).
+					WatchLog(
+						in1.(*WatchLogRequest),
+						&drpcConsensus_WatchLogStream{in2.(drpc.Stream)},
+					)
+			}, DRPCConsensusServer.WatchLog, true
+	default:
+		return "", nil, nil, nil, false
+	}
+}
+
+func DRPCRegisterConsensus(mux drpc.Mux, impl DRPCConsensusServer) error {
+	return mux.Register(impl, DRPCConsensusDescription{})
+}
+
+type DRPCConsensus_AddLogStream interface {
+	drpc.Stream
+	SendAndClose(*Ok) error
+}
+
+type drpcConsensus_AddLogStream struct {
+	drpc.Stream
+}
+
+func (x *drpcConsensus_AddLogStream) SendAndClose(m *Ok) error {
+	if err := x.MsgSend(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}); err != nil {
+		return err
+	}
+	return x.CloseSend()
+}
+
+type DRPCConsensus_AddRecordStream interface {
+	drpc.Stream
+	SendAndClose(*Ok) error
+}
+
+type drpcConsensus_AddRecordStream struct {
+	drpc.Stream
+}
+
+func (x *drpcConsensus_AddRecordStream) SendAndClose(m *Ok) error {
+	if err := x.MsgSend(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}); err != nil {
+		return err
+	}
+	return x.CloseSend()
+}
+
+type DRPCConsensus_WatchLogStream interface {
+	drpc.Stream
+	Send(*WatchLogEvent) error
+}
+
+type drpcConsensus_WatchLogStream struct {
+	drpc.Stream
+}
+
+func (x *drpcConsensus_WatchLogStream) Send(m *WatchLogEvent) error {
+	return x.MsgSend(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{})
+}

consensus/consensusproto/consensuserrs/errors.go

@@ -0,0 +1,16 @@
+package consensuserrs
+
+import (
+	"fmt"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/rpc/rpcerr"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/consensusproto"
+)
+
+var (
+	errGroup = rpcerr.ErrGroup(consensusproto.ErrCodes_ErrorOffset)
+
+	ErrUnexpected  = errGroup.Register(fmt.Errorf("unexpected consensus error"), uint64(consensusproto.ErrCodes_Unexpected))
+	ErrConflict    = errGroup.Register(fmt.Errorf("records conflict"), uint64(consensusproto.ErrCodes_RecordConflict))
+	ErrLogExists   = errGroup.Register(fmt.Errorf("log exists"), uint64(consensusproto.ErrCodes_LogExists))
+	ErrLogNotFound = errGroup.Register(fmt.Errorf("log not found"), uint64(consensusproto.ErrCodes_LogNotFound))
+)

consensus/consensusproto/protos/consensus.proto

@@ -0,0 +1,55 @@
+syntax = "proto3";
+package anyConsensus;
+
+option go_package = "consensus/consensusproto";
+
+enum ErrCodes {
+    Unexpected = 0;
+    LogExists = 1;
+    LogNotFound = 2;
+    RecordConflict = 3;
+    ErrorOffset = 300;
+}
+
+
+message Log {
+    bytes id = 1;
+    repeated Record records = 2;
+    uint64 createdUnix = 3;
+}
+
+message Record {
+    bytes id = 1;
+    bytes prevId = 2;
+    bytes payload = 3;
+    uint64 createdUnix = 4;
+}
+
+service Consensus {
+    // AddLog adds new log to consensus
+    rpc AddLog(AddLogRequest) returns (Ok);
+    // AddRecord adds new record to log
+    rpc AddRecord(AddRecordRequest) returns (Ok);
+    // WatchLog fetches log and subscribes for a changes
+    rpc WatchLog(WatchLogRequest) returns (stream WatchLogEvent);
+}
+
+message Ok {}
+
+message AddLogRequest {
+    Log log = 1;
+}
+
+message AddRecordRequest {
+    bytes logId = 1;
+    Record record = 2;
+}
+
+message WatchLogRequest {
+    bytes logId = 1;
+}
+
+message WatchLogEvent {
+    bytes logId = 1;
+    repeated Record records = 2;
+}

consensus/db/db.go

@@ -0,0 +1,181 @@
+package db
+
+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/consensus"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/config"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/consensusproto/consensuserrs"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.uber.org/zap"
+)
+
+const CName = "consensus.db"
+
+var log = logger.NewNamed(CName)
+
+func New() Service {
+	return &service{}
+}
+
+type ChangeReceiver func(logId []byte, records []consensus.Record)
+
+type Service interface {
+	AddLog(ctx context.Context, log consensus.Log) (err error)
+	AddRecord(ctx context.Context, logId []byte, record consensus.Record) (err error)
+	FetchLog(ctx context.Context, logId []byte) (log consensus.Log, err error)
+	SetChangeReceiver(receiver ChangeReceiver) (err error)
+	app.ComponentRunnable
+}
+
+type service struct {
+	conf           config.Mongo
+	logColl        *mongo.Collection
+	running        bool
+	changeReceiver ChangeReceiver
+
+	streamCtx    context.Context
+	streamCancel context.CancelFunc
+	listenerDone chan struct{}
+}
+
+func (s *service) Init(a *app.App) (err error) {
+	s.conf = a.MustComponent(config.CName).(*config.Config).Mongo
+	return nil
+}
+
+func (s *service) Name() (name string) {
+	return CName
+}
+
+func (s *service) Run(ctx context.Context) (err error) {
+	client, err := mongo.Connect(ctx, options.Client().ApplyURI(s.conf.Connect))
+	if err != nil {
+		return err
+	}
+	s.logColl = client.Database(s.conf.Database).Collection(s.conf.LogCollection)
+	s.running = true
+	if s.changeReceiver != nil {
+		if err = s.runStreamListener(ctx); err != nil {
+			return err
+		}
+	}
+	return
+}
+
+func (s *service) AddLog(ctx context.Context, l consensus.Log) (err error) {
+	_, err = s.logColl.InsertOne(ctx, l)
+	if mongo.IsDuplicateKeyError(err) {
+		return consensuserrs.ErrLogExists
+	}
+	return
+}
+
+type findLogQuery struct {
+	Id []byte `bson:"_id"`
+}
+
+type findRecordQuery struct {
+	Id           []byte `bson:"_id"`
+	LastRecordId []byte `bson:"records.0.id"`
+}
+
+type updateOp struct {
+	Push struct {
+		Records struct {
+			Each []consensus.Record `bson:"$each""`
+			Pos  int                `bson:"$position"`
+		} `bson:"records"`
+	} `bson:"$push"`
+}
+
+func (s *service) AddRecord(ctx context.Context, logId []byte, record consensus.Record) (err error) {
+	var upd updateOp
+	upd.Push.Records.Each = []consensus.Record{record}
+	result, err := s.logColl.UpdateOne(ctx, findRecordQuery{
+		Id:           logId,
+		LastRecordId: record.PrevId,
+	}, upd)
+	if err != nil {
+		log.Error("addRecord update error", zap.Error(err))
+		return consensuserrs.ErrUnexpected
+	}
+	if result.ModifiedCount == 0 {
+		return consensuserrs.ErrConflict
+	}
+	return
+}
+
+func (s *service) FetchLog(ctx context.Context, logId []byte) (l consensus.Log, err error) {
+	if err = s.logColl.FindOne(ctx, findLogQuery{Id: logId}).Decode(&l); err != nil {
+		if err == mongo.ErrNoDocuments {
+			err = consensuserrs.ErrLogNotFound
+		}
+		return
+	}
+	return
+}
+
+func (s *service) SetChangeReceiver(receiver ChangeReceiver) (err error) {
+	if s.running {
+		return fmt.Errorf("set receiver must be called before Run")
+	}
+	s.changeReceiver = receiver
+	return
+}
+
+type matchPipeline struct {
+	Match struct {
+		OT string `bson:"operationType"`
+	} `bson:"$match"`
+}
+
+func (s *service) runStreamListener(ctx context.Context) (err error) {
+	var mp matchPipeline
+	mp.Match.OT = "update"
+	stream, err := s.logColl.Watch(ctx, []matchPipeline{mp})
+	if err != nil {
+		return
+	}
+	s.listenerDone = make(chan struct{})
+	s.streamCtx, s.streamCancel = context.WithCancel(context.Background())
+	go s.streamListener(stream)
+	return
+}
+
+type streamResult struct {
+	DocumentKey struct {
+		Id []byte `bson:"_id"`
+	} `bson:"documentKey"`
+	UpdateDescription struct {
+		UpdateFields struct {
+			Records []consensus.Record `bson:"records"`
+		} `bson:"updatedFields""`
+	} `bson:"updateDescription"`
+}
+
+func (s *service) streamListener(stream *mongo.ChangeStream) {
+	defer close(s.listenerDone)
+	for stream.Next(s.streamCtx) {
+		var res streamResult
+		if err := stream.Decode(&res); err != nil {
+			log.Error("stream decode error:", zap.Error(err))
+		}
+		s.changeReceiver(res.DocumentKey.Id, res.UpdateDescription.UpdateFields.Records)
+	}
+}
+
+func (s *service) Close(ctx context.Context) (err error) {
+	if s.logColl != nil {
+		err = s.logColl.Database().Client().Disconnect(ctx)
+		s.logColl = nil
+	}
+	if s.listenerDone != nil {
+		s.streamCancel()
+		<-s.listenerDone
+	}
+	return
+}

consensus/db/db_test.go

@@ -0,0 +1,218 @@
+package db
+
+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/config"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/consensusproto/consensuserrs"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"testing"
+	"time"
+)
+
+var ctx = context.Background()
+
+func TestService_AddLog(t *testing.T) {
+	t.Run("success", func(t *testing.T) {
+		fx := newFixture(t, nil)
+		defer fx.Finish(t)
+		log := consensus.Log{
+			Id: []byte("logOne"),
+			Records: []consensus.Record{
+				{
+					Id:      []byte("recordOne"),
+					PrevId:  nil,
+					Payload: []byte("payload"),
+					Created: time.Now().Truncate(time.Second).UTC(),
+				},
+			},
+		}
+		require.NoError(t, fx.AddLog(ctx, log))
+		fetched, err := fx.FetchLog(ctx, log.Id)
+		require.NoError(t, err)
+		assert.Equal(t, log, fetched)
+	})
+	t.Run("duplicate error", func(t *testing.T) {
+		fx := newFixture(t, nil)
+		defer fx.Finish(t)
+		log := consensus.Log{
+			Id: []byte("logOne"),
+		}
+		require.NoError(t, fx.AddLog(ctx, log))
+		// TODO: check for specified error
+		require.Error(t, fx.AddLog(ctx, log))
+	})
+}
+
+func TestService_AddRecord(t *testing.T) {
+	t.Run("success", func(t *testing.T) {
+		fx := newFixture(t, nil)
+		defer fx.Finish(t)
+		var records = []consensus.Record{
+			{
+				Id:     []byte("2"),
+				PrevId: []byte("1"),
+			},
+			{
+				Id:     []byte("3"),
+				PrevId: []byte("2"),
+			},
+			{
+				Id:     []byte("4"),
+				PrevId: []byte("3"),
+			},
+		}
+		l := consensus.Log{
+			Id: []byte("logTestRecords"),
+			Records: []consensus.Record{
+				{
+					Id: []byte("1"),
+				},
+			},
+		}
+		require.NoError(t, fx.AddLog(ctx, l))
+		for _, rec := range records {
+			require.NoError(t, fx.AddRecord(ctx, l.Id, rec))
+		}
+		fx.assertLogValid(t, l.Id, 4)
+	})
+	t.Run("conflict", func(t *testing.T) {
+		fx := newFixture(t, nil)
+		defer fx.Finish(t)
+		log := consensus.Log{
+			Id: []byte("logTestRecords"),
+			Records: []consensus.Record{
+				{
+					Id: []byte("1"),
+				},
+			},
+		}
+		require.NoError(t, fx.AddLog(ctx, log))
+		assert.Error(t, fx.AddRecord(ctx, log.Id, consensus.Record{Id: []byte("2"), PrevId: []byte("3")}))
+	})
+}
+
+func TestService_FetchLog(t *testing.T) {
+	t.Run("not found", func(t *testing.T) {
+		fx := newFixture(t, nil)
+		defer fx.Finish(t)
+		l, err := fx.FetchLog(ctx, []byte("not exists"))
+		assert.Empty(t, l)
+		assert.ErrorIs(t, err, consensuserrs.ErrLogNotFound)
+	})
+}
+
+func TestService_ChangeReceive(t *testing.T) {
+	t.Run("set after run", func(t *testing.T) {
+		fx := newFixture(t, nil)
+		defer fx.Finish(t)
+		assert.Error(t, fx.SetChangeReceiver(func(logId []byte, records []consensus.Record) {}))
+	})
+	t.Run("receive changes", func(t *testing.T) {
+		var logs = make(chan consensus.Log, 10)
+		var count int
+		fx := newFixture(t, func(logId []byte, records []consensus.Record) {
+			logs <- consensus.Log{Id: logId, Records: records}
+			count++
+		})
+		defer fx.Finish(t)
+		var l = consensus.Log{
+			Id: []byte("logTestStream"),
+			Records: []consensus.Record{
+				{
+					Id: []byte("1"),
+				},
+			},
+		}
+		var records = []consensus.Record{
+			{
+				Id:     []byte("2"),
+				PrevId: []byte("1"),
+			},
+			{
+				Id:     []byte("3"),
+				PrevId: []byte("2"),
+			},
+			{
+				Id:     []byte("4"),
+				PrevId: []byte("3"),
+			},
+		}
+		require.NoError(t, fx.AddLog(ctx, l))
+		assert.Empty(t, count)
+
+		for _, rec := range records {
+			require.NoError(t, fx.AddRecord(ctx, l.Id, rec))
+		}
+
+		timeout := time.After(time.Second)
+		for i := 0; i < len(records); i++ {
+			select {
+			case resLog := <-logs:
+				assertLogValid(t, resLog, i+2)
+			case <-timeout:
+				require.False(t, true)
+			}
+		}
+	})
+}
+
+func newFixture(t *testing.T, cr ChangeReceiver) *fixture {
+	ctx, cancel := context.WithTimeout(ctx, time.Second)
+	fx := &fixture{
+		Service: New(),
+		cancel:  cancel,
+		a:       new(app.App),
+	}
+	fx.a.Register(&config.Config{
+		Mongo: config.Mongo{
+			Connect:       "mongodb://localhost:27017/?w=majority",
+			Database:      "consensus_test",
+			LogCollection: "log",
+		},
+	})
+	fx.a.Register(fx.Service)
+	require.NoError(t, fx.Service.SetChangeReceiver(cr))
+	err := fx.a.Start(ctx)
+	if err != nil {
+		fx.cancel()
+	}
+	require.NoError(t, err)
+	return fx
+}
+
+type fixture struct {
+	Service
+	a      *app.App
+	cancel context.CancelFunc
+}
+
+func (fx *fixture) Finish(t *testing.T) {
+	if fx.cancel != nil {
+		fx.cancel()
+	}
+	coll := fx.Service.(*service).logColl
+	t.Log(coll.Drop(ctx))
+	assert.NoError(t, fx.a.Close(ctx))
+}
+
+func (fx *fixture) assertLogValid(t *testing.T, logId []byte, count int) {
+	log, err := fx.FetchLog(ctx, logId)
+	require.NoError(t, err)
+	assertLogValid(t, log, count)
+}
+
+func assertLogValid(t *testing.T, log consensus.Log, count int) {
+	if count >= 0 {
+		assert.Len(t, log.Records, count)
+	}
+	var prevId []byte
+	for _, rec := range log.Records {
+		if len(prevId) != 0 {
+			assert.Equal(t, string(prevId), string(rec.Id))
+		}
+		prevId = rec.PrevId
+	}
+}

go.mod

@@ -12,12 +12,13 @@ require (
 	github.com/ipfs/go-cid v0.1.0
 	github.com/libp2p/go-libp2p v0.20.3
 	github.com/libp2p/go-libp2p-core v0.16.1
-	github.com/mr-tron/base58 v1.2.0
+	github.com/minio/sha256-simd v1.0.0
 	github.com/multiformats/go-multibase v0.0.3
 	github.com/multiformats/go-multihash v0.1.0
 	github.com/stretchr/testify v1.8.0
 	github.com/zeebo/blake3 v0.2.3
 	github.com/zeebo/errs v1.3.0
+	go.mongodb.org/mongo-driver v1.10.2
 	go.uber.org/zap v1.21.0
 	gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22
 	gopkg.in/yaml.v3 v3.0.1
@@ -33,11 +34,14 @@ 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/snappy v0.0.1 // indirect
+	github.com/klauspost/compress v1.15.1 // 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
 	github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1 // indirect
-	github.com/minio/sha256-simd v1.0.0 // indirect
+	github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe // indirect
+	github.com/mr-tron/base58 v1.2.0 // indirect
 	github.com/multiformats/go-base32 v0.0.3 // indirect
 	github.com/multiformats/go-base36 v0.1.0 // indirect
 	github.com/multiformats/go-multiaddr v0.5.0 // indirect
@@ -47,11 +51,17 @@ 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/xdg-go/pbkdf2 v1.0.0 // indirect
+	github.com/xdg-go/scram v1.1.1 // indirect
+	github.com/xdg-go/stringprep v1.0.3 // indirect
+	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // 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/crypto v0.0.0-20220622213112-05595931fe9d // indirect
 	golang.org/x/image v0.0.0-20200119044424-58c23975cae1 // indirect
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
 	golang.org/x/sys v0.0.0-20220422013727-9388b58f7150 // indirect
+	golang.org/x/text v0.3.7 // 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/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
+github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/google/go-cmp v0.5.2 h1:X2ev0eStA3AbceY54o37/0PQ/UWqKEiiO2dKL5OPaFM=
+github.com/google/go-cmp v0.5.2/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=
@@ -42,6 +46,9 @@ github.com/ipfs/go-cid v0.1.0/go.mod h1:rH5/Xv83Rfy8Rw6xG+id3DYAMUVmem1MowoKwdXm
 github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
+github.com/klauspost/compress v1.15.1 h1:y9FcTHGyrebwfP0ZZqFiaxTaiDnUrGkJkI+f583BL1A=
+github.com/klauspost/compress v1.15.1/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/klauspost/cpuid/v2 v2.0.4/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.0.12 h1:p9dKCg8i4gmOxtv35DvrYoWqYzQrvEVdjQ762Y0OqZE=
@@ -64,6 +71,8 @@ github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1/go.mod h1:pD8Rv
 github.com/minio/sha256-simd v0.1.1-0.20190913151208-6de447530771/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM=
 github.com/minio/sha256-simd v1.0.0 h1:v1ta+49hkWZyvaKwrQB8elexRqm6Y0aMLjCNsrYxo6g=
 github.com/minio/sha256-simd v1.0.0/go.mod h1:OuYzVNI5vcoYIAmbIvHPl3N3jUzVedXbKy5RFepssQM=
+github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe h1:iruDEfMl2E6fbMZ9s0scYfZQ84/6SPL6zC8ACM2oIL0=
+github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
 github.com/mr-tron/base58 v1.1.0/go.mod h1:xcD2VGqlgYjBdcBLw+TuYLr8afG+Hj8g2eTVqeSzSU8=
 github.com/mr-tron/base58 v1.1.3/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
 github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
@@ -102,10 +111,21 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
+github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
+github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.1 h1:VOMT+81stJgXW3CpHyqHN3AXDYIMsx56mEFrB37Mb/E=
+github.com/xdg-go/scram v1.1.1/go.mod h1:RaEWvsqvNKKvBPvcKeFjrG2cJqOkHTiyTpzz23ni57g=
+github.com/xdg-go/stringprep v1.0.3 h1:kdwGpVNwPFtjs98xCGkHjQtGKh86rDcRZN17QEMCOIs=
+github.com/xdg-go/stringprep v1.0.3/go.mod h1:W3f5j4i+9rC0kuIEJL0ky1VpHXQU3ocBgklLGvcBnW8=
+github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d h1:splanxYIlg+5LfHAM6xpdFEAYOk8iySO56hMFq6uLyA=
+github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
@@ -117,6 +137,8 @@ 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.mongodb.org/mongo-driver v1.10.2 h1:4Wk3cnqOrQCn0P92L3/mmurMxzdvWWs5J9jinAVKD+k=
+go.mongodb.org/mongo-driver v1.10.2/go.mod h1:z4XpeoU6w+9Vht+jAFyLgVrD+jGSQQe0+CBWFHNiHt8=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE=
 go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
@@ -134,8 +156,8 @@ golang.org/x/crypto v0.0.0-20200204104054-c9f3fb736b72/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210220033148-5ea612d1eb83/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
 golang.org/x/crypto v0.0.0-20210506145944-38f3c27a63bf/go.mod h1:P+XmwS30IXTQdn5tA2iutPOUgjI07+tq3H3K9MVA1s8=
-golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4 h1:kUhD7nTDoI3fVd9G4ORWrbV5NY0liEs/Jg2pv5f+bBA=
-golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d h1:sK3txAijHtOK88l68nt020reeT1ZdKLIYetKl95FzVY=
+golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/image v0.0.0-20200119044424-58c23975cae1 h1:5h3ngYt7+vXCDZCup/HkCQgW5XwmSvR/nA2JmJ0RErg=
 golang.org/x/image v0.0.0-20200119044424-58c23975cae1/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
@@ -149,9 +171,11 @@ golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
+golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -161,13 +185,18 @@ golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210309074719-68d13333faf2/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220422013727-9388b58f7150 h1:xHms4gcpe1YE7A3yIllJXP16CMAGuqwO2lX1mTyyRRc=
 golang.org/x/sys v0.0.0-20220422013727-9388b58f7150/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
@@ -177,6 +206,7 @@ golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 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 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 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=

pkg/acl/aclrecordproto/aclrecord.pb.go

@@ -324,6 +324,7 @@ func (m *ACLRoot) GetTimestamp() int64 {
 
 type ACLContentValue struct {
 	// Types that are valid to be assigned to Value:
+	//
 	//	*ACLContentValue_UserAdd
 	//	*ACLContentValue_UserRemove
 	//	*ACLContentValue_UserPermissionChange

pkg/acl/testutils/testchanges/proto/test.pb.go

@@ -60,6 +60,7 @@ var xxx_messageInfo_PlainTextChange proto.InternalMessageInfo
 
 type PlainTextChange_Content struct {
 	// Types that are valid to be assigned to Value:
+	//
 	//	*PlainTextChange_Content_TextAppend
 	Value isPlainTextChange_Content_Value `protobuf_oneof:"value"`
 }