[azservicebus] Update go-amqp (Azure#18345)

Updating our internally vendored go-amqp to 18d378101570d5469d237799a76b77c0bf0c6fe3

This mostly impacted us because the connection error is no longer a sentinel error.

Author: richardpark-msft

sdk/messaging/azservicebus/internal/amqpLinks_test.go

@@ -33,19 +33,22 @@ func (pe fakeNetError) Timeout() bool   { return pe.timeout }
 func (pe fakeNetError) Temporary() bool { return pe.temp }
 func (pe fakeNetError) Error() string   { return "Fake but very permanent error" }
 
-func assertFailedLinks(t *testing.T, lwid *LinksWithID, expectedErr error, expectedRPCError error) {
+func assertFailedLinks[T error, T2 error](t *testing.T, lwid *LinksWithID, expectedErr T, expectedRPCError T2) {
 	err := lwid.Sender.Send(context.TODO(), &amqp.Message{
 		Data: [][]byte{
 			{0},
 		},
 	})
+
+	require.True(t, errors.Is(err, expectedErr) || errors.As(err, &expectedErr))
 	require.ErrorIs(t, err, expectedErr)
 
 	_, err = PeekMessages(context.TODO(), lwid.RPC, lwid.Receiver.LinkName(), 0, 1)
-	require.ErrorIs(t, err, expectedRPCError)
+	require.True(t, errors.Is(err, expectedRPCError) || errors.As(err, &expectedRPCError))
 
 	msg, err := lwid.Receiver.Receive(context.TODO())
 	require.ErrorIs(t, err, expectedErr)
+	require.True(t, errors.Is(err, expectedErr) || errors.As(err, &expectedErr))
 	require.Nil(t, msg)
 
 }
@@ -126,7 +129,7 @@ func TestAMQPLinksLive(t *testing.T) {
 	}()
 
 	require.EqualValues(t, 0, createLinksCalled)
-	require.NoError(t, links.RecoverIfNeeded(context.Background(), LinkID{}, amqp.ErrConnClosed))
+	require.NoError(t, links.RecoverIfNeeded(context.Background(), LinkID{}, &amqp.ConnectionError{}))
 	require.EqualValues(t, 1, createLinksCalled)
 
 	lwr, err := links.Get(context.Background())
@@ -138,10 +141,10 @@ func TestAMQPLinksLive(t *testing.T) {
 	require.NoError(t, amqpClient.Close())
 
 	// all the links are dead because the connection is dead.
-	assertFailedLinks(t, lwr, amqp.ErrConnClosed, amqp.ErrConnClosed)
+	assertFailedLinks(t, lwr, &amqp.ConnectionError{}, &amqp.ConnectionError{})
 
 	// now we'll recover, which should recreate everything
-	require.NoError(t, links.RecoverIfNeeded(context.Background(), lwr.ID, amqp.ErrConnClosed))
+	require.NoError(t, links.RecoverIfNeeded(context.Background(), lwr.ID, &amqp.ConnectionError{}))
 	require.EqualValues(t, 2, createLinksCalled)
 
 	lwr, err = links.Get(context.Background())
@@ -199,7 +202,7 @@ func TestAMQPLinksLiveRecoverLink(t *testing.T) {
 	}()
 
 	require.EqualValues(t, 0, createLinksCalled)
-	require.NoError(t, links.RecoverIfNeeded(context.Background(), LinkID{}, amqp.ErrConnClosed))
+	require.NoError(t, links.RecoverIfNeeded(context.Background(), LinkID{}, &amqp.ConnectionError{}))
 	require.EqualValues(t, 1, createLinksCalled)
 
 	lwr, err := links.Get(context.Background())
@@ -242,7 +245,7 @@ func TestAMQPLinksLiveRace(t *testing.T) {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			err := links.RecoverIfNeeded(context.Background(), LinkID{}, amqp.ErrConnClosed)
+			err := links.RecoverIfNeeded(context.Background(), LinkID{}, &amqp.ConnectionError{})
 			require.NoError(t, err)
 		}()
 	}
@@ -334,7 +337,7 @@ func TestAMQPLinksRetry(t *testing.T) {
 
 	err = links.Retry(context.Background(), log.Event("NotUsed"), "NotUsed", func(ctx context.Context, lwid *LinksWithID, args *utils.RetryFnArgs) error {
 		// force recoveries
-		return amqp.ErrConnClosed
+		return &amqp.ConnectionError{}
 	}, exported.RetryOptions{
 		MaxRetries: 2,
 		// note: omitting MaxRetries just to give a sanity check that
@@ -343,7 +346,8 @@ func TestAMQPLinksRetry(t *testing.T) {
 		MaxRetryDelay: time.Millisecond,
 	})
 
-	require.ErrorIs(t, err, amqp.ErrConnClosed)
+	var connErr *amqp.ConnectionError
+	require.ErrorAs(t, err, &connErr)
 	require.EqualValues(t, 3, createLinksCalled)
 }
 
@@ -530,7 +534,7 @@ func TestAMQPLinksCloseIfNeeded(t *testing.T) {
 		_, err := links.Get(context.Background())
 		require.NoError(t, err)
 
-		rk := links.CloseIfNeeded(context.Background(), amqp.ErrConnClosed)
+		rk := links.CloseIfNeeded(context.Background(), &amqp.ConnectionError{})
 		require.Equal(t, RecoveryKindConn, rk)
 		require.Equal(t, 1, receiver.Closed)
 		require.Equal(t, 1, sender.Closed)
@@ -577,7 +581,7 @@ func TestAMQPLinksRetriesUnit(t *testing.T) {
 		{Err: nil, Attempts: []int32{0}},
 
 		// connection related or unknown failures happen, all attempts exhausted
-		{Err: amqp.ErrConnClosed, Attempts: []int32{0, 1, 2, 3}},
+		{Err: &amqp.ConnectionError{}, Attempts: []int32{0, 1, 2, 3}},
 		{Err: errors.New("unknown error"), Attempts: []int32{0, 1, 2, 3}},
 
 		// fatal errors don't retry at all.
@@ -698,7 +702,7 @@ func TestAMQPLinks_Logging(t *testing.T) {
 		endCapture := test.CaptureLogsForTest()
 		defer endCapture()
 
-		err := links.RecoverIfNeeded(context.Background(), LinkID{}, amqp.ErrConnClosed)
+		err := links.RecoverIfNeeded(context.Background(), LinkID{}, &amqp.ConnectionError{})
 		require.NoError(t, err)
 
 		messages := endCapture()

sdk/messaging/azservicebus/internal/errors.go

@@ -173,7 +173,9 @@ func GetRecoveryKind(err error) RecoveryKind {
 		return RecoveryKindLink
 	}
 
-	if errors.Is(err, amqp.ErrConnClosed) ||
+	var connErr *amqp.ConnectionError
+
+	if errors.As(err, &connErr) ||
 		// session closures appear to leak through when the connection itself is going down.
 		errors.Is(err, amqp.ErrSessionClosed) {
 		return RecoveryKindConn

sdk/messaging/azservicebus/internal/errors_test.go

@@ -115,7 +115,7 @@ func Test_recoveryKind(t *testing.T) {
 		}
 
 		t.Run("sentinel errors", func(t *testing.T) {
-			rk := GetRecoveryKind(amqp.ErrConnClosed)
+			rk := GetRecoveryKind(&amqp.ConnectionError{})
 			require.EqualValues(t, RecoveryKindConn, rk)
 		})
 	})
@@ -188,7 +188,7 @@ func Test_ServiceBusError_ConnectionRecoveryNeeded(t *testing.T) {
 	var connErrors = []error{
 		&amqp.Error{Condition: amqp.ErrorConnectionForced},
 		&amqp.Error{Condition: amqp.ErrorInternalError},
-		amqp.ErrConnClosed,
+		&amqp.ConnectionError{},
 		amqp.ErrSessionClosed,
 		io.EOF,
 		fakeNetError{temp: true},
@@ -274,7 +274,7 @@ func Test_TransformError(t *testing.T) {
 	require.ErrorAs(t, err, &asExportedErr)
 	require.Equal(t, exported.CodeConnectionLost, asExportedErr.Code)
 
-	err = TransformError(amqp.ErrConnClosed)
+	err = TransformError(&amqp.ConnectionError{})
 	require.ErrorAs(t, err, &asExportedErr)
 	require.Equal(t, exported.CodeConnectionLost, asExportedErr.Code)
 

sdk/messaging/azservicebus/internal/go-amqp/conn.go

@@ -8,7 +8,6 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
-	"io"
 	"math"
 	"net"
 	"net/url"
@@ -362,7 +361,10 @@ func (c *conn) Start() error {
 func (c *conn) Close() error {
 	c.closeMuxOnce.Do(func() { close(c.closeMux) })
 	err := c.Err()
-	if err == ErrConnClosed {
+	var connErr *ConnectionError
+	if errors.As(err, &connErr) && connErr.inner == nil {
+		// an empty ConnectionError means the connection was closed by the caller
+		// or as requested by the peer and no error was provided in the close frame.
 		return nil
 	}
 	return err
@@ -388,6 +390,7 @@ func (c *conn) close() {
 	err := c.net.Close()
 	switch {
 	// conn.err already set
+	// TODO: err info is lost, log it?
 	case c.err != nil:
 
 	// conn.err not set and c.net.Close() returned a non-nil error
@@ -396,7 +399,6 @@ func (c *conn) close() {
 
 	// no errors
 	default:
-		c.err = ErrConnClosed
 	}
 
 	// check rxDone after closing net, otherwise may block
@@ -409,7 +411,7 @@ func (c *conn) close() {
 func (c *conn) Err() error {
 	c.errMu.Lock()
 	defer c.errMu.Unlock()
-	return c.err
+	return &ConnectionError{inner: c.err}
 }
 
 // mux is started in it's own goroutine after initial connection establishment.
@@ -456,8 +458,6 @@ func (c *conn) mux() {
 			case *frames.PerformClose:
 				if body.Error != nil {
 					c.err = body.Error
-				} else {
-					c.err = ErrConnClosed
 				}
 				return
 
@@ -481,7 +481,7 @@ func (c *conn) mux() {
 			case *frames.PerformEnd:
 				session, ok = sessionsByRemoteChannel[fr.Channel]
 				if !ok {
-					c.err = fmt.Errorf("%T: didn't find channel %d in sessionsByRemoteChannel", fr.Body, fr.Channel)
+					c.err = fmt.Errorf("%T: didn't find channel %d in sessionsByRemoteChannel (PerformEnd)", fr.Body, fr.Channel)
 					break
 				}
 				// we MUST remove the remote channel from our map as soon as we receive
@@ -657,7 +657,7 @@ func (c *conn) connReader() {
 		// parse the frame
 		b, ok := buf.Next(bodySize)
 		if !ok {
-			c.connErr <- io.EOF
+			c.connErr <- fmt.Errorf("buffer EOF; requested bytes: %d, actual size: %d", bodySize, buf.Len())
 			return
 		}
 
@@ -703,6 +703,7 @@ func (c *conn) connWriter() {
 	var err error
 	for {
 		if err != nil {
+			debug(1, "connWriter error: %v", err)
 			c.connErr <- err
 			return
 		}
@@ -868,7 +869,7 @@ func (c *conn) readProtoHeader() (protoHeader, error) {
 	case fr := <-c.rxFrame:
 		return p, fmt.Errorf("readProtoHeader: unexpected frame %#v", fr)
 	case <-deadline:
-		return p, ErrTimeout
+		return p, errors.New("amqp: timeout waiting for response")
 	}
 }
 
@@ -1037,7 +1038,7 @@ func (c *conn) readFrame() (frames.Frame, error) {
 	case p := <-c.rxProto:
 		return fr, fmt.Errorf("unexpected protocol header %#v", p)
 	case <-deadline:
-		return fr, ErrTimeout
+		return fr, errors.New("amqp: timeout waiting for response")
 	}
 }
 

sdk/messaging/azservicebus/internal/go-amqp/errors.go

@@ -63,13 +63,6 @@ func (e *DetachError) Error() string {
 
 // Errors
 var (
-	ErrTimeout = errors.New("amqp: timeout waiting for response")
-
-	// ErrConnClosed is propagated to Session and Senders/Receivers
-	// when Client.Close() is called or the server closes the connection
-	// without specifying an error.
-	ErrConnClosed = errors.New("amqp: connection closed")
-
 	// ErrSessionClosed is propagated to Sender/Receivers
 	// when Session.Close() is called.
 	ErrSessionClosed = errors.New("amqp: session closed")
@@ -79,6 +72,19 @@ var (
 	ErrLinkClosed = errors.New("amqp: link closed")
 )
 
+// ConnectionError is propagated to Session and Senders/Receivers
+// when the connection has been closed or is no longer functional.
+type ConnectionError struct {
+	inner error
+}
+
+func (c *ConnectionError) Error() string {
+	if c.inner == nil {
+		return "amqp: connection closed"
+	}
+	return c.inner.Error()
+}
+
 // Default link options
 const (
 	DefaultLinkCredit      = 1

sdk/messaging/azservicebus/internal/go-amqp/internal/frames/frames.go

@@ -1353,7 +1353,7 @@ func (si *SASLInit) frameBody() {}
 func (si *SASLInit) Marshal(wr *buffer.Buffer) error {
 	return encoding.MarshalComposite(wr, encoding.TypeCodeSASLInit, []encoding.MarshalField{
 		{Value: &si.Mechanism, Omit: false},
-		{Value: &si.InitialResponse, Omit: len(si.InitialResponse) == 0},
+		{Value: &si.InitialResponse, Omit: false},
 		{Value: &si.Hostname, Omit: len(si.Hostname) == 0},
 	})
 }

sdk/messaging/azservicebus/internal/go-amqp/link.go

@@ -237,7 +237,9 @@ func attachLink(s *Session, r *Receiver, opts []LinkOption) (*link, error) {
 		l.Messages = make(chan Message, l.receiver.maxCredit)
 		l.unsettledMessages = map[string]struct{}{}
 		// copy the received filter values
-		l.Source.Filter = resp.Source.Filter
+		if resp.Source != nil {
+			l.Source.Filter = resp.Source.Filter
+		}
 	} else {
 		if l.Target == nil {
 			l.Target = new(frames.Target)
@@ -681,7 +683,7 @@ func (l *link) muxHandleFrame(fr frames.FrameBody) error {
 		if !fr.Echo {
 			// if the 'drain' flag has been set in the frame sent to the _receiver_ then
 			// we signal whomever is waiting (the service has seen and acknowledged our drain)
-			if fr.Drain && l.receiver.manualCreditor != nil {
+			if fr.Drain && l.receiver != nil && l.receiver.manualCreditor != nil {
 				l.linkCredit = 0 // we have no active credits at this point.
 				l.receiver.manualCreditor.EndDrain()
 			}

sdk/messaging/azservicebus/internal/go-amqp/link_options.go

@@ -102,6 +102,24 @@ func LinkTargetAddress(addr string) LinkOption {
 	}
 }
 
+// LinkTargetCapabilities sets the target capabilities.
+func LinkTargetCapabilities(capabilities ...string) LinkOption {
+	return func(l *link) error {
+		if l.Target == nil {
+			l.Target = new(frames.Target)
+		}
+
+		// Convert string to symbol
+		symbolCapabilities := make([]encoding.Symbol, len(capabilities))
+		for i, v := range capabilities {
+			symbolCapabilities[i] = encoding.Symbol(v)
+		}
+
+		l.Target.Capabilities = append(l.Target.Capabilities, symbolCapabilities...)
+		return nil
+	}
+}
+
 // LinkAddressDynamic requests a dynamically created address from the server.
 func LinkAddressDynamic() LinkOption {
 	return func(l *link) error {

sdk/messaging/azservicebus/internal/go-amqp/receiver.go

@@ -62,7 +62,7 @@ func (r *Receiver) Prefetched(ctx context.Context) (*Message, error) {
 	case msg := <-r.link.Messages:
 		debug(3, "Receive() non blocking %d", msg.deliveryID)
 		msg.link = r.link
-		return acceptIfModeFirst(ctx, r, &msg)
+		return &msg, nil
 	case <-ctx.Done():
 		return nil, ctx.Err()
 	default:
@@ -89,27 +89,14 @@ func (r *Receiver) Receive(ctx context.Context) (*Message, error) {
 	case msg := <-r.link.Messages:
 		debug(3, "Receive() blocking %d", msg.deliveryID)
 		msg.link = r.link
-		return acceptIfModeFirst(ctx, r, &msg)
+		return &msg, nil
 	case <-r.link.Detached:
 		return nil, r.link.err
 	case <-ctx.Done():
 		return nil, ctx.Err()
 	}
 }
 
-// acceptIfModeFirst auto-accepts a message if we are in mode first, otherwise it no-ops.
-func acceptIfModeFirst(ctx context.Context, r *Receiver, msg *Message) (*Message, error) {
-	// for ModeFirst, auto-accept the message
-	if receiverSettleModeValue(r.link.ReceiverSettleMode) == ModeSecond {
-		return msg, nil
-	}
-	if err := r.AcceptMessage(ctx, msg); err != nil {
-		return nil, err
-	}
-	return msg, nil
-
-}
-
 // Accept notifies the server that the message has been
 // accepted and does not require redelivery.
 func (r *Receiver) AcceptMessage(ctx context.Context, msg *Message) error {

sdk/messaging/azservicebus/receiver_unit_test.go

@@ -438,13 +438,13 @@ func TestReceiver_UserFacingErrors(t *testing.T) {
 	require.ErrorAs(t, err, &asSBError)
 	require.Equal(t, CodeConnectionLost, asSBError.Code)
 
-	fakeAMQPLinks.Err = amqp.ErrConnClosed
+	fakeAMQPLinks.Err = &amqp.ConnectionError{}
 	messages, err = receiver.ReceiveDeferredMessages(context.Background(), []int64{1}, nil)
 	require.Empty(t, messages)
 	require.ErrorAs(t, err, &asSBError)
 	require.Equal(t, CodeConnectionLost, asSBError.Code)
 
-	fakeAMQPLinks.Err = amqp.ErrConnClosed
+	fakeAMQPLinks.Err = &amqp.ConnectionError{}
 	messages, err = receiver.ReceiveMessages(context.Background(), 1, nil)
 	require.Empty(t, messages)
 	require.ErrorAs(t, err, &asSBError)