mssqlserver_cdc: support table discovery via cdc.change_tables

[josephwoodward]

If discovering change tables using the default naming convention (cdc.<schema>_<tablename>_CT) fails to resolve to any table then the SQL Server CDC connector will look for matching tables using the cdc.change_tables table.

[claude]

Explicit capture_instance is silently discarded when a convention-named instance exists.

The convention-name branch returns before the override != "" check below, so a user who sets capture_instance: dbo_orders_v2 on a table whose two instances are dbo_orders (convention-named) and dbo_orders_v2 keeps streaming from dbo_orders, and nothing in the warning mentions that their configured value was ignored (asserted by ConventionNamePreferredOverOverride in stream_integration_test.go). That is exactly the "temporary second instance during an online schema change" cutover the field description names as the motivating case, and it leaves no way to select the new instance until the old one is dropped.

Either honour an exactly-matching capture_instance ahead of the convention name, or — if the current precedence is deliberate — include the ignored override in the warning text so the operator can see why their setting had no effect.

CONTRIBUTING.md §1.1.3 ("UX should be intuitive… don't make me think") and §3.2.3 (unfamiliar or confusing UX patterns).

[claude]

The ambiguity error doesn't point at the remedy. This is now the terminal failure for an ambiguous table, and it fails the whole input on every Connect attempt, but the message never mentions that capture_instance exists to resolve it — the operator has to find the field in the docs. Suggest appending something like "set capture_instance to one of these to choose" (the sibling error on the branch above already names the field).

CONTRIBUTING.md §1.2.4 — "Strongly lints and validates user-provided configuration, clearly telling users of any problems" — and §3.2.2 (poor error handling / difficult-to-diagnose bugs).

[claude]

All coverage for this new resolution logic is integration-gated. resolveCaptureInstance is a pure function over []captureInstance — no DB required — yet every one of its six branches (zero instances, single instance, convention-name preference + warning, override hit, override miss, unresolvable ambiguity) is only exercised by tests behind integration.CheckSkip(t) in stream_integration_test.go, which need Docker and a SQL Server container and are skipped in task test:unit.

A table-driven unit test in package replication (alongside stream_message_test.go) with an errContains field would cover the whole decision table cheaply and run by default, keeping the integration tests for the parts that genuinely need the server.

Project test patterns (.claude/agents/tester.md — table-driven tests with errContains) and CONTRIBUTING.md §1.3.2.

[claude]

Documented failure scope doesn't match the implementation. Both this field description and the operational note added at line 83 say "table discovery still fails for it" / "table discovery fails for that table", which reads as "that one table is skipped, the rest keep streaming". In practice VerifyUserDefinedTables returns on the first unresolvable table (stream.go:541-543), so Connect fails and the entire input never starts — a single ambiguous table takes down every other configured table. The new test is even named ..._TwoNonDefaultCaptureInstancesFailsToStart and its comment says "the whole input fails to start".

Please reword both strings to state that the input fails to start, since these are generated into the published docs.

CONTRIBUTING.md §1.2.3 (known limitations and edge cases are documented) and §1.1.1.

[claude]

publisherStub's mutex is being locked from outside the struct. publisherStub already encapsulates its own locking (Publish, count() in snapshot_test.go); this reaches into publisher.mu from the test body instead. Add a small accessor on the stub (e.g. a first()/all() method that locks and returns a copy) and use that here.

Project Go patterns, .claude/agents/godev.md — "Mutex Encapsulation: Never access a struct's mutex from outside the struct. Mutex operations must only happen inside the struct's own methods."

[claude]

Undocumented limitation: capture_instance is connector-wide, not per-table (CONTRIBUTING.md §1.2.3 — "Known limitations and edge cases are documented")

The value is parsed once and threaded through as a single string to VerifyUserDefinedTables, which passes the same capInstanceOverride to resolveCaptureInstance for every table matched by include/exclude:

• connect/internal/impl/mssqlserver/input_mssqlserver_cdc.go

  Lines 358 to 361 in 66d4e33

  	if userTables, err = replication.VerifyUserDefinedTables(ctx, i.db, i.cfg.tablesFilter, i.cfg.captureInstanceOverride, i.log); err != nil {
  	return fmt.Errorf("verifying user defined tables: %w", err)
  	}

