fix(shard distributor): remove heartbeat write cooldown (#7513)

<!-- Describe what has changed in this PR -->
**What changed?**
- Removed `_heartbeatRefreshRate` and the block that early returns in
service/sharddistributor/handler/executor.go, so executor heartbeats are
always persisted.
- Also updated TestHeartbeat in
service/sharddistributor/handler/executor_test.go. Replaced the
“within/after refresh rate” cases that were present before the removal
of `_heartbeatRereshRate` with a single test that expects
`RecordHeartbeat` to be called on a second heartbeat with the same
status.

<!-- Tell your future self why have you made these changes -->
**Why?**
- Previously, the handler skipped writing some heartbeats when:
- the migration mode was 'onboarding', a previous heartbeat existed, the
status was unchanged,
- and most importantly the new heartbeat arrived within
`**_heartbeatRefreshRate**` (2s) of the last one.
- This meant the stored LastHeartbeat in etcd could lag behind the real
heartbeat rate. With a heartbeat ttl being the same value (2s), healthy
executors could be misclassified as stale and removed, which matched
what we saw in the canary (executors disappearing from GetState, and
shards "collapsing" onto only a few executors).
 - Removing the write cooldown gives a simpler and safer behavior:
    - executors heartbeat roughly once a second (in development),
    - every heartbeat is persisted,
- the staleness check is based on the real heartbeat frequency, and
can't be interrupted by another setting somewhere else in the code that
may gate it
    
Two other alternatives were considered, instead of removing the check
and cooldown:
- Increasing the heartbeat TTL (e.g., to 5–10s) while keeping the
cooldown.
  - or decreasing `_heartbeatRefreshRate` (e.g., to 1s).

Both of these alternatives would reduce the chance of misclassifying
healthy executors as stale, but they keep a hidden coupling between
heartbeat.TTL and the write cooldown. Removing the cooldown entirely
makes the behavior easier to reason about and avoids this subtle issue
than can happen in configration.


<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**
- Updated and ran unit tests in
service/sharddistributor/handler/executor_test.go (TestHeartbeat).
- ran the sharddistributor canary with multiple executors and observed
stable executor counts and shard distribution.


<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**
- Higher etcd write load for heartbeats: we now persist every heartbeat
instead of at most once per `_heartbeatRefreshRate`.
- Adopters without external rate limiting (again, not sure if this is
relevant) may see more frequent heartbeat writes than before, so they
should be aware that this change removes a local write throttle and rely
on their own rate limiting

<!-- Is it notable for release? e.g. schema updates, configuration or
data migration required? If so, please mention it, and also update
CHANGELOG.md -->
**Release notes**

<!-- Is there any documentation updates should be made for config,
https://cadenceworkflow.io/docs/operation-guide/setup/ ? If so, please
open an PR in https://github.com/cadence-workflow/cadence-docs -->
**Documentation Changes**

Signed-off-by: Andreas Holt <6665487+AndreasHolt@users.noreply.github.com>

Author: AndreasHolt

service/sharddistributor/handler/executor.go

@@ -16,8 +16,6 @@ import (
 )
 
 const (
-	_heartbeatRefreshRate = 2 * time.Second
-
 	_maxMetadataKeys      = 32
 	_maxMetadataKeyLength = 128
 	_maxMetadataValueSize = 512 * 1024 // 512KB
@@ -76,15 +74,6 @@ func (h *executor) Heartbeat(ctx context.Context, request *types.ExecutorHeartbe
 		shardAssignedInBackground = false
 	}
 
-	// If the state has changed we need to update heartbeat data.
-	// Otherwise, we want to do it with controlled frequency - at most every _heartbeatRefreshRate.
-	if previousHeartbeat != nil && request.Status == previousHeartbeat.Status && mode == types.MigrationModeONBOARDED {
-		lastHeartbeatTime := previousHeartbeat.LastHeartbeat
-		if heartbeatTime.Sub(lastHeartbeatTime) < _heartbeatRefreshRate {
-			return _convertResponse(assignedShards, mode), nil
-		}
-	}
-
 	newHeartbeat := store.HeartbeatState{
 		LastHeartbeat:  heartbeatTime,
 		Status:         request.Status,

service/sharddistributor/handler/executor_test.go

@@ -52,8 +52,8 @@ func TestHeartbeat(t *testing.T) {
 		require.NoError(t, err)
 	})
 
-	// Test Case 2: Subsequent Heartbeat within the refresh rate (no update)
-	t.Run("SubsequentHeartbeatWithinRate", func(t *testing.T) {
+	// Test Case 2: Subsequent heartbeat records a new heartbeat
+	t.Run("SubsequentHeartbeatRecords", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockStore := store.NewMockStore(ctrl)
 		mockTimeSource := clock.NewMockedTimeSourceAt(now)
@@ -72,46 +72,17 @@ func TestHeartbeat(t *testing.T) {
 			Status:        types.ExecutorStatusACTIVE,
 		}
 
-		mockStore.EXPECT().GetHeartbeat(gomock.Any(), namespace, executorID).Return(&previousHeartbeat, nil, nil)
-
-		_, err := handler.Heartbeat(ctx, req)
-		require.NoError(t, err)
-	})
-
-	// Test Case 3: Subsequent Heartbeat after refresh rate (with update)
-	t.Run("SubsequentHeartbeatAfterRate", func(t *testing.T) {
-		ctrl := gomock.NewController(t)
-		mockStore := store.NewMockStore(ctrl)
-		mockTimeSource := clock.NewMockedTimeSourceAt(now)
-		shardDistributionCfg := config.ShardDistribution{}
-		migrationConfig := newMigrationConfig(t, []configEntry{})
-		handler := NewExecutorHandler(testlogger.New(t), mockStore, mockTimeSource, shardDistributionCfg, migrationConfig, metrics.NoopClient)
-
-		req := &types.ExecutorHeartbeatRequest{
-			Namespace:  namespace,
-			ExecutorID: executorID,
-			Status:     types.ExecutorStatusACTIVE,
-		}
-
-		previousHeartbeat := store.HeartbeatState{
-			LastHeartbeat: now,
-			Status:        types.ExecutorStatusACTIVE,
-		}
-
-		// Advance time
-		mockTimeSource.Advance(_heartbeatRefreshRate + time.Second)
-
 		mockStore.EXPECT().GetHeartbeat(gomock.Any(), namespace, executorID).Return(&previousHeartbeat, nil, nil)
 		mockStore.EXPECT().RecordHeartbeat(gomock.Any(), namespace, executorID, store.HeartbeatState{
-			LastHeartbeat: mockTimeSource.Now().UTC(),
+			LastHeartbeat: now,
 			Status:        types.ExecutorStatusACTIVE,
 		})
 
 		_, err := handler.Heartbeat(ctx, req)
 		require.NoError(t, err)
 	})
 
-	// Test Case 4: Status Change (with update)
+	// Test Case 3: Status Change (with update)
 	t.Run("StatusChange", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockStore := store.NewMockStore(ctrl)
@@ -141,7 +112,7 @@ func TestHeartbeat(t *testing.T) {
 		require.NoError(t, err)
 	})
 
-	// Test Case 5: Storage Error
+	// Test Case 4: Storage Error
 	t.Run("StorageError", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockStore := store.NewMockStore(ctrl)
@@ -164,7 +135,7 @@ func TestHeartbeat(t *testing.T) {
 		require.Contains(t, err.Error(), expectedErr.Error())
 	})
 
-	// Test Case 6: Heartbeat with executor associated invalid migration mode
+	// Test Case 5: Heartbeat with executor associated invalid migration mode
 	t.Run("MigrationModeInvald", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockStore := store.NewMockStore(ctrl)
@@ -193,7 +164,7 @@ func TestHeartbeat(t *testing.T) {
 		require.Contains(t, err.Error(), expectedErr.Error())
 	})
 
-	// Test Case 7: Heartbeat with executor associated with local passthrough mode
+	// Test Case 6: Heartbeat with executor associated with local passthrough mode
 	t.Run("MigrationModeLocalPassthrough", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockStore := store.NewMockStore(ctrl)
@@ -222,7 +193,7 @@ func TestHeartbeat(t *testing.T) {
 		require.Contains(t, err.Error(), expectedErr.Error())
 	})
 
-	// Test Case 8: Heartbeat with executor associated with local passthrough shadow
+	// Test Case 7: Heartbeat with executor associated with local passthrough shadow
 	t.Run("MigrationModeLocalPassthroughWithAssignmentChanges", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockStore := store.NewMockStore(ctrl)
@@ -287,7 +258,7 @@ func TestHeartbeat(t *testing.T) {
 	},
 	)
 
-	// Test Case 9: Heartbeat with executor associated with distributed passthrough
+	// Test Case 8: Heartbeat with executor associated with distributed passthrough
 	t.Run("MigrationModeDISTRIBUTEDPASSTHROUGHDeletionFailure", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockStore := store.NewMockStore(ctrl)
@@ -330,7 +301,7 @@ func TestHeartbeat(t *testing.T) {
 		require.Contains(t, err.Error(), expectedErr.Error())
 	})
 
-	// Test Case 10: Heartbeat with executor associated with distributed passthrough
+	// Test Case 9: Heartbeat with executor associated with distributed passthrough
 	t.Run("MigrationModeDISTRIBUTEDPASSTHROUGHAssignmentFailure", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockStore := store.NewMockStore(ctrl)
@@ -374,7 +345,7 @@ func TestHeartbeat(t *testing.T) {
 		require.Contains(t, err.Error(), expectedErr.Error())
 	})
 
-	// Test Case 11: Heartbeat with metadata validation failure - too many keys
+	// Test Case 10: Heartbeat with metadata validation failure - too many keys
 	t.Run("MetadataValidationTooManyKeys", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockStore := store.NewMockStore(ctrl)
@@ -403,7 +374,7 @@ func TestHeartbeat(t *testing.T) {
 		require.Contains(t, err.Error(), "invalid metadata: metadata has 33 keys, which exceeds the maximum of 32")
 	})
 
-	// Test Case: Heartbeat with executor associated with MigrationModeLOCALPASSTHROUGH (should error)
+	// Test Case 11: Heartbeat with executor associated with MigrationModeLOCALPASSTHROUGH (should error)
 	t.Run("MigrationModeLOCALPASSTHROUGH", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockStore := store.NewMockStore(ctrl)