Adding dead-man's switch

common/util.go

@@ -29,6 +29,7 @@ import (
 	"math/rand"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/dgryski/go-farm"
@@ -142,6 +143,22 @@ func AwaitWaitGroup(wg *sync.WaitGroup, timeout time.Duration) bool {
 	}
 }
 
+func WatchMissedDeadline(shouldBeDoneBy time.Duration, deadMansSwitch func()) (isDone func()) {
+	done := atomic.Bool{}
+	go func() {
+		time.Sleep(shouldBeDoneBy)
+		isDone := done.Load()
+		if isDone {
+			return
+		}
+		deadMansSwitch()
+	}()
+
+	return func() {
+		done.Swap(true)
+	}
+}
+
 // CreatePersistenceRetryPolicy creates a retry policy for persistence layer operations
 func CreatePersistenceRetryPolicy() backoff.RetryPolicy {
 	policy := backoff.NewExponentialRetryPolicy(retryPersistenceOperationInitialInterval)

common/util_test.go

@@ -1721,3 +1721,22 @@ func TestCheckEventBlobSizeLimit(t *testing.T) {
 		})
 	}
 }
+
+func TestWatchMissedDeadlineDoesNotFireWhenDone(t *testing.T) {
+	fired := false
+	isDone := WatchMissedDeadline(time.Millisecond*10, func() {
+		fired = true
+	})
+	isDone()
+	assert.False(t, fired)
+}
+
+func TestWatchMissedDeadlineDoesFireWhenNotDone(t *testing.T) {
+	fired := false
+	isDone := WatchMissedDeadline(time.Millisecond*10, func() {
+		fired = true
+	})
+	time.Sleep(time.Millisecond * 11)
+	isDone()
+	assert.True(t, fired)
+}

service/history/queue/transfer_queue_processor.go

@@ -192,13 +192,17 @@ func (t *transferQueueProcessor) Stop() {
 	defer t.logger.Info("Transfer queue processor stopped")
 
 	if !t.shard.GetConfig().QueueProcessorEnableGracefulSyncShutdown() {
+		isDone := common.WatchMissedDeadline(time.Minute, func() {
+			t.logger.Error("Graceful shutdown of transfer queue processor has failed")
+		})
 		t.activeQueueProcessor.Stop()
 		for _, standbyQueueProcessor := range t.standbyQueueProcessors {
 			standbyQueueProcessor.Stop()
 		}
 
 		close(t.shutdownChan)
 		common.AwaitWaitGroup(&t.shutdownWG, time.Minute)
+		isDone()
 		return
 	}