Use context.WithoutCancel in azbus (#4)

* Use context.WithoutCancel in azbus

The handcrafted contextWithoutCancel is replaced with correct
standard context.WithoutCancel from Go 1.21.

If processing a message takes longer than the message deadline a
WARNING is emitted with suggested remedial action.

Type aliases in aliases.go have been moved closer to their point
of usage and aliases.go deleted.

Internal logger is indexed with the name of the sender/receiver.

Mock generation Taskfile is added.

AB#8276

.gitignore

@@ -0,0 +1 @@
+.task/

Taskfile.yml

@@ -18,6 +18,8 @@ version: '3'
 includes:
   codeqa:
     taskfile: ./taskfiles/Taskfile_codeqa.yml
+  mocks:
+    taskfile: ./taskfiles/Taskfile_mocks.yml
 
 tasks:
 
@@ -37,4 +39,4 @@ tasks:
   test:
     desc: run the tests
     cmds:
-      - task: codeqa:unit-tests
+      - task: codeqa:unit-tests

azbus/correlationid.go

@@ -16,11 +16,3 @@ func ContextFromReceivedMessage(ctx context.Context, message *ReceivedMessage) c
 	}
 	return correlationid.ContextWithCorrelationID(ctx, cid.(string))
 }
-
-func AddCorrelationIDOption(ctx context.Context, opts ...OutMessageOption) []OutMessageOption {
-	correlationID := correlationid.FromContext(ctx)
-	if correlationID == "" {
-		return opts
-	}
-	return append(opts, WithProperty(correlationid.CorrelationIDKey, correlationID))
-}

azbus/error.go

@@ -6,9 +6,6 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
 )
 
-// alias so we dont have to import azservicebus elsewhere
-type ServicebusCode = azservicebus.Code
-
 // Azure package expects the user to elucidate errors like so:
 //
 //	    var servicebusError *azservicebus.Error
@@ -30,13 +27,12 @@ var (
 	ErrTimeout            = errors.New("timeout")
 )
 
-// CodeUnauthorizedAccess is in the latest code but is not in the current version used.
 func NewAzbusError(err error) error {
 	var servicebusError *azservicebus.Error
 	if errors.As(err, &servicebusError) {
 		switch servicebusError.Code {
-		// case azservicebus.CodeUnauthorizedAccess:
-		// 	return ErrUnauthorizedAccess
+		case azservicebus.CodeUnauthorizedAccess:
+			return ErrUnauthorizedAccess
 		case azservicebus.CodeConnectionLost:
 			return ErrConnectionLost
 		case azservicebus.CodeLockLost:

azbus/logger.go

@@ -0,0 +1,7 @@
+package azbus
+
+import (
+	"github.com/rkvst/go-rkvstcommon/logger"
+)
+
+type Logger = logger.Logger

azbus/message.go

@@ -2,6 +2,8 @@ package azbus
 
 import (
 	"context"
+	"errors"
+	"time"
 )
 
 // Set a timeout for processing the message, this should be no later than
@@ -27,18 +29,24 @@ import (
 // If it does timeout then it is too late anyway as the peeklock will already be released.
 //
 // for the time being we impose a timeout as it is safe.
-func setTimeout(ctx context.Context, log Logger, msg *ReceivedMessage) (context.Context, context.CancelFunc) {
+var (
+	ErrPeekLockTimeout = errors.New("peeklock deadline reached")
+)
+
+func setTimeout(ctx context.Context, log Logger, msg *ReceivedMessage) (context.Context, context.CancelFunc, time.Duration) {
 
 	var cancel context.CancelFunc
 
-	msgLockedUntil := msg.LockedUntil
-	if msgLockedUntil != nil {
-		ctx, cancel = context.WithDeadline(ctx, *msgLockedUntil)
-		log.Debugf("context deadline from message lock deadline: %v", ctx)
-		return ctx, cancel
+	log.Debugf("msg locked until %s", msg.LockedUntil)
+	if msg.LockedUntil != nil {
+		msgLockedUntil := *msg.LockedUntil
+		ctx, cancel = context.WithDeadlineCause(ctx, msgLockedUntil, ErrPeekLockTimeout)
+		maxDuration := msgLockedUntil.Sub(time.Now())
+		log.Debugf("msg must be processed in %s", maxDuration)
+		return ctx, cancel, maxDuration
 	}
 
-	ctx, cancel = context.WithTimeout(ctx, RenewalTime)
+	ctx, cancel = context.WithTimeoutCause(ctx, RenewalTime, ErrPeekLockTimeout)
 	log.Infof("could not get lock deadline from message, using fixed timeout %v", ctx)
-	return ctx, cancel
+	return ctx, cancel, RenewalTime
 }