feat(oplog-connector): split into message/collections deployments; rename subjects to chat.migration.oplog.*

data-migration/CDC_COVERAGE.md

@@ -10,6 +10,11 @@
 
 The connector forwards raw change-stream events with **no `updateLookup`** and **no `fullDocumentBeforeChange`**:
 
+> **Deployment note:** the connector runs as two deployments — `oplog-connector-messages`
+> (only `rocketchat_message`) and `oplog-connector-collections` (all other watched
+> collections) — with disjoint `WATCH_COLLECTIONS`, so a collection-side fault cannot stall
+> message CDC. Coverage below is unchanged by the split.
+
 | Op | Payload carried | Source lookup by `_id` |
 |---|---|---|
 | `insert` | full `fullDocument` | in payload |

data-migration/README.md

@@ -95,7 +95,7 @@ and the matching `docs/superpowers/plans/2026-06-11-…` + `…2026-06-15-oplog-
     delete-cap reseed window). It's **scoped to the one message collection** (other watchers advance
     normally), and is a **connector-internal** concern; advancing from the post-batch resume token (PBRT)
     is a future connector change, not part of the federation-filter work.
-- **Subjects:** `chat.oplog.{siteID}.{rawCollection}.{op}`; dedup via
+- **Subjects:** `chat.migration.oplog.{siteID}.{rawCollection}.{op}`; dedup via
   `Nats-Msg-Id` = change-stream `_id._data`.
 - **Checkpoints:** one doc per collection in `oplog_checkpoints` (in `CHECKPOINT_DB`
   on the source RS). The resume token is the real checkpoint; saved only after a

data-migration/oplog-connector/bootstrap_test.go

@@ -51,7 +51,7 @@ func TestBootstrapStreams_EnabledCreatesSchemaOnly(t *testing.T) {
 	require.Len(t, fsm.created, 1)
 	got := fsm.created[0]
 	assert.Equal(t, "MIGRATION_OPLOG_site1", got.Name)
-	assert.Equal(t, []string{"chat.oplog.site1.>"}, got.Subjects)
+	assert.Equal(t, []string{"chat.migration.oplog.site1.>"}, got.Subjects)
 	// Federation config is ops/IaC-owned — never set here.
 	assert.Nil(t, got.Sources)
 	assert.Empty(t, fsm.streamHit, "enabled path does not verify")

data-migration/oplog-connector/config.go

@@ -24,8 +24,7 @@ type config struct {
 
 	WatchCollections []string `env:"WATCH_COLLECTIONS,required"`
 
-	// MessageCollection is the one watched collection the federation-origin $match is scoped to.
-	// Other collections (rooms/users) are forwarded unfiltered so we don't silently drop foreign ones.
+	// MessageCollection scopes the federation-origin $match; collections-role deployments legitimately don't watch it.
 	MessageCollection string `env:"MESSAGE_COLLECTION" envDefault:"rocketchat_message"`
 
 	ReadPreference string `env:"READ_PREFERENCE" envDefault:"secondary"`
@@ -83,19 +82,20 @@ func parseConfig() (config, error) {
 	if dup := firstDuplicate(cfg.WatchCollections); dup != "" {
 		return config{}, fmt.Errorf("WATCH_COLLECTIONS has duplicate entry %q (each collection maps to one watcher and one checkpoint)", dup)
 	}
-	// Fail-open guard: if MESSAGE_COLLECTION is empty or isn't actually watched, the
-	// federation-origin $match never runs and ALL foreign messages migrate silently
-	// (double-deliver). Fail fast instead.
+	// MESSAGE_COLLECTION must always be defined but need NOT be watched — "watched ⟹ filtered"
+	// holds by construction: the $match applies to whichever watcher's name equals it.
 	cfg.MessageCollection = strings.TrimSpace(cfg.MessageCollection)
 	if cfg.MessageCollection == "" {
 		return config{}, fmt.Errorf("MESSAGE_COLLECTION must be non-empty (the federation-origin $match would never run)")
 	}
-	if !slices.Contains(cfg.WatchCollections, cfg.MessageCollection) {
-		return config{}, fmt.Errorf("MESSAGE_COLLECTION %q is not present in WATCH_COLLECTIONS — the federation-origin $match would never run", cfg.MessageCollection)
-	}
 	return cfg, nil
 }
 
+// watchesMessages reports whether this deployment watches the federated message collection (message role).
+func (c *config) watchesMessages() bool {
+	return slices.Contains(c.WatchCollections, c.MessageCollection)
+}
+
 // firstDuplicate returns the first repeated (trimmed) entry, or "" if all unique.
 func firstDuplicate(items []string) string {
 	seen := make(map[string]bool, len(items))

data-migration/oplog-connector/config_test.go

@@ -58,15 +58,21 @@ func TestParseConfig_TrimsCollectionWhitespace(t *testing.T) {
 	assert.Equal(t, []string{"rocketchat_message", "users"}, cfg.WatchCollections)
 }
 
-func TestParseConfig_MessageCollectionNotWatchedFails(t *testing.T) {
+func TestParseConfig_MessageCollectionNotWatchedPasses(t *testing.T) {
 	setRequiredEnv(t)
-	// MESSAGE_COLLECTION is not present in WATCH_COLLECTIONS — the federation
-	// $match would never run, silently migrating all foreign messages.
+	// Collections role: the message collection is tailed by a separate deployment, so its absence is legitimate.
 	t.Setenv("WATCH_COLLECTIONS", "rocketchat_room,users")
 	t.Setenv("MESSAGE_COLLECTION", "rocketchat_message")
-	_, err := parseConfig()
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "MESSAGE_COLLECTION")
+	cfg, err := parseConfig()
+	require.NoError(t, err)
+	assert.False(t, cfg.watchesMessages())
+}
+
+func TestParseConfig_WatchesMessagesWhenPresent(t *testing.T) {
+	setRequiredEnv(t) // WATCH_COLLECTIONS includes rocketchat_message
+	cfg, err := parseConfig()
+	require.NoError(t, err)
+	assert.True(t, cfg.watchesMessages())
 }
 
 func TestParseConfig_EmptyMessageCollectionFails(t *testing.T) {

data-migration/oplog-connector/deploy/docker-compose.yml

@@ -27,7 +27,11 @@ services:
     networks:
       - oplog-local
 
-  oplog-connector:
+  # Deployment split (spec 2026-07-09): the message pump runs alone so a fault on the
+  # collection side can't stall message CDC. Watch sets MUST stay disjoint, and exactly
+  # one service includes rocketchat_message. No host ports are published, so both keep
+  # the default METRICS_ADDR :9090 inside their own container.
+  oplog-connector-messages:
     build:
       context: ../..
       dockerfile: data-migration/oplog-connector/deploy/Dockerfile
@@ -42,7 +46,29 @@ services:
       - SOURCE_DB=rocketchat
       - CHECKPOINT_DB=migration
       - NATS_URL=nats://nats:4222
-      - WATCH_COLLECTIONS=rocketchat_message,rocketchat_room,rocketchat_subscription,rocketchat_uploads,tsmc_room_members,tsmc_thread_subscriptions,tsmc_hr_acct_org,users,rocketchat_avatar,tsmc_apps_v,tsmc_bot_cmd_men,tsmc_tsso_tokens,tsmc_bot_authorization,ufsTokens,user_devices
+      - WATCH_COLLECTIONS=rocketchat_message
+      - READ_PREFERENCE=primaryPreferred
+      - BOOTSTRAP_STREAMS=true
+      - LOG_LEVEL=info
+    networks:
+      - oplog-local
+
+  oplog-connector-collections:
+    build:
+      context: ../..
+      dockerfile: data-migration/oplog-connector/deploy/Dockerfile
+    depends_on:
+      source-mongo:
+        condition: service_healthy
+      nats:
+        condition: service_started
+    environment:
+      - SITE_ID=site-local
+      - SOURCE_MONGO_URI=mongodb://source-mongo:27017/?replicaSet=rs0
+      - SOURCE_DB=rocketchat
+      - CHECKPOINT_DB=migration
+      - NATS_URL=nats://nats:4222
+      - WATCH_COLLECTIONS=rocketchat_room,rocketchat_subscription,rocketchat_uploads,tsmc_room_members,tsmc_thread_subscriptions,tsmc_hr_acct_org,users,rocketchat_avatar,tsmc_apps_v,tsmc_bot_cmd_men,tsmc_tsso_tokens,tsmc_bot_authorization,ufsTokens,user_devices
       - READ_PREFERENCE=primaryPreferred
       - BOOTSTRAP_STREAMS=true
       - LOG_LEVEL=info

data-migration/oplog-connector/envelope_test.go

@@ -27,10 +27,10 @@ func TestBuildEnvelope_OpsAndSubjects(t *testing.T) {
 		hasUpdateDesc bool // delta (update)
 		wantSubject   string
 	}{
-		{"insert", "insert", true, false, "chat.oplog.site1.rocketchat_message.insert"},
-		{"update", "update", false, true, "chat.oplog.site1.rocketchat_message.update"},
-		{"replace", "replace", true, false, "chat.oplog.site1.rocketchat_message.replace"},
-		{"delete", "delete", false, false, "chat.oplog.site1.rocketchat_message.delete"},
+		{"insert", "insert", true, false, "chat.migration.oplog.site1.rocketchat_message.insert"},
+		{"update", "update", false, true, "chat.migration.oplog.site1.rocketchat_message.update"},
+		{"replace", "replace", true, false, "chat.migration.oplog.site1.rocketchat_message.replace"},
+		{"delete", "delete", false, false, "chat.migration.oplog.site1.rocketchat_message.delete"},
 	}
 
 	for _, tc := range tests {

data-migration/oplog-connector/handler.go

@@ -183,7 +183,9 @@ func (w *watcher) publishWithRetry(ctx context.Context, ev *changeEvent) error {
 	subj, msgID, evt := buildEnvelope(ev, w.siteID, w.now())
 	data, err := json.Marshal(evt)
 	if err != nil {
-		w.log.Error("marshal oplog event failed — skipping event", "eventId", ev.EventID, "error", err)
+		// This event is DROPPED (never reaches the stream), so name the op — eventId alone
+		// can't be looked up via Nats-Msg-Id like published events can.
+		w.log.Error("marshal oplog event failed — skipping event", "eventId", ev.EventID, "op", ev.Op, "error", err)
 		w.metrics.onSkipped(ctx, w.collection)
 		return nil
 	}

data-migration/oplog-connector/handler_test.go

@@ -157,8 +157,8 @@ func TestWatcher_PublishesInOrderAndCheckpointsOnDrain(t *testing.T) {
 	require.NoError(t, w.run(context.Background()))
 
 	require.Len(t, pub.msgs, 3)
-	assert.Equal(t, "chat.oplog.site1.rocketchat_message.insert", pub.msgs[0].Subject)
-	assert.Equal(t, "chat.oplog.site1.rocketchat_message.delete", pub.msgs[2].Subject)
+	assert.Equal(t, "chat.migration.oplog.site1.rocketchat_message.insert", pub.msgs[0].Subject)
+	assert.Equal(t, "chat.migration.oplog.site1.rocketchat_message.delete", pub.msgs[2].Subject)
 	assert.Equal(t, "E1", pub.msgs[0].Header.Get("Nats-Msg-Id"))
 	assert.Equal(t, []string{"E1", "E2", "E3"}, msgIDs(pub.msgs), "publish order = oplog order")
 

data-migration/oplog-connector/integration_test.go

@@ -5,6 +5,7 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -103,7 +104,7 @@ func TestConnector_RealPublishEndToEnd(t *testing.T) {
 	require.Eventually(t, func() bool {
 		cons, cerr := js.CreateOrUpdateConsumer(ctx, "MIGRATION_OPLOG_site1", jetstream.ConsumerConfig{
 			AckPolicy:      jetstream.AckExplicitPolicy,
-			FilterSubjects: []string{"chat.oplog.site1.>"},
+			FilterSubjects: []string{"chat.migration.oplog.site1.>"},
 		})
 		if cerr != nil {
 			return false
@@ -114,7 +115,7 @@ func TestConnector_RealPublishEndToEnd(t *testing.T) {
 		}
 		for m := range batch.Messages() {
 			assert.NoError(t, m.Ack())
-			if m.Subject() == "chat.oplog.site1.rocketchat_message.insert" {
+			if m.Subject() == "chat.migration.oplog.site1.rocketchat_message.insert" {
 				gotID = m.Headers().Get("Nats-Msg-Id")
 			}
 		}
@@ -166,9 +167,9 @@ func TestOplogConnector_ChangeStreamEndToEnd(t *testing.T) {
 	require.GreaterOrEqual(t, len(msgs), 3)
 
 	// Assert the first three ops in oplog order.
-	assert.Equal(t, "chat.oplog.site1.rocketchat_message.insert", msgs[0].Subject)
-	assert.Equal(t, "chat.oplog.site1.rocketchat_message.update", msgs[1].Subject)
-	assert.Equal(t, "chat.oplog.site1.rocketchat_message.delete", msgs[2].Subject)
+	assert.Equal(t, "chat.migration.oplog.site1.rocketchat_message.insert", msgs[0].Subject)
+	assert.Equal(t, "chat.migration.oplog.site1.rocketchat_message.update", msgs[1].Subject)
+	assert.Equal(t, "chat.migration.oplog.site1.rocketchat_message.delete", msgs[2].Subject)
 
 	for _, m := range msgs[:3] {
 		assert.NotEmpty(t, m.Header.Get("Nats-Msg-Id"), "every event carries a dedup id")
@@ -304,3 +305,67 @@ func TestMongoCheckpointStore_Integration(t *testing.T) {
 	assert.Equal(t, "EVT2", reloaded.EventID)
 	assert.Equal(t, "runtime", reloaded.Source)
 }
+
+// TestConnector_CollectionsRole_DisjointSet starts a collections-role connector (message collection
+// configured but NOT watched) and asserts it publishes only its own collections' subjects.
+func TestConnector_CollectionsRole_DisjointSet(t *testing.T) {
+	client, uri := startReplicaSet(t)
+	rooms := createSourceCollection(t, client.Database("rocketchat"), "rocketchat_room")
+	msgs := createSourceCollection(t, client.Database("rocketchat"), "rocketchat_message")
+
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	cfg := config{
+		SiteID:            "sitecr",
+		SourceMongoURI:    uri,
+		SourceDB:          "rocketchat",
+		CheckpointDB:      "migration",
+		NatsURL:           testutil.NATS(t),
+		WatchCollections:  []string{"rocketchat_room"},
+		MessageCollection: "rocketchat_message", // not watched — collections role
+		ReadPreference:    "primaryPreferred",
+		CheckpointEvery:   1,
+		StartMode:         "now",
+		Bootstrap:         bootstrapConfig{Enabled: true},
+	}
+	conn, err := start(ctx, &cfg, nil)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	_, err = rooms.InsertOne(ctx, bson.M{"_id": "r1", "name": "general"})
+	require.NoError(t, err)
+	_, err = msgs.InsertOne(ctx, bson.M{"_id": "m1", "msg": "hi"}) // no watcher — must not be forwarded
+	require.NoError(t, err)
+
+	nc, err := natsutil.Connect(cfg.NatsURL, "")
+	require.NoError(t, err)
+	defer func() { assert.NoError(t, nc.Drain()) }()
+	js, err := oteljetstream.New(nc)
+	require.NoError(t, err)
+
+	var subjects []string
+	require.Eventually(t, func() bool {
+		cons, cerr := js.CreateOrUpdateConsumer(ctx, "MIGRATION_OPLOG_sitecr", jetstream.ConsumerConfig{
+			AckPolicy:      jetstream.AckExplicitPolicy,
+			FilterSubjects: []string{"chat.migration.oplog.sitecr.>"},
+		})
+		if cerr != nil {
+			return false
+		}
+		batch, berr := cons.Fetch(10, jetstream.FetchMaxWait(500*time.Millisecond))
+		if berr != nil {
+			return false
+		}
+		for m := range batch.Messages() {
+			assert.NoError(t, m.Ack())
+			subjects = append(subjects, m.Subject())
+		}
+		return slices.Contains(subjects, "chat.migration.oplog.sitecr.rocketchat_room.insert")
+	}, 40*time.Second, 500*time.Millisecond, "room insert must land on MIGRATION_OPLOG")
+
+	for _, s := range subjects {
+		assert.NotContains(t, s, "rocketchat_message",
+			"collections-role deployment must never publish message subjects")
+	}
+}

data-migration/oplog-connector/main.go

@@ -48,7 +48,12 @@ func main() {
 		slog.Error("init meter failed", "error", err)
 		os.Exit(1)
 	}
-	m, err := newMetrics()
+	// role distinguishes the two split deployments in logs and metrics (PR #482 review).
+	role := "collections"
+	if cfg.watchesMessages() {
+		role = "messages"
+	}
+	m, err := newMetrics(role)
 	if err != nil {
 		slog.Error("init metrics failed", "error", err)
 		os.Exit(1)
@@ -74,8 +79,15 @@ func main() {
 		slog.Error("startup failed", "error", err)
 		os.Exit(1)
 	}
-	slog.Info("oplog-connector started", "site", cfg.SiteID, "collections", cfg.WatchCollections)
-	slog.Info("federation-origin filter active", "message_collection", cfg.MessageCollection)
+	slog.Info("oplog-connector started", "site", cfg.SiteID, "role", role, "collections", cfg.WatchCollections)
+	if cfg.watchesMessages() {
+		slog.Info("federation-origin filter active", "role", role, "message_collection", cfg.MessageCollection)
+	} else {
+		// Warn, not Info: legitimate for a collections-role pod, but conspicuous when a MESSAGE_COLLECTION
+		// typo leaves a message pod forwarding foreign-origin messages unfiltered (double-deliver).
+		slog.Warn("no message collection watched — federation-origin filter inactive",
+			"role", role, "message_collection", cfg.MessageCollection)
+	}
 
 	// A fatal watcher error (e.g. lost resume token) exits non-zero without waiting
 	// for a signal — recovery is operator-driven. Also exits on Done(), so no leak.