Add fanout params for mongodb planning stage

[adiom-mark]

Summary by CodeRabbit

• New Features

  • Added namespace-fanout and documentdb-sampling-fanout configuration options for MongoDB and DocumentDB connectors with configurable defaults.

• Tests

  • Added unit tests verifying fanout limit configuration defaults and custom settings.

[coderabbitai]

📝 Walkthrough

Walkthrough

The PR introduces configurable concurrency limits for MongoDB connector planning and DocumentDB sampling operations. New NamespaceFanout and DocumentDBSamplingFanout settings are added to control errgroup fanout; defaults are applied in NewConn, populated from CLI flags via a helper, and enforced in planning and sampling code paths.

Changes

MongoDB Connector Fanout Concurrency Control

Layer / File(s)	Summary
Proto schema and connector settings proto/connectorcommands/adiom/commands/connectors/v1/mongo.proto, connectors/mongo/conn.go	MongoBaseFlags adds namespace_fanout field; ConnectorSettings gains NamespaceFanout and DocumentDBSamplingFanout int fields with corresponding default limit constants; NewConn sets defaults for the new fields.
Planning fanout concurrency control connectors/mongo/conn.go	planningFanoutLimit() method computes namespace planning concurrency cap from NamespaceFanout or default; GeneratePlan enforces the limit via eg.SetLimit(); debug log added when computing sample counts during initial sync partition planning.
DocumentDB sampling concurrency control connectors/mongo/docdb.go	sampleIDs applies eg.SetLimit() using documentDBSamplingFanout() helper to constrain concurrent document sampling goroutines in DocumentDB flavor.
CLI flag wiring and initialization internal/app/options/connectorflags.go	mongoFlagsCommandFlags() helper extends generated Mongo flags with hidden documentdb-sampling-fanout flag; MongoDB connector registration uses the helper and populates DocumentDBSamplingFanout from CLI flag before connector creation.
Unit tests for fanout configuration connectors/mongo/conn_unit_test.go	TestPlanningFanoutLimit verifies planning fanout defaults and NamespaceFanout overrides; TestDocumentDBSamplingFanout verifies sampling fanout defaults and DocumentDBSamplingFanout overrides across MongoDB and DocumentDB flavors.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

Possibly related PRs

• adiom-data/dsync#378: Main PR extends commandargs-based Mongo connector CLI and proto wiring that was introduced in this PR.
• adiom-data/dsync#374: Main PR adds DocumentDB-specific sampleIDs concurrency control that directly extends the retrieved PR's new docdb.go implementation.
• adiom-data/dsync#336: Both PRs extend ConnectorSettings with new fields that modify MongoDB sampling and planning behavior.

Poem

🐰 Goroutines dance with ordered grace,
Fanout limits keep pace,
Namespaces plan, samples flow,
Concurrency tamed, both fast and slow. ✨

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 12.50% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately summarizes the main change: adding fanout parameters to control concurrency in the MongoDB planning stage.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch mdb-plan-fanout

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

🧹 Nitpick comments (2)

internal/app/options/connectorflags.go (1)

734-741: 💤 Low value

Optional: avoid the hardcoded 100 default.

