forked from Azure/azure-sdk-for-go
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[azservicebus] Make the client side idle timer work across multiple c…
…alls (Azure#19506) I added in a simple idle timer in Azure#19465, which would expire the link if our internal message receive went longer than 5 minutes. This extends that to track it across multiple consecutive calls as well, in case the user calls and cancels multiple times in a row, eating up 5 minutes of wall-clock time. This is actually pretty similar to the workaround applied by the customer here in Azure#18517 but tries to take into account multiple calls and also recovers the link without exiting ReceiveMessages().
- Loading branch information
1 parent
8606a5b
commit 719d2cb
Showing
8 changed files
with
296 additions
and
96 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
|
|
||
| package internal | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "time" | ||
| ) | ||
|
|
||
| type LocalIdleTracker struct { | ||
| // MaxDuration controls how long we'll wait, on the client side, for the first message | ||
| MaxDuration time.Duration | ||
|
|
||
| // IdleStart marks the first time the user cancelled in a string of cancellations. | ||
| // It gets reset any time there is a success or a non-cancel related failure. | ||
| // NOTE: this is public for some unit tests but isn't intended to be set by external code. | ||
| IdleStart time.Time | ||
| } | ||
|
|
||
| var localIdleError = errors.New("link was idle, detaching (will be reattached).") | ||
|
|
||
| func IsLocalIdleError(err error) bool { | ||
| return errors.Is(err, localIdleError) | ||
| } | ||
|
|
||
| // NewContextWithDeadline creates a context that has an appropriate deadline that will expire | ||
| // when the idle period has completed. | ||
| func (idle *LocalIdleTracker) NewContextWithDeadline(ctx context.Context) (context.Context, context.CancelFunc) { | ||
| if idle.IdleStart.IsZero() { | ||
| // we're not in the middle of an idle period, so we'll start from now. | ||
| return context.WithTimeout(ctx, idle.MaxDuration) | ||
| } | ||
|
|
||
| // we've already idled before. | ||
| return context.WithDeadline(ctx, idle.IdleStart.Add(idle.MaxDuration)) | ||
| } | ||
|
|
||
| // Check checks if we are actually idle, taking into account when we initially | ||
| // started being idle ([idle.IdleStart]) vs the current time. | ||
| // | ||
| // If it turns out the link should be considered idle it'll return idleError. | ||
| // Else, it'll return the err parameter. | ||
| func (idle *LocalIdleTracker) Check(parentCtx context.Context, operationStart time.Time, err error) error { | ||
| if err == nil || !IsCancelError(err) { | ||
| // either no error occurred (in which case the link is working) | ||
| // or a non-cancel error happened. The non-cancel error will just | ||
| // be handled by the normal recovery path. | ||
| idle.IdleStart = time.Time{} | ||
| return err | ||
| } | ||
|
|
||
| // okay, we're dealing with a cancellation error. Was it the user cancelling (ie, parentCtx) or | ||
| // was it our idle timer firing? | ||
| if parentCtx.Err() != nil { | ||
| // The user cancelled. These cancels come from a single Receive() call on the link, which means we | ||
| // didn't get a message back. | ||
| if idle.IdleStart.IsZero() { | ||
| idle.IdleStart = operationStart | ||
| } | ||
|
|
||
| return err | ||
| } | ||
|
|
||
| // It's our idle timeout that caused us to cancel, which means the idle interval has expired. | ||
| // We'll clear our internally stored time and indicate we're idle with the sentinel 'idleError' | ||
| idle.IdleStart = time.Time{} | ||
|
|
||
| return localIdleError | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,112 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
|
|
||
| package internal | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestLocalIdleTracker(t *testing.T) { | ||
| t.Run("is hierarchical", func(t *testing.T) { | ||
| idleTracker := &LocalIdleTracker{MaxDuration: time.Hour} | ||
|
|
||
| parentCtx, cancelParent := context.WithCancel(context.Background()) | ||
|
|
||
| ctx, cancel := idleTracker.NewContextWithDeadline(parentCtx) | ||
| defer cancel() | ||
|
|
||
| require.NoError(t, ctx.Err()) | ||
|
|
||
| cancelParent() | ||
|
|
||
| require.ErrorIs(t, ctx.Err(), context.Canceled) | ||
| }) | ||
|
|
||
| t.Run("expires in MaxDuration if no previous cancel time exists", func(t *testing.T) { | ||
| maxDuration := 2 * time.Second | ||
|
|
||
| idleTracker := &LocalIdleTracker{ | ||
| MaxDuration: maxDuration, | ||
| } | ||
|
|
||
| ctx, cancel := idleTracker.NewContextWithDeadline(context.Background()) | ||
| defer cancel() | ||
| require.Nil(t, ctx.Err()) | ||
|
|
||
| deadline, ok := ctx.Deadline() | ||
| require.True(t, ok) | ||
| require.GreaterOrEqual(t, deadline.Add(-maxDuration).UnixNano(), int64(0), "our deadline was set appropriately into the future") | ||
| }) | ||
|
|
||
| t.Run("with a previous cancel", func(t *testing.T) { | ||
| maxWait := 2 * time.Second | ||
| idleStartTime := time.Now().Add(time.Hour) | ||
|
|
||
| idleTracker := &LocalIdleTracker{ | ||
| MaxDuration: maxWait, | ||
| IdleStart: idleStartTime, | ||
| } | ||
|
|
||
| ctx, cancel := idleTracker.NewContextWithDeadline(context.Background()) | ||
| defer cancel() | ||
| require.Nil(t, ctx.Err()) | ||
|
|
||
| deadline, ok := ctx.Deadline() | ||
| require.True(t, ok) | ||
| require.Equal(t, deadline.Add(-maxWait), idleStartTime, "deadline used our idle start time as the base, not time.Now()") | ||
| }) | ||
|
|
||
| t.Run("user cancels", func(t *testing.T) { | ||
| idleTracker := &LocalIdleTracker{ | ||
| MaxDuration: 30 * time.Minute, | ||
| } | ||
|
|
||
| parentCtx, cancelParent := context.WithCancel(context.Background()) | ||
| cancelParent() | ||
|
|
||
| // The user cancelled here - since that cancellation is specifically for the _first_ | ||
| // message it means they didn't receive anything. If they do this for long enough the | ||
| // link will be considered idle. | ||
|
|
||
| twoHoursFromNow := time.Now().Add(2 * time.Hour) | ||
| require.Zero(t, idleTracker.IdleStart) | ||
| err := idleTracker.Check(parentCtx, twoHoursFromNow, context.Canceled) | ||
|
|
||
| require.ErrorIs(t, err, context.Canceled) | ||
| require.Equal(t, idleTracker.IdleStart, twoHoursFromNow, "time of first cancel is recorded (gets used as the base for future idle calculations)") | ||
|
|
||
| // now we have a successful call, and it resets the idle time back to zero (ie, we're no longer in danger of being idle) | ||
| err = idleTracker.Check(parentCtx, twoHoursFromNow, nil) | ||
| require.NoError(t, err) | ||
| require.Zero(t, idleTracker.IdleStart, "a successful call resets our idle tracking") | ||
|
|
||
| // we also reset the idle time back to zero if there's any other error since | ||
| // those errors will be dealt with by the recovery code. | ||
| idleTracker.IdleStart = time.Now() | ||
|
|
||
| err = idleTracker.Check(parentCtx, twoHoursFromNow, errors.New("some other error")) | ||
| require.EqualError(t, err, "some other error") | ||
| require.Zero(t, idleTracker.IdleStart, "an error is also considered as proof that the link is alive") | ||
| }) | ||
|
|
||
| t.Run("idle deadline expires", func(t *testing.T) { | ||
| twoHoursAndOneMinuteAgo := time.Now().Add(-time.Hour - time.Minute) | ||
| idleTracker := &LocalIdleTracker{ | ||
| MaxDuration: time.Hour, | ||
| IdleStart: twoHoursAndOneMinuteAgo, | ||
| } | ||
|
|
||
| parentCtx, cancelParent := context.WithCancel(context.Background()) | ||
| defer cancelParent() | ||
|
|
||
| err := idleTracker.Check(parentCtx, time.Now(), context.DeadlineExceeded) | ||
| require.ErrorIs(t, err, localIdleError) | ||
| require.True(t, IsLocalIdleError(err)) | ||
| }) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.