dispatcher: rebuild closed managers during maintainer failover (#6674)

downstreamadapter/dispatchermanager/dispatcher_manager.go

@@ -1054,6 +1054,11 @@ func (e *DispatcherManager) mergeEventDispatcher(dispatcherIDs []common.Dispatch
 
 // ==== remove and clean related functions ====
 
+// IsClosingOrClosed reports whether shutdown has started or completed.
+func (e *DispatcherManager) IsClosingOrClosed() bool {
+	return e.closing.Load() || e.closed.Load()
+}
+
 func (e *DispatcherManager) TryClose(removeChangefeed bool) bool {
 	if removeChangefeed {
 		e.removeChangefeedRequested.Store(true)

downstreamadapter/dispatchermanager/dispatcher_manager_test.go

@@ -518,6 +518,7 @@ func TestTryCloseRemovedRequestAfterClosedReturnsImmediatelyAndTriggersCleanup(t
 
 func TestLocalFenceCancelsWritePathWithoutWaitingForCleanup(t *testing.T) {
 	manager := createTestManager(t)
+	require.False(t, manager.IsClosingOrClosed())
 	ctx, cancel := context.WithCancel(context.Background())
 	manager.ctx = ctx
 	manager.cancel = cancel
@@ -535,11 +536,14 @@ func TestLocalFenceCancelsWritePathWithoutWaitingForCleanup(t *testing.T) {
 		require.FailNow(t, "local fence should not wait for dispatcher manager cleanup")
 	}
 	require.ErrorIs(t, ctx.Err(), context.Canceled)
+	require.True(t, manager.IsClosingOrClosed())
+	require.False(t, manager.TryClose(false))
 
 	manager.wg.Done()
 	require.Eventually(t, func() bool {
 		return manager.TryClose(false)
 	}, time.Second, 10*time.Millisecond)
+	require.True(t, manager.IsClosingOrClosed())
 }
 
 func TestLocalFenceDoesNotWaitForBootstrapWriteBlockEvent(t *testing.T) {

downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go

@@ -197,6 +197,23 @@ func (m *DispatcherOrchestrator) handleBootstrapRequest(
 	closedEpoch, closed := m.closedMaintainerEpochs[cfId]
 	m.mutex.Unlock()
 
+	if exists && manager.IsClosingOrClosed() {
+		manager.MaintainerFenceMu.Lock()
+		if !manager.CanUpdateMaintainer(from, maintainerEpoch) || !manager.TryClose(false) {
+			manager.MaintainerFenceMu.Unlock()
+			// Leave bootstrap pending until the old manager has released its resources.
+			return nil
+		}
+		m.mutex.Lock()
+		delete(m.dispatcherManagers, cfId)
+		m.recordClosedMaintainerEpochLocked(cfId, manager.GetMaintainerEpoch(), false)
+		closedEpoch, closed = m.closedMaintainerEpochs[cfId]
+		m.mutex.Unlock()
+		manager.MaintainerFenceMu.Unlock()
+		metrics.DispatcherManagerGauge.WithLabelValues(cfId.Keyspace(), cfId.Name()).Dec()
+		exists = false
+	}
+
 	var err error
 	if !exists {
 		if closed && (maintainerEpoch == 0 || maintainerEpoch <= closedEpoch) {
@@ -668,10 +685,13 @@ func (m *DispatcherOrchestrator) handleCloseRequest(
 		decGauge := false
 		manager.MaintainerFenceMu.Lock()
 		if manager.IsMaintainerRequestAllowed(from, req.MaintainerEpoch) {
+			// Preserve removal tombstones if bootstrap later retires this manager.
+			m.mutex.Lock()
+			m.recordClosedMaintainerEpochLocked(cfId, req.MaintainerEpoch, req.Removed)
+			m.mutex.Unlock()
 			if closed := manager.TryClose(req.Removed); closed {
 				m.mutex.Lock()
 				delete(m.dispatcherManagers, cfId)
-				m.recordClosedMaintainerEpochLocked(cfId, req.MaintainerEpoch, req.Removed)
 				m.mutex.Unlock()
 				decGauge = true
 				response.Success = true

downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go

@@ -844,6 +844,146 @@ func TestHandleCloseRequestAcksStaleMaintainerEpoch(t *testing.T) {
 	require.NotContains(t, orchestrator.closedMaintainerEpochs, cfID)
 }
 
+func TestHandleBootstrapRequestCanCreateDispatcherAfterHandover(t *testing.T) {
+	for _, tc := range []struct {
+		state    string
+		oldEpoch uint64
+		newEpoch uint64
+	}{
+		{"running", 2, 3},
+		{"closing", 2, 3},
+		{"closed", 2, 3},
+		{"closed", 0, 0},
+	} {
+		t.Run(fmt.Sprintf("%s epoch %d to %d", tc.state, tc.oldEpoch, tc.newEpoch), func(t *testing.T) {
+			mc := messaging.NewMockMessageCenter()
+			setupBootstrapTestServices(t, mc, true)
+
+			cfID := common.NewChangeFeedIDWithName("cf", "default")
+			configBytes, err := json.Marshal(newBootstrapResponseTestChangefeedConfig(cfID))
+			require.NoError(t, err)
+			orchestrator := newTestDispatcherOrchestrator()
+			orchestrator.mc = mc
+			req := &heartbeatpb.MaintainerBootstrapRequest{
+				ChangefeedID:    cfID.ToPB(),
+				Config:          configBytes,
+				StartTs:         100,
+				MaintainerEpoch: tc.oldEpoch,
+			}
+			require.NoError(t, orchestrator.handleBootstrapRequest(node.ID("old-maintainer"), req))
+			requireMessage(t, mc.GetMessageChannel(), "initial bootstrap response was not sent")
+			manager := orchestrator.dispatcherManagers[cfID]
+			require.NotNil(t, manager)
+			cleanupDispatcherManager(t, manager)
+			if tc.state != "running" {
+				require.NoError(t, orchestrator.handleCloseRequest(node.ID("old-maintainer"), &heartbeatpb.MaintainerCloseRequest{
+					ChangefeedID:    cfID.ToPB(),
+					MaintainerEpoch: tc.oldEpoch,
+				}))
+				msg := requireMessage(t, mc.GetMessageChannel(), "close response was not sent")
+				require.False(t, msg.Message[0].(*heartbeatpb.MaintainerCloseResponse).Success)
+
+				// The old maintainer dies before retrying close. Local cleanup finishes,
+				// but no further close request removes the manager from the orchestrator.
+				if tc.state == "closed" {
+					require.Eventually(t, func() bool {
+						return manager.TryClose(false)
+					}, time.Second, 10*time.Millisecond)
+				}
+				require.Same(t, manager, orchestrator.dispatcherManagers[cfID])
+			}
+
+			req.MaintainerEpoch = tc.newEpoch
+			var response *heartbeatpb.MaintainerBootstrapResponse
+			bootstrap := func() bool {
+				require.NoError(t, orchestrator.handleBootstrapRequest(node.ID("new-maintainer"), req))
+				select {
+				case msg := <-mc.GetMessageChannel():
+					response = msg.Message[0].(*heartbeatpb.MaintainerBootstrapResponse)
+					return true
+				default:
+					require.Same(t, manager, orchestrator.dispatcherManagers[cfID])
+					require.Equal(t, node.ID("old-maintainer"), manager.GetMaintainerID())
+					return false
+				}
+			}
+			if !bootstrap() {
+				require.Eventually(t, bootstrap, time.Second, 10*time.Millisecond)
+			}
+			recovered := orchestrator.dispatcherManagers[cfID]
+			require.NotNil(t, recovered)
+			if recovered != manager {
+				cleanupDispatcherManager(t, recovered)
+			}
+			if tc.state == "running" {
+				require.Same(t, manager, recovered)
+			} else {
+				require.NotSame(t, manager, recovered)
+			}
+			require.Nil(t, response.Err)
+			require.Equal(t, tc.newEpoch, response.MaintainerEpoch)
+			require.Equal(t, node.ID("new-maintainer"), recovered.GetMaintainerID())
+
+			// A successful handover must leave a manager capable of creating dispatchers.
+			err = recovered.NewTableTriggerEventDispatcher(common.NewDispatcherID().ToPB(), 100, false)
+			require.NoError(t, err, "bootstrap succeeded but the dispatcher manager write path is still closed")
+		})
+	}
+}
+
+func TestHandleBootstrapRequestRejectsStaleEpochAfterManagerClose(t *testing.T) {
+	for _, tc := range []struct {
+		oldEpoch     uint64
+		requestEpoch uint64
+		removed      bool
+	}{
+		{2, 0, false},
+		{2, 1, false},
+		{2, 2, false},
+		{0, 0, true},
+	} {
+		t.Run(fmt.Sprintf("epoch %d to %d removed %t", tc.oldEpoch, tc.requestEpoch, tc.removed), func(t *testing.T) {
+			mc := messaging.NewMockMessageCenter()
+			setupBootstrapTestServices(t, mc, false)
+			cfID := common.NewChangeFeedIDWithName("cf", "default")
+			configBytes, err := json.Marshal(newBootstrapResponseTestChangefeedConfig(cfID))
+			require.NoError(t, err)
+			orchestrator := newTestDispatcherOrchestrator()
+			orchestrator.mc = mc
+			req := &heartbeatpb.MaintainerBootstrapRequest{
+				ChangefeedID:    cfID.ToPB(),
+				Config:          configBytes,
+				StartTs:         100,
+				MaintainerEpoch: tc.oldEpoch,
+			}
+			require.NoError(t, orchestrator.handleBootstrapRequest(node.ID("old-maintainer"), req))
+			requireMessage(t, mc.GetMessageChannel(), "initial bootstrap response was not sent")
+			manager := orchestrator.dispatcherManagers[cfID]
+			cleanupDispatcherManager(t, manager)
+			require.NoError(t, orchestrator.handleCloseRequest(node.ID("old-maintainer"), &heartbeatpb.MaintainerCloseRequest{
+				ChangefeedID:    cfID.ToPB(),
+				MaintainerEpoch: tc.oldEpoch,
+				Removed:         tc.removed,
+			}))
+			msg := requireMessage(t, mc.GetMessageChannel(), "close response was not sent")
+			require.False(t, msg.Message[0].(*heartbeatpb.MaintainerCloseResponse).Success)
+			require.Contains(t, orchestrator.closedMaintainerEpochs, cfID)
+			require.Equal(t, tc.oldEpoch, orchestrator.closedMaintainerEpochs[cfID])
+			require.Eventually(t, func() bool {
+				return manager.TryClose(false)
+			}, time.Second, 10*time.Millisecond)
+
+			req.MaintainerEpoch = tc.requestEpoch
+			require.NoError(t, orchestrator.handleBootstrapRequest(node.ID("old-maintainer"), req))
+			require.Empty(t, mc.GetMessageChannel(), "a closed maintainer epoch must not bootstrap again")
+			if remaining, exists := orchestrator.dispatcherManagers[cfID]; exists {
+				require.Same(t, manager, remaining)
+				require.True(t, remaining.IsClosingOrClosed())
+			}
+		})
+	}
+}
+
 func TestHandlePostBootstrapWritePathClosedReleasesMaintainerFence(t *testing.T) {
 	mc := messaging.NewMockMessageCenter()
 	setupBootstrapTestServices(t, mc, true)