[PML-49] recovery on OS process crash (#32)

Author: defbin

config/const.go

@@ -20,9 +20,23 @@ const (
 const (
 	// MongoLinkDatabase is the name of the MongoDB database used by MongoLink.
 	MongoLinkDatabase = "percona_mongolink"
+	// TickCollection is the name of the collection used for ticks during change replication.
+	TickCollection = "ticks"
+	// RecoveryCollection is the name of the collection used for recovery data.
+	RecoveryCollection = "checkpoints"
+	// HeartbeatCollection is the name of the collection used for heartbeats.
+	HeartbeatCollection = "heartbeats"
+)
 
-	// TickCollection is the name of the collection used for ticks.
-	TickCollection = "tick"
+const (
+	// RecoveryCheckpointingInternal is the interval for recovery checkpointing.
+	RecoveryCheckpointingInternal = time.Minute
+	// HeartbeatInternal is the interval for heartbeats.
+	HeartbeatInternal = 30 * time.Second
+	// HeartbeatTimeout is the timeout duration for heartbeats.
+	HeartbeatTimeout = 5 * time.Second
+	// StaleHeartbeatDuration is the duration after which a heartbeat is considered stale.
+	StaleHeartbeatDuration = HeartbeatInternal + HeartbeatTimeout
 )
 
 // Change stream and replication settings.

heartbeat.go

@@ -0,0 +1,130 @@
+package main
+
+import (
+	"context"
+	"time"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+
+	"github.com/percona-lab/percona-mongolink/config"
+	"github.com/percona-lab/percona-mongolink/errors"
+	"github.com/percona-lab/percona-mongolink/log"
+)
+
+var errConcurrentProcess = errors.New("detected concurrent process")
+
+const heartbeatID = "mongolink"
+
+type StopHeartbeat func(context.Context) error
+
+func RunHeartbeat(ctx context.Context, m *mongo.Client) (StopHeartbeat, error) {
+	lastBeat, err := doFirstHeartbeat(ctx, m)
+	if err != nil {
+		return nil, errors.Wrap(err, "first")
+	}
+
+	lg := log.New("heartbeat")
+	lg.With(log.Int64("hb", lastBeat)).Trace("")
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	go func() {
+		lastBeat := lastBeat
+
+		for {
+			time.Sleep(config.HeartbeatInternal)
+
+			savedBeat, err := doHeartbeat(ctx, m, lastBeat)
+			if err != nil {
+				lg.Error(err, "beat")
+
+				continue
+			}
+
+			lastBeat = savedBeat
+			lg.With(log.Int64("hb", lastBeat)).Trace("")
+		}
+	}()
+
+	stop := func(ctx context.Context) error {
+		cancel()
+
+		return DeleteHeartbeat(ctx, m)
+	}
+
+	return stop, nil
+}
+
+func doFirstHeartbeat(ctx context.Context, m *mongo.Client) (int64, error) {
+	timeoutCtx, cancel := context.WithTimeout(ctx, config.HeartbeatTimeout)
+	defer cancel()
+
+	currBeat := time.Now().Unix()
+
+	_, err := m.Database(config.MongoLinkDatabase).
+		Collection(config.HeartbeatCollection).
+		InsertOne(timeoutCtx, bson.D{{"_id", heartbeatID}, {"time", currBeat}})
+	if err == nil {
+		return currBeat, nil
+	}
+
+	if !mongo.IsDuplicateKeyError(err) {
+		return 0, err //nolint:wrapcheck
+	}
+
+	raw, err := m.Database(config.MongoLinkDatabase).
+		Collection(config.HeartbeatCollection).
+		FindOne(ctx, bson.D{{"_id", heartbeatID}}).
+		Raw()
+	if err != nil {
+		return 0, errors.Wrap(err, "find")
+	}
+
+	lastBeat, _ := raw.Lookup("time").AsInt64OK()
+
+	if time.Since(time.Unix(lastBeat, 0)) < config.StaleHeartbeatDuration {
+		return 0, errConcurrentProcess
+	}
+
+	currBeat, err = doHeartbeat(ctx, m, lastBeat)
+	if err != nil {
+		return 0, errors.Wrap(err, "beat")
+	}
+
+	return currBeat, nil
+}
+
+func doHeartbeat(ctx context.Context, m *mongo.Client, lastBeat int64) (int64, error) {
+	timeoutCtx, cancel := context.WithTimeout(ctx, config.HeartbeatTimeout)
+	defer cancel()
+
+	currBeat := time.Now().Unix()
+
+	raw, err := m.Database(config.MongoLinkDatabase).
+		Collection(config.HeartbeatCollection).
+		FindOneAndUpdate(timeoutCtx,
+			bson.D{{"_id", heartbeatID}},
+			bson.D{{"$set", bson.D{{"time", currBeat}}}},
+			options.FindOneAndUpdate().SetReturnDocument(options.Before)).
+		Raw()
+	if err != nil {
+		return 0, err //nolint:wrapcheck
+	}
+
+	savedBeat, _ := raw.Lookup("time").AsInt64OK()
+	if savedBeat != lastBeat {
+		return 0, errConcurrentProcess
+	}
+
+	return currBeat, nil
+}
+
+func DeleteHeartbeat(ctx context.Context, m *mongo.Client) error {
+	_, err := m.Database(config.MongoLinkDatabase).
+		Collection(config.HeartbeatCollection).
+		DeleteOne(ctx, bson.D{{"_id", heartbeatID}})
+
+	return err //nolint:wrapcheck
+}

log/log.go

@@ -85,6 +85,12 @@ func Tx(txn *int64, lsid []byte) AttrFn {
 	}
 }
 
+func Int64(key string, val int64) AttrFn {
+	return func(l zerolog.Context) zerolog.Context {
+		return l.Int64(key, val)
+	}
+}
+
 // New creates a new Logger with the specified scope.
 func New(scope string) Logger {
 	log := zerolog.Ctx(context.Background()).With().Logger()