fix(shard-distributor): separate watch event processing from the rebalancing loop (#7669)

<!-- 1-2 line summary of WHAT changed technically:
- Always link the relevant projects GitHub issue, unless it is a minor
bugfix
- Good: "Modified FailoverDomain mapper to allow null ActiveClusterName
#320"
- Bad: "added nil check" -->
**What changed?**
* Add `runRebalanceTriggeringLoop` method for `namespaceProcessor` 


<!-- Your goal is to provide all the required context for a future
maintainer
to understand the reasons for making this change (see
https://cbea.ms/git-commit/#why-not-how).
How did this work previously (and what was wrong with it)? What has
changed, and why did you solve it
this way?
- Good: "Active-active domains have independent cluster attributes per
region. Previously,
modifying cluster attributes required spedifying the default
ActiveClusterName which
updates the global domain default. This prevents operators from updating
regional
configurations without affecting the primary cluster designation. This
change allows
  attribute updates to be independent of active cluster selection."
- Bad: "Improves domain handling" -->
**Why?**
We observed a server instabily In case of a high number of events and a
long shard rebalancing. It caused a growing number of events not
processed on the server side and etcd's side.


<!-- Include specific test commands and setup. Please include the exact
commands such that
another maintainer or contributor can reproduce the test steps taken. 
- e.g Unit test commands with exact invocation
`go test -v ./common/types/mapper/proto -run TestFailoverDomainRequest`
- For integration tests include setup steps and test commands
Example: "Started local server with `./cadence start`, then ran `make
test_e2e`"
- For local simulation testing include setup steps for the server and
how you ran the tests
- Good: Full commands that reviewers can copy-paste to verify
- Bad: "Tested locally" or "Added tests" -->
**How did you test it?**
* Unit test
* Run on dev cluster


<!-- If there are risks that the release engineer should know about
document them here.
For example:
- Has an API/IDL been modified? Is it backwards/forwards compatible? If
not, what are the repecussions?
- Has a schema change been introduced? Is it possible to roll back?
- Has a feature flag been re-used for a new purpose? 
- Is there a potential performance concern? Is the change modifying core
task processing logic?
- If truly N/A, you can mark it as such -->
**Potential risks**
N/A


<!-- If this PR completes a user facing feature or changes functionality
add release notes here.
Your release notes should allow a user and the release engineer to
understand the changes with little context.
Always ensure that the description contains a link to the relevant
GitHub issue. -->
**Release notes**
N/A


<!-- Consider whether this change requires documentation updates in the
Cadence-Docs repo
- If yes: mention what needs updating (or link to docs PR in
cadence-docs repo)
- If in doubt, add a note about potential doc needs
- Only mark N/A if you're certain no docs are affected -->
**Documentation Changes**
N/A


---

## Reviewer Validation

**PR Description Quality** (check these before reviewing code):

- [ ] **"What changed"** provides a clear 1-2 line summary
  - [ ] Project Issue is linked
- [ ] **"Why"** explains the full motivation with sufficient context
- [ ] **Testing is documented:**
- [ ] Unit test commands are included (with exact `go test` invocation)
- [ ] Integration test setup/commands included (if integration tests
were run)
  - [ ] Canary testing details included (if canary was mentioned)
- [ ] **Potential risks** section is thoughtfully filled out (or
legitimately N/A)
- [ ] **Release notes** included if this completes a user-facing feature
- [ ] **Documentation** needs are addressed (or noted if uncertain)

Author: arzonus

service/sharddistributor/leader/process/processor.go

@@ -180,18 +180,16 @@ func (p *namespaceProcessor) runProcess(ctx context.Context) {
 
 // runRebalancingLoop handles shard assignment and redistribution.
 func (p *namespaceProcessor) runRebalancingLoop(ctx context.Context) {
-	ticker := p.timeSource.NewTicker(p.cfg.Period)
-	defer ticker.Stop()
 
 	// Perform an initial rebalance on startup.
 	err := p.rebalanceShards(ctx)
 	if err != nil {
 		p.logger.Error("initial rebalance failed", tag.Error(err))
 	}
 
-	updateChan, err := p.shardStore.Subscribe(ctx, p.namespaceCfg.Name)
+	updateChan, err := p.runRebalanceTriggeringLoop(ctx)
 	if err != nil {
-		p.logger.Error("Failed to subscribe to state changes, stopping rebalancing loop.", tag.Error(err))
+		p.logger.Error("failed to start rebalance triggering loop", tag.Error(err))
 		return
 	}
 
@@ -200,22 +198,65 @@ func (p *namespaceProcessor) runRebalancingLoop(ctx context.Context) {
 		case <-ctx.Done():
 			p.logger.Info("Rebalancing loop cancelled.")
 			return
+
+		case update := <-updateChan:
+			p.logger.Info("Rebalancing triggered", tag.Dynamic("reason", update))
+			if err := p.rebalanceShards(ctx); err != nil {
+				p.logger.Error("rebalance failed", tag.Error(err))
+			}
+		}
+	}
+}
+
+// runRebalanceTriggeringLoop monitors for state changes and periodic triggers to initiate rebalancing.
+// it doesn't block Subscribe calls to avoid a growing backlog of updates.
+func (p *namespaceProcessor) runRebalanceTriggeringLoop(ctx context.Context) (<-chan string, error) {
+	// Buffered channel to allow one pending rebalance trigger.
+	triggerChan := make(chan string, 1)
+
+	updateChan, err := p.shardStore.Subscribe(ctx, p.namespaceCfg.Name)
+	if err != nil {
+		p.logger.Error("Failed to subscribe to state changes, stopping rebalancing loop.", tag.Error(err))
+		return nil, err
+	}
+
+	go p.rebalanceTriggeringLoop(ctx, updateChan, triggerChan)
+	return triggerChan, nil
+}
+
+func (p *namespaceProcessor) rebalanceTriggeringLoop(ctx context.Context, updateChan <-chan int64, triggerChan chan<- string) {
+	defer close(triggerChan)
+
+	ticker := p.timeSource.NewTicker(p.cfg.Period)
+	defer ticker.Stop()
+
+	tryTriggerRebalancing := func(reason string) {
+		select {
+		case triggerChan <- reason:
+		default:
+			p.logger.Info("Rebalance already pending, skipping trigger attempt", tag.Dynamic("reason", reason))
+		}
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			p.logger.Info("Rebalance triggering loop cancelled")
+			return
+
+		case <-ticker.Chan():
+			tryTriggerRebalancing("Periodic reconciliation triggered")
+
 		case latestRevision, ok := <-updateChan:
 			if !ok {
-				p.logger.Info("Update channel closed, stopping rebalancing loop.")
+				p.logger.Info("Update channel closed, stopping rebalance triggering loop")
 				return
 			}
 			if latestRevision <= p.lastAppliedRevision {
 				continue
 			}
-			p.logger.Info("State change detected, triggering rebalance.")
-			err = p.rebalanceShards(ctx)
-		case <-ticker.Chan():
-			p.logger.Info("Periodic reconciliation triggered, rebalancing.")
-			err = p.rebalanceShards(ctx)
-		}
-		if err != nil {
-			p.logger.Error("rebalance failed", tag.Error(err))
+
+			tryTriggerRebalancing("State change detected")
 		}
 	}
 }

service/sharddistributor/leader/process/processor_test.go

@@ -1414,3 +1414,177 @@ func TestEmitOldestExecutorHeartbeatLag(t *testing.T) {
 		})
 	}
 }
+
+func TestRunRebalanceTriggeringLoop(t *testing.T) {
+	t.Run("no events from subscribe, trigger from ticker", func(t *testing.T) {
+		mocks := setupProcessorTest(t, config.NamespaceTypeFixed)
+		defer mocks.ctrl.Finish()
+		processor := mocks.factory.CreateProcessor(mocks.cfg, mocks.store, mocks.election).(*namespaceProcessor)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		updateChan := make(chan int64)
+		triggerChan := make(chan string, 1)
+
+		go processor.rebalanceTriggeringLoop(ctx, updateChan, triggerChan)
+
+		// Wait for ticker to be created
+		mocks.timeSource.BlockUntil(1)
+
+		// Advance time to trigger the ticker
+		mocks.timeSource.Advance(processor.cfg.Period)
+
+		// Expect trigger from periodic reconciliation
+		select {
+		case reason := <-triggerChan:
+			assert.Equal(t, "Periodic reconciliation triggered", reason)
+		case <-time.After(time.Second):
+			t.Fatal("expected trigger from ticker, but timed out")
+		}
+
+		cancel()
+	})
+
+	t.Run("events from subscribe before period, trigger from state change", func(t *testing.T) {
+		mocks := setupProcessorTest(t, config.NamespaceTypeFixed)
+		defer mocks.ctrl.Finish()
+		processor := mocks.factory.CreateProcessor(mocks.cfg, mocks.store, mocks.election).(*namespaceProcessor)
+		processor.lastAppliedRevision = 0
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		updateChan := make(chan int64, 1)
+		triggerChan := make(chan string, 1)
+
+		go processor.rebalanceTriggeringLoop(ctx, updateChan, triggerChan)
+
+		// Wait for ticker to be created
+		mocks.timeSource.BlockUntil(1)
+
+		// Send a state change event before the ticker fires
+		updateChan <- 1
+
+		// Expect trigger from state change
+		select {
+		case reason := <-triggerChan:
+			assert.Equal(t, "State change detected", reason)
+		case <-time.After(time.Second):
+			t.Fatal("expected trigger from state change, but timed out")
+		}
+
+		cancel()
+	})
+
+	t.Run("triggerChan full, multiple subscribe events, loop not stuck", func(t *testing.T) {
+		mocks := setupProcessorTest(t, config.NamespaceTypeFixed)
+		defer mocks.ctrl.Finish()
+		processor := mocks.factory.CreateProcessor(mocks.cfg, mocks.store, mocks.election).(*namespaceProcessor)
+		processor.lastAppliedRevision = 0
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		// Use unbuffered channel for updates to ensure they are processed one at a time
+		updateChan := make(chan int64)
+		triggerChan := make(chan string, 1)
+
+		go processor.rebalanceTriggeringLoop(ctx, updateChan, triggerChan)
+
+		// Wait for ticker to be created
+		mocks.timeSource.BlockUntil(1)
+
+		// Don't read from triggerChan yet to keep it full
+		// Send multiple state change events
+		for i := int64(0); i <= 10; i++ {
+			select {
+			case updateChan <- i:
+			case <-time.After(time.Second):
+				// Expect that the loop is not stuck
+				t.Fatalf("failed to send update %d, channel blocked", i)
+			}
+		}
+
+		// Expect trigger from state change
+		select {
+		case reason := <-triggerChan:
+			assert.Equal(t, "State change detected", reason)
+		case <-time.After(time.Second):
+			t.Fatal("expected trigger from state change, but timed out")
+		}
+
+		cancel()
+	})
+
+	t.Run("stale revision ignored", func(t *testing.T) {
+		mocks := setupProcessorTest(t, config.NamespaceTypeFixed)
+		defer mocks.ctrl.Finish()
+		processor := mocks.factory.CreateProcessor(mocks.cfg, mocks.store, mocks.election).(*namespaceProcessor)
+		processor.lastAppliedRevision = 5
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		updateChan := make(chan int64, 1)
+		triggerChan := make(chan string, 1)
+
+		go processor.rebalanceTriggeringLoop(ctx, updateChan, triggerChan)
+
+		// Wait for ticker to be created
+		mocks.timeSource.BlockUntil(1)
+
+		// Send stale revision (less than or equal to lastAppliedRevision)
+		updateChan <- 3
+
+		// Should not trigger - verify by advancing ticker and getting that trigger instead
+		mocks.timeSource.Advance(processor.cfg.Period)
+
+		select {
+		case reason := <-triggerChan:
+			assert.Equal(t, "Periodic reconciliation triggered", reason)
+		case <-time.After(time.Second):
+			t.Fatal("expected trigger from ticker")
+		}
+
+		cancel()
+	})
+
+	t.Run("update channel closed stops loop", func(t *testing.T) {
+		mocks := setupProcessorTest(t, config.NamespaceTypeFixed)
+		defer mocks.ctrl.Finish()
+		processor := mocks.factory.CreateProcessor(mocks.cfg, mocks.store, mocks.election).(*namespaceProcessor)
+
+		ctx := context.Background()
+
+		updateChan := make(chan int64)
+		triggerChan := make(chan string, 1)
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			processor.rebalanceTriggeringLoop(ctx, updateChan, triggerChan)
+		}()
+
+		// Wait for ticker to be created
+		mocks.timeSource.BlockUntil(1)
+
+		// Close update channel
+		close(updateChan)
+
+		// Wait for loop to exit
+		done := make(chan struct{})
+		go func() {
+			wg.Wait()
+			close(done)
+		}()
+
+		select {
+		case <-done:
+			// Loop exited as expected
+		case <-time.After(time.Second):
+			t.Fatal("loop did not exit after updateChan closed")
+		}
+	})
+}