Skip to content

Fail fast on mixed id types for mongo connector. #382

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
adiom-mark merged 1 commit into from
Apr 29, 2026
Merged

Fail fast on mixed id types for mongo connector. #382

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 28 additions & 8 deletions connectors/mongo/conn.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,21 @@ func findLargestID(ctx context.Context, col *mongo.Collection) (bson.RawValue, e
return bson.RawValue{}, nil
}

func findSmallestID(ctx context.Context, col *mongo.Collection) (bson.RawValue, error) {
res, err := col.Find(ctx, bson.M{}, options.Find().SetSort(bson.M{"_id": 1}).SetLimit(1).SetProjection(bson.D{{"_id", 1}}))
if err != nil {
return bson.RawValue{}, fmt.Errorf("err in find: %w", err)
}
defer res.Close(ctx)
for res.Next(ctx) {
return res.Current.Lookup("_id"), nil
}
if res.Err() != nil {
return bson.RawValue{}, fmt.Errorf("err in iterating: %w", res.Err())
}
return bson.RawValue{}, nil
}

// GeneratePlan implements adiomv1connect.ConnectorServiceHandler.
func (c *conn) GeneratePlan(ctx context.Context, r *connect.Request[adiomv1.GeneratePlanRequest]) (*connect.Response[adiomv1.GeneratePlanResponse], error) {
partitions, err := NamespacePartitions(ctx, r.Msg.GetNamespaces(), c.client)
Expand Down Expand Up @@ -241,15 +256,24 @@ func (c *conn) GeneratePlan(ctx context.Context, r *connect.Request[adiomv1.Gene
eg.Go(func() error {
ns, _ := ToNS(partition.Namespace)
col := c.client.Database(ns.Db).Collection(ns.Col)

lowest, err := findSmallestID(ctx, col)
if err != nil {
return fmt.Errorf("err finding smallest id: %w", err)
}
high, err := findLargestID(ctx, col)
if err != nil {
return fmt.Errorf("err finding largest id: %w", err)
}
if !lowest.IsZero() && !high.IsZero() && lowest.Type != high.Type {
return fmt.Errorf("mixed _id types not supported in %v: found %v and %v", partition.Namespace, lowest.Type, high.Type)
}
Comment on lines +260 to +270

@coderabbitai coderabbitai Bot Apr 28, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Fail-fast mixed _id validation should use the same filter scope as sync reads.

At Line 260-Line 270, _id type validation runs on the full collection, while read path filtering uses c.query (Line 460-Line 463). This can fail planning for a valid filtered sync if mixed types exist outside the filter.

Proposed fix 
-			lowest, err := findSmallestID(ctx, col)
+			planFilter := bson.D{}
+			if len(c.query) > 0 {
+				planFilter = append(planFilter, c.query...)
+			}
+			lowest, err := findSmallestID(ctx, col, planFilter)
 			if err != nil {
 				return fmt.Errorf("err finding smallest id: %w", err)
 			}
-			high, err := findLargestID(ctx, col)
+			high, err := findLargestID(ctx, col, planFilter)
 			if err != nil {
 				return fmt.Errorf("err finding largest id: %w", err)
 			}
// Update helpers to accept a filter:
func findSmallestID(ctx context.Context, col *mongo.Collection, filter bson.D) (bson.RawValue, error)
func findLargestID(ctx context.Context, col *mongo.Collection, filter bson.D) (bson.RawValue, error)
🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@connectors/mongo/conn.go` around lines 260 - 270, The _id type validation
currently checks the entire collection but the sync read path uses the filtered
query (c.query), so change the validation to use the same filter scope: update
findSmallestID and findLargestID to accept a filter parameter (e.g., bson.D) and
pass c.query (the same filter used in the read path) when calling them from the
validation block that currently references partition.Namespace; ensure the
implementations use the provided filter in their Mongo queries so mixed _id
types outside the filter won't cause a false-positive error.


count, err := col.EstimatedDocumentCount(ctx)
if err != nil {
return err
}
if count < c.settings.TargetDocCountPerPartition*2 {
high, err := findLargestID(ctx, col)
if err != nil {
return fmt.Errorf("err finding largest id: %w", err)
}
ch <- &adiomv1.Partition{
Namespace: partition.GetNamespace(),
EstimatedCount: uint64(count),
Expand Down Expand Up @@ -284,10 +308,6 @@ func (c *conn) GeneratePlan(ctx context.Context, r *connect.Request[adiomv1.Gene
}
low = high
}
high, err := findLargestID(ctx, col)
if err != nil {
return fmt.Errorf("err finding largest id: %w", err)
}
ch <- &adiomv1.Partition{
Namespace: partition.GetNamespace(),
EstimatedCount: uint64(c.settings.TargetDocCountPerPartition),
Expand Down
Loading