chore: Exponential Backoff For Backfill In Mongodb

drivers/base/utils.go

@@ -6,14 +6,14 @@ import (
 	"github.com/datazip-inc/olake/logger"
 )
 
-func RetryOnFailure(attempts int, sleep *time.Duration, f func() error) (err error) {
-	for i := 0; i < attempts; i++ {
+func RetryOnBackoff(attempts int, sleep time.Duration, f func() error) (err error) {
+	for cur := 1; cur <= attempts; cur++ {
 		if err = f(); err == nil {
 			return nil
 		}
-
-		logger.Infof("Retrying after %v...", sleep)
-		time.Sleep(*sleep)
+		logger.Infof("Retrying after %f seconds due to err: %s", sleep.Seconds(), err)
+		time.Sleep(sleep)
+		sleep = sleep * 2
 	}
 
 	return err

drivers/mongodb/README.md

@@ -42,7 +42,8 @@ Add MongoDB credentials in following format in config.json file
       "server-ram": 16,
       "database": "database",
       "max_threads": 50,
-      "default_mode" :"cdc"
+      "default_mode" :"cdc",
+      "backoff_retry_count": 2
    }
 ```
 

drivers/mongodb/internal/backfill.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/datazip-inc/olake/constants"
+	"github.com/datazip-inc/olake/drivers/base"
 	"github.com/datazip-inc/olake/logger"
 	"github.com/datazip-inc/olake/protocol"
 	"github.com/datazip-inc/olake/types"
@@ -74,46 +75,49 @@ func (m *Mongo) backfill(stream protocol.Stream, pool *protocol.WriterPool) erro
 		defer cancelThread()
 
 		opts := options.Aggregate().SetAllowDiskUse(true).SetBatchSize(int32(math.Pow10(6)))
-		cursor, err := collection.Aggregate(ctx, generatepipeline(one.StartID, one.EndID), opts)
-		if err != nil {
-			return fmt.Errorf("collection.Find: %s", err)
-		}
-		defer cursor.Close(ctx)
-
-		waitChannel := make(chan struct{})
-		defer func() {
-			if stream.GetSyncMode() == types.CDC {
-				// only wait in cdc mode
-				// make sure it get called after insert.Close()
-				<-waitChannel
-			}
-			logger.Infof("Finished full load chunk number %d.", number)
-		}()
-
-		insert, err := pool.NewThread(threadContext, stream, protocol.WithNumber(number), protocol.WithWaitChannel(waitChannel))
-		if err != nil {
-			return err
-		}
-		defer insert.Close()
-
-		for cursor.Next(ctx) {
-			var doc bson.M
-			if _, err = cursor.Current.LookupErr("_id"); err != nil {
-				return fmt.Errorf("looking up idProperty: %s", err)
-			} else if err = cursor.Decode(&doc); err != nil {
-				return fmt.Errorf("backfill decoding document: %s", err)
+		mainBackfill := func() error {
+			cursor, err := collection.Aggregate(ctx, generatepipeline(one.StartID, one.EndID), opts)
+			if err != nil {
+				return fmt.Errorf("collection.Find: %s", err)
 			}
-
-			handleObjectID(doc)
-			exit, err := insert.Insert(types.CreateRawRecord(utils.GetKeysHash(doc, constants.MongoPrimaryID), doc, 0))
+			defer cursor.Close(ctx)
+
+			waitChannel := make(chan struct{})
+			defer func() {
+				if stream.GetSyncMode() == types.CDC {
+					// only wait in cdc mode
+					// make sure it get called after insert.Close()
+					<-waitChannel
+				}
+				logger.Infof("Finished full load chunk number %d.", number)
+			}()
+
+			insert, err := pool.NewThread(threadContext, stream, protocol.WithNumber(number), protocol.WithWaitChannel(waitChannel))
 			if err != nil {
-				return fmt.Errorf("failed to finish backfill chunk %d: %s", number, err)
+				return err
 			}
-			if exit {
-				return nil
+			defer insert.Close()
+
+			for cursor.Next(ctx) {
+				var doc bson.M
+				if _, err = cursor.Current.LookupErr("_id"); err != nil {
+					return fmt.Errorf("looking up idProperty: %s", err)
+				} else if err = cursor.Decode(&doc); err != nil {
+					return fmt.Errorf("backfill decoding document: %s", err)
+				}
+
+				handleObjectID(doc)
+				exit, err := insert.Insert(types.CreateRawRecord(utils.GetKeysHash(doc, constants.MongoPrimaryID), doc, 0))
+				if err != nil {
+					return fmt.Errorf("failed to finish backfill chunk %d: %s", number, err)
+				}
+				if exit {
+					return nil
+				}
 			}
+			return cursor.Err()
 		}
-		return cursor.Err()
+		return base.RetryOnBackoff(m.config.RetryCount, 1*time.Minute, mainBackfill)
 	})
 
 }

drivers/mongodb/internal/config.go

@@ -21,6 +21,7 @@ type Config struct {
 	MaxThreads     int            `json:"max_threads"`
 	Database       string         `json:"database"`
 	DefaultMode    types.SyncMode `json:"default_mode"`
+	RetryCount     int            `json:"backoff_retry_count"`
 }
 
 func (c *Config) URI() string {

drivers/mongodb/internal/mon.go

@@ -17,8 +17,9 @@ import (
 )
 
 const (
-	discoverTime   = 5 * time.Minute // maximum time allowed to discover all the streams
-	cdcCursorField = "_data"
+	discoverTime        = 5 * time.Minute // maximum time allowed to discover all the streams
+	cdcCursorField      = "_data"
+	defaultBackoffCount = 3
 )
 
 type Mongo struct {
@@ -54,6 +55,11 @@ func (m *Mongo) Setup() error {
 	m.client = conn
 	// no need to check from discover if it have cdc support or not
 	m.CDCSupport = true
+	// check for default backoff count
+	if m.config.RetryCount == 0 {
+		logger.Info("setting backoff retry count to default value %d", defaultBackoffCount)
+		m.config.RetryCount = defaultBackoffCount
+	}
 	return nil
 }