Enforce timeout when updating shard
@@ -79,6 +79,10 @@ const (
contextStateStopped
)
+const (
+ shardIOTimeout = 5 * time.Second
+)
+
type (
contextState int32
@@ -1140,7 +1144,9 @@ func (s *ContextImpl) renewRangeLocked(isStealing bool) error {
updatedShardInfo.StolenSinceRenew++
}
- err := s.persistenceShardManager.UpdateShard(context.TODO(), &persistence.UpdateShardRequest{
+ ctx, cancel := context.WithTimeout(context.Background(), shardIOTimeout)
+ defer cancel()
+ err := s.persistenceShardManager.UpdateShard(ctx, &persistence.UpdateShardRequest{
ShardInfo: updatedShardInfo.ShardInfo,
PreviousRangeID: s.shardInfo.GetRangeId()})
if err != nil {
@@ -1190,7 +1196,9 @@ func (s *ContextImpl) updateShardInfoLocked() error {
updatedShardInfo := copyShardInfo(s.shardInfo)
s.emitShardInfoMetricsLogsLocked()
- err = s.persistenceShardManager.UpdateShard(context.TODO(), &persistence.UpdateShardRequest{
+ ctx, cancel := context.WithTimeout(context.Background(), shardIOTimeout)
+ defer cancel()
+ err = s.persistenceShardManager.UpdateShard(ctx, &persistence.UpdateShardRequest{
ShardInfo: updatedShardInfo.ShardInfo,
PreviousRangeID: s.shardInfo.GetRangeId(),
})
@@ -1640,7 +1648,9 @@ func (s *ContextImpl) loadShardMetadata(ownershipChanged *bool) error {
s.rUnlock()
// We don't have any shardInfo yet, load it (outside of context rwlock)
- resp, err := s.persistenceShardManager.GetOrCreateShard(context.TODO(), &persistence.GetOrCreateShardRequest{
+ ctx, cancel := context.WithTimeout(context.Background(), shardIOTimeout)
+ defer cancel()
+ resp, err := s.persistenceShardManager.GetOrCreateShard(ctx, &persistence.GetOrCreateShardRequest{
ShardID: s.shardID,
LifecycleContext: s.lifecycleCtx,
})
