chore(ci): update CI

Signed-off-by: Valery Piashchynski <[email protected]>

.github/workflows/linters.yml

@@ -18,6 +18,6 @@ jobs:
       - name: Run linter
         uses: golangci/[email protected] # Action page: <https://github.com/golangci/golangci-lint-action>
         with:
-          version: v1.54 # without patch version
+          version: v1.59 # without patch version
           only-new-issues: false # show only new issues if it's a pull request
           args: --timeout=10m --build-tags=race ./...

.github/workflows/linux.yml

@@ -17,9 +17,9 @@ jobs:
     timeout-minutes: 60
     strategy:
       matrix:
-        php: [ "8.2" ]
-        go: [ stable ]
-        os: [ "ubuntu-latest" ]
+        php: ["8.3"]
+        go: [stable]
+        os: ["ubuntu-latest"]
     steps:
       - name: Set up Go ${{ matrix.go }}
         uses: actions/setup-go@v5 # action page: <https://github.com/actions/setup-go>
@@ -71,13 +71,13 @@ jobs:
           sudo cp mkcert-v*-linux-amd64 /usr/local/bin/mkcert
           mkcert -install
           mkcert localhost 127.0.0.1 ::1
-          mkcert -client localhost 127.0.0.1 ::1 
+          mkcert -client localhost 127.0.0.1 ::1
           cp -r localhost+2-client-key.pem localhost+2-client.pem localhost+2-key.pem localhost+2.pem test-certs/
           cp -r $(mkcert -CAROOT)/rootCA.pem test-certs/
 
           docker compose -f env/docker-compose-amqp.yaml up -d
           sleep 30
-          
+
           mkdir ./coverage-ci
           go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=$(cat pkgs.txt) -coverprofile=./coverage-ci/amqp.out -covermode=atomic jobs_amqp_test.go
           docker compose -f env/docker-compose-amqp.yaml down

.github/workflows/linux_durability.yml

@@ -17,9 +17,9 @@ jobs:
     timeout-minutes: 60
     strategy:
       matrix:
-        php: [ "8.2" ]
-        go: [ stable ]
-        os: [ "ubuntu-latest" ]
+        php: ["8.3"]
+        go: [stable]
+        os: ["ubuntu-latest"]
     steps:
       - name: Set up Go ${{ matrix.go }}
         uses: actions/setup-go@v5 # action page: <https://github.com/actions/setup-go>
@@ -64,7 +64,7 @@ jobs:
           cd tests
           docker compose -f env/docker-compose-amqp.yaml up -d
           sleep 30
-          
+
           mkdir ./coverage-ci
           go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=$(cat pkgs.txt) -coverprofile=./coverage-ci/amqp.out -covermode=atomic jobs_amqp_durability_test.go
           docker compose -f env/docker-compose-amqp.yaml down

.golangci.yml

@@ -2,21 +2,11 @@
 
 run:
   timeout: 1m
-  skip-dirs:
-    - .github
-    - .git
   allow-parallel-runners: true
 
-output:
-  format: colored-line-number # colored-line-number|line-number|json|tab|checkstyle|code-climate
-
 linters-settings:
   wsl:
     allow-assign-and-anything: true
-  govet:
-    check-shadowing: true
-  golint:
-    min-confidence: 0.1
   godot:
     scope: declarations
     capital: true
@@ -34,7 +24,6 @@ linters-settings:
     range-loops: true
     for-loops: true
   nolintlint:
-    allow-leading-space: false
     require-specific: true
 
 linters: # All available linters list: <https://golangci-lint.run/usage/linters/>
@@ -75,6 +64,9 @@ linters: # All available linters list: <https://golangci-lint.run/usage/linters/
     - whitespace # Tool for detection of leading and trailing whitespace
 
 issues:
+  exclude-dirs:
+    - .github
+    - .git
   exclude-rules:
     - path: _test\.go
       linters:

amqpjobs/driver.go

