[azservicebus] Handle 410 properly for session and non-session based links. (#17382)

410 (lock lost) handling needed to be separated for sessions and non-session links.

For non-session links losing a link is essentially unrecoverable - the lock cannot be reobtained with just the lock token so further retries will always fail.

For _session_ based links it's different since you actually obtain a session lock by opening a link for a particular session ID. You can retry re-opening the link and there are cases where, on detach, that you might have some overlap until the session is actually available again (this is shown in the `TestSessionReceiver_Detach` test).

So the split here is simple - all retrying is really handled in amqpLinks, so now it takes a parameter giving it the proper function to determine if an error is fatal or retryable. As part of this I also made the parameters to amqpLinks an arg, which had some small rippling effects in callers.

Fixes #17325
(probable fix for the latest issue mentioned in #17017 since this affects settlement time)

Author: richardpark-msft

sdk/messaging/azservicebus/CHANGELOG.md

@@ -6,13 +6,13 @@
 
 - Support for using a SharedAccessSignature in a connection string. Ex: `Endpoint=sb://<sb>.servicebus.windows.net;SharedAccessSignature=SharedAccessSignature sr=<sb>.servicebus.windows.net&sig=<base64-sig>&se=<expiry>&skn=<keyname>` (#17314)
 
-### Breaking Changes
-
 ### Bugs Fixed
 
 - Fixed bug where message batch size calculation was inaccurate, resulting in batches that were too large to be sent. (#17318)
-
-### Other Changes
+- Fixing an issue with an entity not being found leading to a longer timeout than needed. (#17279)
+- Fixed the RPCLink so it does better handling of connection/link failures. (#17389)
+- Fixed issue where a message lock expiring would cause unnecessary retries. These retries could cause message settlement calls (ex: Receiver.CompleteMessage) 
+  to appear to hang. (#17382)
 
 ## 0.3.6 (2022-03-08)
 

sdk/messaging/azservicebus/client.go

@@ -156,9 +156,10 @@ func newClientImpl(creds clientCreds, options *ClientOptions) (*Client, error) {
 func (client *Client) NewReceiverForQueue(queueName string, options *ReceiverOptions) (*Receiver, error) {
 	id, cleanupOnClose := client.getCleanupForCloseable()
 	receiver, err := newReceiver(newReceiverArgs{
-		cleanupOnClose: cleanupOnClose,
-		ns:             client.namespace,
-		entity:         entity{Queue: queueName},
+		cleanupOnClose:      cleanupOnClose,
+		ns:                  client.namespace,
+		entity:              entity{Queue: queueName},
+		getRecoveryKindFunc: internal.GetRecoveryKind,
 	}, options)
 
 	if err != nil {
@@ -173,9 +174,10 @@ func (client *Client) NewReceiverForQueue(queueName string, options *ReceiverOpt
 func (client *Client) NewReceiverForSubscription(topicName string, subscriptionName string, options *ReceiverOptions) (*Receiver, error) {
 	id, cleanupOnClose := client.getCleanupForCloseable()
 	receiver, err := newReceiver(newReceiverArgs{
-		cleanupOnClose: cleanupOnClose,
-		ns:             client.namespace,
-		entity:         entity{Topic: topicName, Subscription: subscriptionName},
+		cleanupOnClose:      cleanupOnClose,
+		ns:                  client.namespace,
+		entity:              entity{Topic: topicName, Subscription: subscriptionName},
+		getRecoveryKindFunc: internal.GetRecoveryKind,
 	}, options)
 
 	if err != nil {

sdk/messaging/azservicebus/client_test.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"net/http"
 	"os"
 	"testing"
 	"time"
@@ -346,5 +347,5 @@ func assertRPCNotFound(t *testing.T, err error) {
 	}
 
 	require.ErrorAs(t, err, &rpcError)
-	require.Equal(t, 404, rpcError.RPCCode())
+	require.Equal(t, http.StatusNotFound, rpcError.RPCCode())
 }

sdk/messaging/azservicebus/internal/amqpLinks.go

@@ -49,7 +49,7 @@ type AMQPLinks interface {
 
 	// CloseIfNeeded closes the links or connection if the error is recoverable.
 	// Use this if you don't want to recreate the connection/links at this point.
-	CloseIfNeeded(ctx context.Context, err error) recoveryKind
+	CloseIfNeeded(ctx context.Context, err error) RecoveryKind
 
 	// ClosedPermanently is true if AMQPLinks.Close(ctx, true) has been called.
 	ClosedPermanently() bool
@@ -76,6 +76,8 @@ type AMQPLinksImpl struct {
 	audience       string
 	createLink     CreateLinkFunc
 
+	getRecoveryKindFunc func(err error) RecoveryKind
+
 	mu sync.RWMutex
 
 	// RPCLink lets you interact with the $management link for your entity.
@@ -103,16 +105,24 @@ type AMQPLinksImpl struct {
 // *amqp.Sender or a *amqp.Receiver. AMQPLinks handles it either way.
 type CreateLinkFunc func(ctx context.Context, session AMQPSession) (AMQPSenderCloser, AMQPReceiverCloser, error)
 
+type NewAMQPLinksArgs struct {
+	NS                  NamespaceForAMQPLinks
+	EntityPath          string
+	CreateLinkFunc      CreateLinkFunc
+	GetRecoveryKindFunc func(err error) RecoveryKind
+}
+
 // NewAMQPLinks creates a session, starts the claim refresher and creates an associated
 // management link for a specific entity path.
-func NewAMQPLinks(ns NamespaceForAMQPLinks, entityPath string, createLink CreateLinkFunc) AMQPLinks {
+func NewAMQPLinks(args NewAMQPLinksArgs) AMQPLinks {
 	l := &AMQPLinksImpl{
-		entityPath:        entityPath,
-		managementPath:    fmt.Sprintf("%s/$management", entityPath),
-		audience:          ns.GetEntityAudience(entityPath),
-		createLink:        createLink,
-		closedPermanently: false,
-		ns:                ns,
+		entityPath:          args.EntityPath,
+		managementPath:      fmt.Sprintf("%s/$management", args.EntityPath),
+		audience:            args.NS.GetEntityAudience(args.EntityPath),
+		createLink:          args.CreateLinkFunc,
+		closedPermanently:   false,
+		getRecoveryKindFunc: args.GetRecoveryKindFunc,
+		ns:                  args.NS,
 	}
 
 	return l
@@ -189,17 +199,17 @@ func (links *AMQPLinksImpl) RecoverIfNeeded(ctx context.Context, theirID LinkID,
 	default:
 	}
 
-	sbe := GetSBErrInfo(origErr)
+	rk := links.getRecoveryKindFunc(origErr)
 
-	if sbe.RecoveryKind == RecoveryKindLink {
+	if rk == RecoveryKindLink {
 		if err := links.recoverLink(ctx, theirID); err != nil {
 			log.Writef(EventConn, "failed to recreate link: %s", err.Error())
 			return err
 		}
 
 		log.Writef(EventConn, "Recovered links")
 		return nil
-	} else if sbe.RecoveryKind == RecoveryKindConn {
+	} else if rk == RecoveryKindConn {
 		if err := links.recoverConnection(ctx, theirID); err != nil {
 			log.Writef(EventConn, "failed to recreate connection: %s", err.Error())
 			return err
@@ -293,6 +303,10 @@ func (l *AMQPLinksImpl) Retry(ctx context.Context, name string, fn RetryWithLink
 
 	didQuickRetry := false
 
+	isFatalErrorFunc := func(err error) bool {
+		return l.getRecoveryKindFunc(err) == RecoveryKindFatal
+	}
+
 	return utils.Retry(ctx, name, func(ctx context.Context, args *utils.RetryFnArgs) error {
 		if err := l.RecoverIfNeeded(ctx, lastID, args.LastErr); err != nil {
 			return err
@@ -335,7 +349,7 @@ func (l *AMQPLinksImpl) Retry(ctx context.Context, name string, fn RetryWithLink
 		}
 
 		return nil
-	}, IsFatalSBError, o)
+	}, isFatalErrorFunc, o)
 }
 
 // EntityPath is the full entity path for the queue/topic/subscription.
@@ -368,11 +382,11 @@ func (l *AMQPLinksImpl) Close(ctx context.Context, permanent bool) error {
 // eats the cost of recovery, instead of doing it immediately. This is useful
 // if you're trying to exit out of a function quickly but still need to react
 // to a returned error.
-func (l *AMQPLinksImpl) CloseIfNeeded(ctx context.Context, err error) recoveryKind {
+func (l *AMQPLinksImpl) CloseIfNeeded(ctx context.Context, err error) RecoveryKind {
 	l.mu.Lock()
 	defer l.mu.Unlock()
 
-	rk := GetRecoveryKind(err)
+	rk := l.getRecoveryKindFunc(err)
 
 	switch rk {
 	case RecoveryKindLink: