Merge pull request #68 from anyproto/GO-6351-deletion-log

GO-6351 deletion log

Author: cheggaaa

Makefile

@@ -1,6 +1,6 @@
 .PHONY: proto build test deps
 export GOPRIVATE=github.com/anyproto
-export PATH:=deps:$(PATH)
+export PATH:=$(CURDIR)/deps:$(PATH)
 BUILD_GOOS:=$(shell go env GOOS)
 BUILD_GOARCH:=$(shell go env GOARCH)
 
@@ -14,3 +14,8 @@ test:
 deps:
 	go mod download
 	go build -o deps github.com/ahmetb/govvv
+	go build -o deps go.uber.org/mock/mockgen
+
+mocks:
+	echo 'Generating mocks...'
+	go generate ./...

cmd/consensusnode.go

@@ -30,6 +30,7 @@ import (
 	"github.com/anyproto/any-sync-consensusnode/config"
 	"github.com/anyproto/any-sync-consensusnode/consensusrpc"
 	"github.com/anyproto/any-sync-consensusnode/db"
+	"github.com/anyproto/any-sync-consensusnode/deletelog"
 	"github.com/anyproto/any-sync-consensusnode/stream"
 	// import this to keep govvv in go.mod on mod tidy
 	_ "github.com/ahmetb/govvv/integration-test/app-different-package/mypkg"
@@ -115,5 +116,6 @@ func Bootstrap(a *app.App) {
 		Register(server.New()).
 		Register(db.New()).
 		Register(stream.New()).
-		Register(consensusrpc.New())
+		Register(consensusrpc.New()).
+		Register(deletelog.New())
 }

config/config.go

@@ -1,6 +1,8 @@
 package config
 
 import (
+	"os"
+
 	commonaccount "github.com/anyproto/any-sync/accountservice"
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/app/logger"
@@ -10,7 +12,9 @@ import (
 	"github.com/anyproto/any-sync/net/transport/yamux"
 	"github.com/anyproto/any-sync/nodeconf"
 	"gopkg.in/yaml.v3"
-	"os"
+
+	"github.com/anyproto/any-sync-consensusnode/db"
+	"github.com/anyproto/any-sync-consensusnode/deletelog"
 )
 
 const CName = "config"
@@ -33,11 +37,12 @@ type Config struct {
 	Network                  nodeconf.Configuration `yaml:"network"`
 	NetworkStorePath         string                 `yaml:"networkStorePath"`
 	NetworkUpdateIntervalSec int                    `yaml:"networkUpdateIntervalSec"`
-	Mongo                    Mongo                  `yaml:"mongo"`
+	Mongo                    db.Config              `yaml:"mongo"`
 	Metric                   metric.Config          `yaml:"metric"`
 	Log                      logger.Config          `yaml:"log"`
 	Yamux                    yamux.Config           `yaml:"yamux"`
 	Quic                     quic.Config            `yaml:"quic"`
+	Deletion                 deletelog.Config       `yaml:"deletion"`
 }
 
 func (c *Config) Init(a *app.App) (err error) {
@@ -48,7 +53,7 @@ func (c Config) Name() (name string) {
 	return CName
 }
 
-func (c Config) GetMongo() Mongo {
+func (c Config) GetMongo() db.Config {
 	return c.Mongo
 }
 
@@ -83,3 +88,11 @@ func (c Config) GetYamux() yamux.Config {
 func (c Config) GetQuic() quic.Config {
 	return c.Quic
 }
+
+func (c Config) GetDeletion() deletelog.Config {
+	return c.Deletion
+}
+
+func (c Config) GetDB() db.Config {
+	return c.Mongo
+}

config/mongo.go db/config.goconfig/mongo.go renamed to db/config.go

@@ -1,6 +1,10 @@
-package config
+package db
 
-type Mongo struct {
+type configGetter interface {
+	GetDB() Config
+}
+
+type Config struct {
 	Connect       string `yaml:"connect"`
 	Database      string `yaml:"database"`
 	LogCollection string `yaml:"logCollection"`

db/db.go

@@ -3,23 +3,26 @@ package db
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/anyproto/any-sync/consensus/consensusproto/consensuserr"
+	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
 	"go.mongodb.org/mongo-driver/mongo/readpref"
 	"go.uber.org/zap"
 
 	consensus "github.com/anyproto/any-sync-consensusnode"
-	"github.com/anyproto/any-sync-consensusnode/config"
 )
 
 const CName = "consensus.db"
 
+const settingsColl = "settings"
+
 var log = logger.NewNamed(CName)
 
 func New() Service {
@@ -40,12 +43,17 @@ type Service interface {
 	FetchLog(ctx context.Context, logId string) (log consensus.Log, err error)
 	// SetChangeReceiver sets the receiver for updates, it must be called before app.Run stage
 	SetChangeReceiver(receiver ChangeReceiver) (err error)
+	// SetDeletionId sets the last deleted log id
+	SetDeletionId(ctx context.Context, lastId string) (err error)
+	// GetDeletionId gets the last deletion log id
+	GetDeletionId(ctx context.Context) (lastId string, err error)
 	app.ComponentRunnable
 }
 
 type service struct {
-	conf           config.Mongo
+	conf           Config
 	logColl        *mongo.Collection
+	settingsColl   *mongo.Collection
 	running        bool
 	changeReceiver ChangeReceiver
 
@@ -55,7 +63,7 @@ type service struct {
 }
 
 func (s *service) Init(a *app.App) (err error) {
-	s.conf = a.MustComponent(config.CName).(*config.Config).Mongo
+	s.conf = a.MustComponent("config").(configGetter).GetDB()
 	return nil
 }
 
@@ -80,6 +88,7 @@ func (s *service) Run(ctx context.Context) (err error) {
 			return err
 		}
 	}
+	s.settingsColl = client.Database(s.conf.Database).Collection(settingsColl)
 	return
 }
 
@@ -200,6 +209,23 @@ func (s *service) streamListener(stream *mongo.ChangeStream) {
 	}
 }
 
+func (s *service) SetDeletionId(ctx context.Context, lastId string) (err error) {
+	updateOpts := options.Update().SetUpsert(true)
+	_, err = s.settingsColl.UpdateOne(ctx, bson.D{{"_id", "settings"}}, bson.D{{"$set", bson.D{{"logId", lastId}}}}, updateOpts)
+	return
+}
+
+func (s *service) GetDeletionId(ctx context.Context) (lastId string, err error) {
+	var res struct {
+		LogId string `bson:"logId"`
+	}
+	err = s.settingsColl.FindOne(ctx, bson.D{{"_id", "settings"}}).Decode(&res)
+	if errors.Is(err, mongo.ErrNoDocuments) {
+		err = nil
+	}
+	return res.LogId, err
+}
+
 func (s *service) Close(ctx context.Context) (err error) {
 	if s.logColl != nil {
 		err = s.logColl.Database().Client().Disconnect(ctx)

db/db_test.go

@@ -2,14 +2,15 @@ package db
 
 import (
 	"context"
-	consensus "github.com/anyproto/any-sync-consensusnode"
-	"github.com/anyproto/any-sync-consensusnode/config"
+	"testing"
+	"time"
+
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/consensus/consensusproto/consensuserr"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"testing"
-	"time"
+
+	consensus "github.com/anyproto/any-sync-consensusnode"
 )
 
 var ctx = context.Background()
@@ -180,20 +181,33 @@ func TestService_ChangeReceive(t *testing.T) {
 	})
 }
 
+func TestService_SetDeletionId(t *testing.T) {
+	t.Run("empty", func(t *testing.T) {
+		fx := newFixture(t, nil)
+		defer fx.Finish(t)
+		_ = fx.Service.(*service).settingsColl.Drop(ctx)
+		recId, err := fx.GetDeletionId(ctx)
+		assert.Empty(t, recId)
+		assert.NoError(t, err)
+	})
+	t.Run("set get", func(t *testing.T) {
+		fx := newFixture(t, nil)
+		defer fx.Finish(t)
+		require.NoError(t, fx.SetDeletionId(ctx, "123"))
+		logId, err := fx.GetDeletionId(ctx)
+		require.NoError(t, err)
+		assert.Equal(t, "123", logId)
+	})
+}
+
 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(&testConfig{})
 	fx.a.Register(fx.Service)
 	require.NoError(t, fx.Service.SetChangeReceiver(cr))
 	err := fx.a.Start(ctx)
@@ -237,3 +251,15 @@ func assertLogValid(t *testing.T, log consensus.Log, count int) {
 		prevId = rec.PrevId
 	}
 }
+
+type testConfig struct {
+	Config
+}
+
+func (c *testConfig) Init(a *app.App) error { return nil }
+
+func (c *testConfig) Name() string { return "config" }
+
+func (c *testConfig) GetDB() Config {
+	return Config{Connect: "mongodb://localhost:27017/?w=majority", Database: "consensus_test", LogCollection: "log"}
+}