@@ -13,7 +13,7 @@ import (
 	"github.com/goccy/go-json"
 	"github.com/google/uuid"
 	amqp "github.com/rabbitmq/amqp091-go"
-	"github.com/roadrunner-server/api/v4/plugins/v3/jobs"
+	"github.com/roadrunner-server/api/v4/plugins/v4/jobs"
 	"github.com/roadrunner-server/errors"
 	jprop "go.opentelemetry.io/contrib/propagators/jaeger"
 	"go.opentelemetry.io/otel"

amqpjobs/item.go

@@ -3,13 +3,14 @@ package amqpjobs
 import (
 	"context"
 	stderr "errors"
+	"maps"
 	"sync/atomic"
 	"time"
 
 	"github.com/goccy/go-json"
 	"github.com/google/uuid"
 	amqp "github.com/rabbitmq/amqp091-go"
-	"github.com/roadrunner-server/api/v4/plugins/v3/jobs"
+	"github.com/roadrunner-server/api/v4/plugins/v4/jobs"
 	"github.com/roadrunner-server/errors"
 	"go.uber.org/zap"
 )
@@ -41,7 +42,7 @@ type Options struct {
 	// Pipeline manually specified pipeline.
 	Pipeline string `json:"pipeline,omitempty"`
 	// Delay defines time duration to delay execution for. Defaults to none.
-	Delay int64 `json:"delay,omitempty"`
+	Delay int `json:"delay,omitempty"`
 	// AutoAck option
 	AutoAck bool `json:"auto_ack"`
 	// AMQP Queue
@@ -131,6 +132,19 @@ func (i *Item) Ack() error {
 	return i.Options.ack(i.Options.multipleAck)
 }
 
+func (i *Item) NackWithOptions(requeue bool, delay int) error {
+	if atomic.LoadUint64(i.Options.stopped) == 1 {
+		return errors.Str("failed to acknowledge the JOB, the pipeline is probably stopped")
+	}
+
+	// there is no native mechanism to delay requeue in AMQP
+	if i.Options.Delay > 0 {
+		return i.Requeue(nil, delay)
+	}
+
+	return i.Options.nack(true, requeue)
+}
+
 func (i *Item) Nack() error {
 	if atomic.LoadUint64(i.Options.stopped) == 1 {
 		return errors.Str("failed to acknowledge the JOB, the pipeline is probably stopped")
@@ -142,7 +156,7 @@ func (i *Item) Nack() error {
 }
 
 // Requeue with the provided delay, handled by the Nack
-func (i *Item) Requeue(headers map[string][]string, delay int64) error {
+func (i *Item) Requeue(headers map[string][]string, delay int) error {
 	if atomic.LoadUint64(i.Options.stopped) == 1 {
 		return errors.Str("failed to acknowledge the JOB, the pipeline is probably stopped")
 	}
@@ -152,7 +166,7 @@ func (i *Item) Requeue(headers map[string][]string, delay int64) error {
 
 	// overwrite the delay
 	i.Options.Delay = delay
-	i.headers = headers
+	maps.Copy(i.headers, headers)
 
 	err := i.Options.requeueFn(context.Background(), i)
 	if err != nil {
@@ -164,7 +178,7 @@ func (i *Item) Requeue(headers map[string][]string, delay int64) error {
 		return err
 	}
 
-	// ack the job
+	// ack the previous message to avoid duplicates
 	err = i.Options.ack(false)
 	if err != nil {
 		return err
@@ -215,7 +229,7 @@ func fromJob(job jobs.Message) *Item {
 		Options: &Options{
 			Priority: job.Priority(),
 			Pipeline: job.GroupID(),
-			Delay:    job.Delay(),
+			Delay:    int(job.Delay()),
 			AutoAck:  job.AutoAck(),
 		},
 	}
@@ -280,7 +294,7 @@ func (d *Driver) unpack(deliv amqp.Delivery) *Item {
 	if t, ok := deliv.Headers[jobs.RRDelay]; ok {
 		switch t.(type) {
 		case int, int16, int32, int64:
-			item.Options.Delay = t.(int64)
+			item.Options.Delay = t.(int)
 		default:
 			d.log.Warn("unknown delay type", zap.Strings("want", []string{"int, int16, int32, int64"}), zap.Any("actual", t))
 		}

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/goccy/go-json v0.10.3
 	github.com/google/uuid v1.6.0
 	github.com/rabbitmq/amqp091-go v1.10.0
-	github.com/roadrunner-server/api/v4 v4.12.0
+	github.com/roadrunner-server/api/v4 v4.13.0
 	github.com/roadrunner-server/endure/v2 v2.4.5
 	github.com/roadrunner-server/errors v1.4.0
 	github.com/stretchr/testify v1.9.0

go.sum

@@ -26,8 +26,8 @@ github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRI
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw=
 github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
-github.com/roadrunner-server/api/v4 v4.12.0 h1:N8AC+b7uzrDpTPnFTBVWNIs9ZMV42hwKDCo29X84iS8=
-github.com/roadrunner-server/api/v4 v4.12.0/go.mod h1:nLV2f4O7tDh5DaMDff4oX1bNJ9erz7eyq+4TajgKGck=
+github.com/roadrunner-server/api/v4 v4.13.0 h1:ciC5nzkIHqfbnzFgaUb1If/+fKLbL39OVqUdnEKrXdE=
+github.com/roadrunner-server/api/v4 v4.13.0/go.mod h1:J8ZzDCNFs3BQDql793eBeshpjNJxLrCpEMooGzapTLo=
 github.com/roadrunner-server/endure/v2 v2.4.5 h1:GoZm/1HjKCKm8TpaP/Pm2KbN0X9gLyN840cA3Fn/TCE=
 github.com/roadrunner-server/endure/v2 v2.4.5/go.mod h1:83UvLdt+RNxELTSna+SZMWQiu+Thj6wOz6hmlp65XFI=
 github.com/roadrunner-server/errors v1.4.0 h1:Odjg3VZrj1q5Y8ILwoN+JgERyv0pkhrWPNOM4h68iQ8=