fixes

Author: jgao54

flow/connectors/mongo/cdc.go

@@ -326,13 +326,15 @@ func (c *MongoConnector) PullRecords(
 				if err := recreateChangeStream(false); err != nil {
 					return fmt.Errorf("failed to recreate change stream: %w", err)
 				}
+				c.logger.Info("[mongo] recreated change stream because context deadline exceeded", slog.Duration("elapsed", time.Since(pullStart)))
 				continue
 			}
 
 			if isResumeTokenNotFoundError(err) {
 				if err := recreateChangeStream(true); err != nil {
 					return fmt.Errorf("failed to recreate change stream: %w", err)
 				}
+				c.logger.Info("[mongo] recreated change stream because resume token not found", slog.Duration("elapsed", time.Since(pullStart)))
 				continue
 			}
 
@@ -462,14 +464,19 @@ func createPipeline(tableNameMapping map[string]model.NameAndExclude) (mongo.Pip
 	return pipeline, nil
 }
 
-// createChangeStream calls client.Watch with a 5 minute context deadline
-// so the driver sends it as maxTimeMS over the wire. Otherwise server-
-// side default maxTimeMS is used which can sometimes be too short.
-func createChangeStream(client *mongo.Client, parent context.Context, pipeline mongo.Pipeline, opts *options.ChangeStreamOptionsBuilder,
+// createChangeStream calls client.Watch with a context deadline so the driver
+// sends it as maxTimeMS over the wire. Without this, the server-side default
+// maxTimeMS may be too short for the initial aggregate.
+func createChangeStream(
+	client *mongo.Client, parent context.Context, pipeline mongo.Pipeline, opts *options.ChangeStreamOptionsBuilder,
 ) (*mongo.ChangeStream, error) {
 	watchCtx, cancel := context.WithTimeout(parent, 5*time.Minute)
 	defer cancel()
-	return client.Watch(watchCtx, pipeline, opts)
+	cs, err := client.Watch(watchCtx, pipeline, opts)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open change stream: %w", err)
+	}
+	return cs, nil
 }
 
 // This can happen if the resumeToken we are attempting to `ResumeAfter` refers to a table that has been