• connect/internal/impl/mssqlserver/replication/stream.go

  Lines 540 to 550 in 66d4e33

  	userTables := make([]UserDefinedTable, 0, len(order))
  	for _, fullName := range order {
  	tbl := tables[fullName]
  	if err := resolveCaptureInstance(&tbl, instances[fullName], capInstanceOverride, log); err != nil {
  	return nil, err
  	}
  	if len(tbl.startLSN) == 0 {
  	return nil, fmt.Errorf("field 'start_lsn' in change table '%s' expected to be set but was not", tbl.ToChangeTable())
  	}
  	userTables = append(userTables, tbl)
  	}

include is a list of regexes, so a realistic online-schema-change rollout has several tables in flight at once, each with its own temporary capture instance. Only one of them can be named here; the rest fall back to the convention-named instance with a warning, or fail table discovery outright if neither instance is convention-named. Nothing in this field description — nor in the "Operational notes" bullet added to the docs — tells the operator the setting is global, so this reads as a per-table selector.

Suggested fix: state explicitly in the field description that the override applies to every matched table and that only one ambiguous table can be disambiguated at a time (or accept a map/list of table -> capture_instance so concurrent migrations are expressible), and regenerate the docs with task docs.

[claude]

Test gap: startLSN selection is never asserted

The table fixtures already give each capture instance a distinct LSN ({0x01} vs {0x02}), but the only assertion here is on tbl.CaptureInstance — tbl.startLSN is never checked.

resolveCaptureInstance sets the name and the LSN as two separate statements in three different branches:

connect/internal/impl/mssqlserver/replication/stream.go

Lines 559 to 601 in 66d4e33

	func resolveCaptureInstance(tbl *UserDefinedTable, instances []captureInstance, override string, log *service.Logger) error {
	switch len(instances) {
	case 0:
	return fmt.Errorf("no change table found for table '%s': is CDC enabled for this table?", tbl.FullName())
	case 1:
	tbl.CaptureInstance = instances[0].name
	tbl.startLSN = instances[0].startLSN
	return nil
	}
	names := make([]string, len(instances))
	for i, inst := range instances {
	names[i] = inst.name
	}
	if override != "" {
	for _, inst := range instances {
	if inst.name == override {
	tbl.CaptureInstance = inst.name
	tbl.startLSN = inst.startLSN
	return nil
	}
	}
	}
	conventionName := fmt.Sprintf("%s_%s", tbl.Schema, tbl.Name)
	for _, inst := range instances {
	if inst.name != conventionName {
	continue
	}
	if override != "" {
	log.Warnf("Table '%s' has multiple CDC capture instances (%s); configured capture_instance '%s' does not match either, falling back to the default-named instance '%s'. "+
	"If this is a mid-migration cutover, check capture_instance matches the new instance's actual name.",
	tbl.FullName(), strings.Join(names, ", "), override, conventionName)
	} else {
	log.Warnf("Table '%s' has multiple CDC capture instances (%s); preferring the default-named instance '%s'. "+
	"If this is a mid-migration cutover, set capture_instance to the new instance's name, or drop the old one once the migration completes.",
	tbl.FullName(), strings.Join(names, ", "), conventionName)
	}
	tbl.CaptureInstance = inst.name
	tbl.startLSN = inst.startLSN
	return nil
	}

so a crossed assignment (e.g. instances[1].name paired with instances[0].startLSN) passes every case in this table. That failure would not surface downstream either: VerifyUserDefinedTables only checks the LSN is non-empty, and ReadChangeTables seeds the iterator from it, so the wrong start LSN silently skips or replays change rows on a cutover instead of erroring.

connect/internal/impl/mssqlserver/replication/stream.go

Lines 545 to 549 in 66d4e33

	}
	if len(tbl.startLSN) == 0 {
	return nil, fmt.Errorf("field 'start_lsn' in change table '%s' expected to be set but was not", tbl.ToChangeTable())
	}
	userTables = append(userTables, tbl)

Suggested fix: add a wantStartLSN column to the test table and assert it alongside wantInstance.