Add dynamic config for ShardIOTimeout (#5966)

## What changed?
<!-- Describe what has changed in this PR -->
Made `shardIOTimeout` configurable via dynamic config.

## Why?
<!-- Tell your future self why have you made these changes -->
Depending on the persistence backend, a longer timeout may be necessary
in some cases.

## How did you test it?
<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
Existing tests should cover this change, and since the defaults are the
same as the prior values, no impact is expected. We will also test this
in a test environment.

## Potential risks
<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
An improperly configured timeout could cause poor performance and hide
issues with the persistence layer.

## Documentation
<!-- Have you made sure this change doesn't falsify anything currently
stated in `docs/`? If significant
new behavior is added, have you described that in `docs/`? -->

## Is hotfix candidate?
<!-- Is this PR a hotfix candidate or does it require a notification to
be sent to the broader community? (Yes/No) -->

common/dynamicconfig/constants.go

@@ -30,6 +30,7 @@ import (
 
 	enumspb "go.temporal.io/api/enums/v1"
 
+	"go.temporal.io/server/common/debug"
 	"go.temporal.io/server/common/primitives"
 	"go.temporal.io/server/common/retrypolicy"
 )
@@ -1337,6 +1338,11 @@ to this require a restart to take effect.`,
 		1,
 		`ShardIOConcurrency controls the concurrency of persistence operations in shard context`,
 	)
+	ShardIOTimeout = NewGlobalDurationSetting(
+		"history.shardIOTimeout",
+		5*time.Second*debug.TimeoutMultiplier,
+		`ShardIOTimeout sets the timeout for persistence operations in the shard context`,
+	)
 	StandbyClusterDelay = NewGlobalDurationSetting(
 		"history.standbyClusterDelay",
 		5*time.Minute,

service/history/configs/config.go

@@ -92,6 +92,7 @@ type Config struct {
 	AcquireShardInterval           dynamicconfig.DurationPropertyFn
 	AcquireShardConcurrency        dynamicconfig.IntPropertyFn
 	ShardIOConcurrency             dynamicconfig.IntPropertyFn
+	ShardIOTimeout                 dynamicconfig.DurationPropertyFn
 	ShardLingerOwnershipCheckQPS   dynamicconfig.IntPropertyFn
 	ShardLingerTimeLimit           dynamicconfig.DurationPropertyFn
 	ShardOwnershipAssertionEnabled dynamicconfig.BoolPropertyFn
@@ -404,6 +405,7 @@ func NewConfig(
 		AcquireShardInterval:           dynamicconfig.AcquireShardInterval.Get(dc),
 		AcquireShardConcurrency:        dynamicconfig.AcquireShardConcurrency.Get(dc),
 		ShardIOConcurrency:             dynamicconfig.ShardIOConcurrency.Get(dc),
+		ShardIOTimeout:                 dynamicconfig.ShardIOTimeout.Get(dc),
 		ShardLingerOwnershipCheckQPS:   dynamicconfig.ShardLingerOwnershipCheckQPS.Get(dc),
 		ShardLingerTimeLimit:           dynamicconfig.ShardLingerTimeLimit.Get(dc),
 		ShardOwnershipAssertionEnabled: dynamicconfig.ShardOwnershipAssertionEnabled.Get(dc),

service/history/shard/context_impl.go

@@ -89,7 +89,6 @@ const (
 )
 
 const (
-	shardIOTimeout = 5 * time.Second * debug.TimeoutMultiplier
 	// ShardUpdateQueueMetricsInterval is the minimum amount of time between updates to a shard's queue metrics
 	queueMetricUpdateInterval = 5 * time.Minute
 
@@ -249,7 +248,7 @@ func (s *ContextImpl) GetPingChecks() []common.PingCheck {
 			Name: s.String() + "-shard-lock",
 			// rwLock may be held for the duration of renewing shard rangeID, which are called with a
 			// timeout of shardIOTimeout. add a few more seconds for reliability.
-			Timeout: shardIOTimeout + 5*time.Second,
+			Timeout: s.config.ShardIOTimeout() + 5*time.Second,
 			Ping: func() []common.Pingable {
 				// call rwLock.Lock directly to bypass metrics since this isn't a real request
 				s.rwLock.Lock()
@@ -1910,7 +1909,7 @@ func (s *ContextImpl) getOrUpdateRemoteClusterInfoLocked(clusterName string) *re
 func (s *ContextImpl) acquireShard() {
 	// This is called in two contexts: initially acquiring the rangeid lock, and trying to
 	// re-acquire it after a persistence error. In both cases, we retry the acquire operation
-	// (renewRangeLocked) for 5 minutes. Each individual attempt uses shardIOTimeout (5s) as
+	// (renewRangeLocked) for 5 minutes. Each individual attempt uses shardIOTimeout (by default, 5s) as
 	// the timeout. This lets us handle a few minutes of persistence unavailability without
 	// dropping and reloading the whole shard context, which is relatively expensive (includes
 	// caches that would have to be refilled, etc.).
@@ -2220,7 +2219,7 @@ func (s *ContextImpl) newDetachedContext(
 }
 
 func (s *ContextImpl) newIOContext() (context.Context, context.CancelFunc) {
-	ctx, cancel := context.WithTimeout(s.lifecycleCtx, shardIOTimeout)
+	ctx, cancel := context.WithTimeout(s.lifecycleCtx, s.config.ShardIOTimeout())
 	ctx = headers.SetCallerInfo(ctx, headers.SystemBackgroundCallerInfo)
 
 	return ctx, cancel

service/history/shard/controller_impl.go

@@ -427,7 +427,7 @@ func (c *ControllerImpl) acquireShards(ctx context.Context) {
 		defer engineCancel()
 		_, _ = shard.GetEngine(engineCtx)
 
-		assertCtx, assertCancel := context.WithTimeout(ctx, shardIOTimeout)
+		assertCtx, assertCancel := context.WithTimeout(ctx, c.config.ShardIOTimeout())
 		defer assertCancel()
 		// trust the AssertOwnership will handle shard ownership lost
 		_ = shard.AssertOwnership(assertCtx)