The hidden flag duplicates the magic number 100, which can silently diverge from mongo.defaultDocumentDBSamplingFanoutLimit. Since the connector already applies that default in NewConn via setDefault, you could drop the flag's Value (or export/reuse the constant) so there is a single source of truth. Also note this flag is wired manually here while namespace-fanout flows through the proto/generated path — intentional given it's hidden, but worth keeping in mind for consistency.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@internal/app/options/connectorflags.go` around lines 734 - 741, The hidden
flag in mongoFlagsCommandFlags currently hardcodes Value: 100 for
"documentdb-sampling-fanout"; instead either remove the Value so the connector's
NewConn/setDefault logic (mongo.defaultDocumentDBSamplingFanoutLimit) remains
the single source of truth, or reuse/export that constant for the flag's
default; update mongoFlagsCommandFlags to not duplicate the magic number
(reference: function mongoFlagsCommandFlags, flag name
"documentdb-sampling-fanout", NewConn and setDefault, and
mongo.defaultDocumentDBSamplingFanoutLimit).

connectors/mongo/conn.go (1)

249-249: Note the multiplicative fanout for DocumentDB.

eg.SetLimit(c.planningFanoutLimit()) bounds concurrent namespace planning, but each planning goroutine calls c.sampleIDs, which (for the DocumentDB flavor) opens its own errgroup bounded by documentDBSamplingFanout(). With both defaulting to 100, peak concurrency reaches ~100 × 100 = 10000 simultaneous $sample aggregations against the same cluster, which can exhaust the driver connection pool and pressure the server. This is still a large improvement over the previously unbounded fanout, so no change is required, but consider whether the effective product should be capped or documented.

Also applies to: 300-303

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@connectors/mongo/conn.go` at line 249, The current concurrency bounding uses
eg.SetLimit(c.planningFanoutLimit()) but each planner goroutine calls
c.sampleIDs which itself creates an inner errgroup bounded by
documentDBSamplingFanout(), producing a multiplicative fanout (eg.SetLimit ×
documentDBSamplingFanout) that can overwhelm the driver; update the logic so the
effective parallelism is capped or controlled—either reduce one of the limits,
enforce a global cap (e.g., compute min(c.planningFanoutLimit(),
maxGlobalSampling/concurrentSampleLimit) before calling eg.SetLimit), or add
coordination so c.sampleIDs uses a shared semaphore instead of its own
uncoordinated errgroup; locate and modify eg.SetLimit(...) and c.sampleIDs (and
documentDBSamplingFanout()) to implement the chosen cap or add documentation
explaining the multiplicative effect and recommended defaults.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@connectors/mongo/conn.go`:
- Line 249: The current concurrency bounding uses
eg.SetLimit(c.planningFanoutLimit()) but each planner goroutine calls
c.sampleIDs which itself creates an inner errgroup bounded by
documentDBSamplingFanout(), producing a multiplicative fanout (eg.SetLimit ×
documentDBSamplingFanout) that can overwhelm the driver; update the logic so the
effective parallelism is capped or controlled—either reduce one of the limits,
enforce a global cap (e.g., compute min(c.planningFanoutLimit(),
maxGlobalSampling/concurrentSampleLimit) before calling eg.SetLimit), or add
coordination so c.sampleIDs uses a shared semaphore instead of its own
uncoordinated errgroup; locate and modify eg.SetLimit(...) and c.sampleIDs (and
documentDBSamplingFanout()) to implement the chosen cap or add documentation
explaining the multiplicative effect and recommended defaults.

In `@internal/app/options/connectorflags.go`:
- Around line 734-741: The hidden flag in mongoFlagsCommandFlags currently
hardcodes Value: 100 for "documentdb-sampling-fanout"; instead either remove the
Value so the connector's NewConn/setDefault logic
(mongo.defaultDocumentDBSamplingFanoutLimit) remains the single source of truth,
or reuse/export that constant for the flag's default; update
mongoFlagsCommandFlags to not duplicate the magic number (reference: function
mongoFlagsCommandFlags, flag name "documentdb-sampling-fanout", NewConn and
setDefault, and mongo.defaultDocumentDBSamplingFanoutLimit).

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: d46922df-f4ee-4c39-bc23-9bd0705aecb0

📥 Commits

Reviewing files that changed from the base of the PR and between b49ed34 and 8af67b0.

⛔ Files ignored due to path filters (3)

• gen/adiom/commands/connectors/v1/cosmos_commandargs.go is excluded by !**/gen/**
• gen/adiom/commands/connectors/v1/mongo.pb.go is excluded by !**/*.pb.go, !**/gen/**
• gen/adiom/commands/connectors/v1/mongo_commandargs.go is excluded by !**/gen/**

📒 Files selected for processing (5)

• connectors/mongo/conn.go
• connectors/mongo/conn_unit_test.go
• connectors/mongo/docdb.go
• internal/app/options/connectorflags.go
• proto/connectorcommands/adiom/commands/connectors/v1/mongo.proto