Merge pull request #747 from datazip-inc/staging

chore: staging -> master v0.3.13

Author: vikaxsh

drivers/mongodb/internal/cdc.go

@@ -56,19 +56,16 @@ func (m *Mongo) PreCDC(cdcCtx context.Context, streams []types.StreamInterface)
 		}
 		m.cdcCursor.Store(stream.ID(), prevResumeToken)
 	}
+	return nil
+}
 
-	// TODO:move it to stream change function
-	lastOplogTime, err := m.getClusterOpTime(cdcCtx, m.config.Database)
+func (m *Mongo) StreamChanges(ctx context.Context, stream types.StreamInterface, OnMessage abstract.CDCMsgFn) error {
+	// lastOplogTime is the latest timestamp of any operation applied in the MongoDB cluster
+	lastOplogTime, err := m.getClusterOpTime(ctx, m.config.Database)
 	if err != nil {
 		logger.Warnf("Failed to get cluster op time: %s", err)
 		return err
 	}
-	m.LastOplogTime = lastOplogTime
-
-	return nil
-}
-
-func (m *Mongo) StreamChanges(ctx context.Context, stream types.StreamInterface, OnMessage abstract.CDCMsgFn) error {
 	pipeline := mongo.Pipeline{
 		{{Key: "$match", Value: bson.D{
 			{Key: "operationType", Value: bson.D{{Key: "$in", Value: bson.A{"insert", "update", "delete"}}}},
@@ -92,32 +89,36 @@ func (m *Mongo) StreamChanges(ctx context.Context, stream types.StreamInterface,
 	defer cursor.Close(ctx)
 
 	for {
-		if !cursor.TryNext(ctx) {
-			if err := cursor.Err(); err != nil {
-				return fmt.Errorf("change stream error: %s", err)
-			}
+		hasNext := cursor.TryNext(ctx)
+		if err := cursor.Err(); err != nil {
+			return fmt.Errorf("change stream error: %s", err)
+		}
 
-			// PBRT checkpoint and termination check
-			if err := m.handleIdleCheckpoint(ctx, cursor, stream); err != nil {
-				if errors.Is(err, ErrIdleTermination) {
-					// graceful termination requested by helper
-					logger.Infof("change stream %s caught up to cluster opTime; terminating gracefully", stream.ID())
-					return nil
-				}
+		if hasNext {
+			if err := m.handleChangeDoc(ctx, cursor, stream, OnMessage); err != nil {
 				return err
 			}
-			// Wait before for a brief pause before the next iteration of the loop
-			time.Sleep(changeStreamRetryDelay)
-			continue
 		}
 
-		if err := m.handleChangeDoc(ctx, cursor, stream, OnMessage); err != nil {
+		// Check boundary AFTER emitting
+		if err := m.handleStreamCatchup(ctx, cursor, stream, lastOplogTime); err != nil {
+			if errors.Is(err, ErrIdleTermination) {
+				// graceful termination requested by helper
+				logger.Infof("change stream %s caught up to cluster opTime; terminating gracefully", stream.ID())
+				return nil
+			}
 			return err
 		}
+
+		if !hasNext {
+			// Wait before for a brief pause before the next iteration of the loop
+			// TryNext() method behaves differently when connecting through mongos vs. direct replica set connections:TryNext() is non-blocking and may return false immediately even when events exist
+			time.Sleep(changeStreamRetryDelay)
+		}
 	}
 }
 
-func (m *Mongo) handleIdleCheckpoint(_ context.Context, cursor *mongo.ChangeStream, stream types.StreamInterface) error {
+func (m *Mongo) handleStreamCatchup(_ context.Context, cursor *mongo.ChangeStream, stream types.StreamInterface, lastOplogTime primitive.Timestamp) error {
 	finalToken := cursor.ResumeToken()
 	if finalToken == nil {
 		return fmt.Errorf("no resume token available for stream %s after TryNext", stream.ID())
@@ -138,7 +139,7 @@ func (m *Mongo) handleIdleCheckpoint(_ context.Context, cursor *mongo.ChangeStre
 	}
 
 	// If stream is caught up -> request graceful termination
-	if !m.LastOplogTime.After(streamOpTime) {
+	if !lastOplogTime.After(streamOpTime) {
 		return ErrIdleTermination
 	}
 

drivers/mongodb/internal/mon.go

@@ -26,13 +26,12 @@ const (
 )
 
 type Mongo struct {
-	config        *Config
-	client        *mongo.Client
-	CDCSupport    bool // indicates if the MongoDB instance supports Change Streams
-	cdcCursor     sync.Map
-	state         *types.State        // reference to globally present state
-	LastOplogTime primitive.Timestamp // Cluster opTime is the latest timestamp of any operation applied in the MongoDB cluster
-	sshDialer     *MongoSSHDialer
+	config     *Config
+	client     *mongo.Client
+	CDCSupport bool // indicates if the MongoDB instance supports Change Streams
+	cdcCursor  sync.Map
+	state      *types.State // reference to globally present state
+	sshDialer  *MongoSSHDialer
 }
 
 // MongoSSHDialer implements a custom dialer for SSH tunnel connections.

drivers/postgres/internal/cdc.go

@@ -97,13 +97,9 @@ func (p *Postgres) PostCDC(ctx context.Context, _ types.StreamInterface, noErr b
 	return nil
 }
 
-func doesReplicationSlotExists(ctx context.Context, conn *sqlx.DB, slotName string, publication string) (bool, error) {
+func doesReplicationSlotExists(ctx context.Context, conn *sqlx.DB, slotName string, publication string, database string) (bool, error) {
 	var exists bool
-	err := conn.QueryRowContext(
-		ctx,
-		"SELECT EXISTS(Select 1 from pg_replication_slots where slot_name = $1)",
-		slotName,
-	).Scan(&exists)
+	err := conn.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = $1 AND database = current_database())`, slotName).Scan(&exists)
 	if err != nil {
 		return false, err
 	}

drivers/postgres/internal/postgres.go

@@ -115,7 +115,7 @@ func (p *Postgres) Setup(ctx context.Context) error {
 
 		logger.Infof("CDC initial wait time set to: %d", cdc.InitialWaitTime)
 
-		exists, err := doesReplicationSlotExists(ctx, pgClient, cdc.ReplicationSlot, cdc.Publication)
+		exists, err := doesReplicationSlotExists(ctx, pgClient, cdc.ReplicationSlot, cdc.Publication, p.config.Database)
 		if err != nil {
 			if strings.Contains(err.Error(), "sql: no rows in result set") {
 				err = fmt.Errorf("no record found")
@@ -124,7 +124,7 @@ func (p *Postgres) Setup(ctx context.Context) error {
 		}
 
 		if !exists {
-			return fmt.Errorf("provided replication slot %s does not exist", cdc.ReplicationSlot)
+			return fmt.Errorf("replication slot '%s' does not exist in the current database '%s'", cdc.ReplicationSlot, p.config.Database)
 		}
 		// no use of it if check not being called while sync run
 		p.CDCSupport = true