[azservicebus] Removing the idle tracking code. (Azure#19581)

Client-side idle tracking was added in v1.1.2, but it seems be causing issues for customers and is actually making their receiving _less_ reliable. I had previously removed it for just sessions, but I'm now also removing it for non-session receivers as well.

I'm leaving the idle check code, just not hooked up for now, because it still seems like a good defensive feature but it requires more testing and stressing to be fit for production.

This rolls back the changes in Azure#19465 (simple idle timer), Azure#19492 (ignore ctx on link close) and Azure#19506 (track idle across multiple receives).

Author: richardpark-msft

sdk/messaging/azservicebus/CHANGELOG.md

@@ -1,16 +1,12 @@
 # Release History
 
-## 1.1.3 (Unreleased)
-
-### Features Added
-
-### Breaking Changes
+## 1.1.3 (2022-11-16)
 
 ### Bugs Fixed
 
-- Disable the idle link tracking on session links (#19571)
-
-### Other Changes
+- Removing changes for client-side idle timer and closing without timeout. Combined these are 
+  causing issues with links not properly recovering or closing. Investigating an alternative
+  for a future release.
 
 ## 1.1.2 (2022-11-08)
 

sdk/messaging/azservicebus/client.go

@@ -167,10 +167,9 @@ func (client *Client) NewReceiverForQueue(queueName string, options *ReceiverOpt
 	id, cleanupOnClose := client.getCleanupForCloseable()
 	receiver, err := newReceiver(newReceiverArgs{
 		cleanupOnClose:      cleanupOnClose,
+		ns:                  client.namespace,
 		entity:              entity{Queue: queueName},
 		getRecoveryKindFunc: internal.GetRecoveryKind,
-		idleTimeout:         0,
-		ns:                  client.namespace,
 		retryOptions:        client.retryOptions,
 	}, options)
 
@@ -187,10 +186,9 @@ func (client *Client) NewReceiverForSubscription(topicName string, subscriptionN
 	id, cleanupOnClose := client.getCleanupForCloseable()
 	receiver, err := newReceiver(newReceiverArgs{
 		cleanupOnClose:      cleanupOnClose,
+		ns:                  client.namespace,
 		entity:              entity{Topic: topicName, Subscription: subscriptionName},
 		getRecoveryKindFunc: internal.GetRecoveryKind,
-		idleTimeout:         0,
-		ns:                  client.namespace,
 		retryOptions:        client.retryOptions,
 	}, options)
 

sdk/messaging/azservicebus/go.mod

@@ -2,6 +2,10 @@ module github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus
 
 go 1.18
 
+retract (
+    v1.1.2 // Breaks customers in situations where close is slow/infinite.
+)
+
 require (
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.0.0
 	github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.0.0

sdk/messaging/azservicebus/internal/amqpLinks.go

@@ -468,9 +468,7 @@ func (l *AMQPLinksImpl) initWithoutLocking(ctx context.Context) error {
 
 // close closes the link.
 // NOTE: No locking is done in this function, call `Close` if you require locking.
-func (l *AMQPLinksImpl) closeWithoutLocking(_ context.Context, permanent bool) error {
-	ctx := context.Background()
-
+func (l *AMQPLinksImpl) closeWithoutLocking(ctx context.Context, permanent bool) error {
 	if l.closedPermanently {
 		return nil
 	}

sdk/messaging/azservicebus/internal/amqp_test_utils.go

@@ -43,9 +43,6 @@ type FakeAMQPLinks struct {
 
 	Closed              int
 	CloseIfNeededCalled int
-	CloseIfNeededArgs   []error
-
-	GetFn func(ctx context.Context) (*LinksWithID, error)
 
 	// values to be returned for each `Get` call
 	Revision LinkID
@@ -191,10 +188,6 @@ func (l *FakeAMQPLinks) Get(ctx context.Context) (*LinksWithID, error) {
 	case <-ctx.Done():
 		return nil, ctx.Err()
 	default:
-		if l.GetFn != nil {
-			return l.GetFn(ctx)
-		}
-
 		return &LinksWithID{
 			Sender:   l.Sender,
 			Receiver: l.Receiver,
@@ -225,7 +218,6 @@ func (l *FakeAMQPLinks) Close(ctx context.Context, permanently bool) error {
 
 func (l *FakeAMQPLinks) CloseIfNeeded(ctx context.Context, err error) RecoveryKind {
 	l.CloseIfNeededCalled++
-	l.CloseIfNeededArgs = append(l.CloseIfNeededArgs, err)
 	return GetRecoveryKind(err)
 }
 

sdk/messaging/azservicebus/internal/errors.go

@@ -169,10 +169,6 @@ func GetRecoveryKind(err error) RecoveryKind {
 		return RecoveryKindFatal
 	}
 
-	if IsLocalIdleError(err) {
-		return RecoveryKindLink
-	}
-
 	var netErr net.Error
 
 	// these are errors that can flow from the go-amqp connection to

sdk/messaging/azservicebus/internal/errors_test.go

@@ -286,7 +286,3 @@ func Test_TransformError(t *testing.T) {
 	// and it's okay, for convenience, to pass a nil.
 	require.Nil(t, TransformError(nil))
 }
-
-func Test_IdleError(t *testing.T) {
-	require.Equal(t, RecoveryKindLink, GetRecoveryKind(localIdleError))
-}

sdk/messaging/azservicebus/internal/idle_check.go

@@ -11,11 +11,9 @@ import (
 	"sync"
 	"time"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/internal/log"
 	azlog "github.com/Azure/azure-sdk-for-go/sdk/internal/log"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/uuid"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/exported"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/go-amqp"
 )
 
@@ -100,14 +98,6 @@ type RPCLinkArgs struct {
 	LogEvent azlog.Event
 }
 
-func closeOrLog(name string, closeable interface {
-	Close(ctx context.Context) error
-}) {
-	if err := closeable.Close(context.Background()); err != nil {
-		log.Writef(exported.EventAuth, "Failed closing %s for RPC Link: %s", name, err.Error())
-	}
-}
-
 // NewRPCLink will build a new request response link
 func NewRPCLink(ctx context.Context, args RPCLinkArgs) (*rpcLink, error) {
 	session, err := args.Client.NewSession(ctx, nil)
@@ -118,7 +108,6 @@ func NewRPCLink(ctx context.Context, args RPCLinkArgs) (*rpcLink, error) {
 
 	linkID, err := uuid.New()
 	if err != nil {
-		closeOrLog("session", session)
 		return nil, err
 	}
 
@@ -140,7 +129,6 @@ func NewRPCLink(ctx context.Context, args RPCLinkArgs) (*rpcLink, error) {
 		nil,
 	)
 	if err != nil {
-		closeOrLog("session", session)
 		return nil, err
 	}
 
@@ -161,8 +149,11 @@ func NewRPCLink(ctx context.Context, args RPCLinkArgs) (*rpcLink, error) {
 
 	receiver, err := session.NewReceiver(ctx, args.Address, receiverOpts)
 	if err != nil {
-		closeOrLog("sender", sender)
-		closeOrLog("session", session)
+		// make sure we close the sender
+		clsCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		_ = sender.Close(clsCtx)
 		return nil, err
 	}
 
@@ -331,11 +322,7 @@ func (l *rpcLink) RPC(ctx context.Context, msg *amqp.Message) (*RPCResponse, err
 }
 
 // Close the link receiver, sender and session
-func (l *rpcLink) Close(_ context.Context) error {
-	// we're finding, in practice, that allowing cancellations when cleaning up state
-	// just results in inconsistencies. We'll cut cancellation off here for now.
-	ctx := context.Background()
-
+func (l *rpcLink) Close(ctx context.Context) error {
 	l.rpcLinkCtxCancel()
 
 	if err := l.closeReceiver(ctx); err != nil {