Pass at least one key to script executions

bench/worker_bench_test.go

@@ -34,10 +34,10 @@ func BenchmarkWorkerRunJob(b *testing.B) {
 			b.StopTimer()
 
 			for n := 0; n < b.N; n++ {
-				require.NoError(b, redistest.Reset(client))
+				require.NoError(b, redistest.Reset(client, "{ns-bench}"))
 
 				wp := work.NewWorkerPoolWithOptions(
-					struct{}{}, 1, "{ns1}", pool,
+					struct{}{}, 1, "{ns-bench}", pool,
 					work.WorkerPoolOptions{
 						SleepBackoffs: []int64{1000},
 					},
@@ -49,7 +49,7 @@ func BenchmarkWorkerRunJob(b *testing.B) {
 					return nil
 				})
 
-				enqueuer := work.NewEnqueuer("{ns1}", pool)
+				enqueuer := work.NewEnqueuer("{ns-bench}", pool)
 				for i := 0; i < k; i++ {
 					_, err := enqueuer.Enqueue("test", nil)
 					require.NoError(b, err)
@@ -68,11 +68,11 @@ func BenchmarkWorkerRunJob(b *testing.B) {
 			b.StopTimer()
 
 			for n := 0; n < b.N; n++ {
-				require.NoError(b, redistest.Reset(client))
+				require.NoError(b, redistest.Reset(client, "{ns-bench}"))
 
 				queue := work2.NewRedisQueue(client)
 				w := work2.NewWorker(&work2.WorkerOptions{
-					Namespace: "{ns1}",
+					Namespace: "{ns-bench}",
 					Queue:     queue,
 				})
 				var wg sync.WaitGroup
@@ -93,7 +93,7 @@ func BenchmarkWorkerRunJob(b *testing.B) {
 					job := work2.NewJob()
 
 					err := queue.Enqueue(job, &work2.EnqueueOptions{
-						Namespace: "{ns1}",
+						Namespace: "{ns-bench}",
 						QueueID:   "test",
 					})
 					require.NoError(b, err)

http/server_test.go

@@ -16,7 +16,7 @@ import (
 func TestServer(t *testing.T) {
 	client := redistest.NewClient()
 	defer client.Close()
-	require.NoError(t, redistest.Reset(client))
+	require.NoError(t, redistest.Reset(client, "{ns-http}"))
 	q := work.NewRedisQueue(client)
 
 	srv := NewServer(&ServerOptions{
@@ -89,15 +89,15 @@ func TestServer(t *testing.T) {
 		},
 		{
 			reqMethod: "DELETE",
-			reqURL:    "http://example.com/jobs?namespace=%7Bns1%7D&job_id=xxx",
+			reqURL:    "http://example.com/jobs?namespace=%7Bns-http%7D&job_id=xxx",
 			respCode:  200,
-			respBody:  "{\"namespace\":\"{ns1}\",\"queue_id\":\"\",\"job\":{\"ID\":\"xxx\",\"CreatedAt\":\"0001-01-01T00:00:00Z\",\"UpdatedAt\":\"0001-01-01T00:00:00Z\",\"EnqueuedAt\":\"0001-01-01T00:00:00Z\",\"Payload\":null,\"Retries\":0,\"LastError\":\"\"}}\n",
+			respBody:  "{\"namespace\":\"{ns-http}\",\"queue_id\":\"\",\"job\":{\"ID\":\"xxx\",\"CreatedAt\":\"0001-01-01T00:00:00Z\",\"UpdatedAt\":\"0001-01-01T00:00:00Z\",\"EnqueuedAt\":\"0001-01-01T00:00:00Z\",\"Payload\":null,\"Retries\":0,\"LastError\":\"\"}}\n",
 		},
 		{
 			reqMethod: "GET",
-			reqURL:    "http://example.com/jobs?namespace=%7Bns1%7D&job_id=xxx",
+			reqURL:    "http://example.com/jobs?namespace=%7Bns-http%7D&job_id=xxx",
 			respCode:  200,
-			respBody:  "{\"namespace\":\"{ns1}\",\"status\":\"completed\",\"job\":{\"ID\":\"xxx\",\"CreatedAt\":\"0001-01-01T00:00:00Z\",\"UpdatedAt\":\"0001-01-01T00:00:00Z\",\"EnqueuedAt\":\"0001-01-01T00:00:00Z\",\"Payload\":null,\"Retries\":0,\"LastError\":\"\"}}\n",
+			respBody:  "{\"namespace\":\"{ns-http}\",\"status\":\"completed\",\"job\":{\"ID\":\"xxx\",\"CreatedAt\":\"0001-01-01T00:00:00Z\",\"UpdatedAt\":\"0001-01-01T00:00:00Z\",\"EnqueuedAt\":\"0001-01-01T00:00:00Z\",\"Payload\":null,\"Retries\":0,\"LastError\":\"\"}}\n",
 		},
 		{
 			// bad duration
@@ -122,91 +122,91 @@ func TestServer(t *testing.T) {
 			reqMethod: "POST",
 			reqURL:    "http://example.com/jobs",
 			reqBody: `{
-				"namespace": "{ns1}",
+				"namespace": "{ns-http}",
 				"queue_id": "q1",
 				"payload": "payload1",
 				"delay": "10s"
 			}`,
 			respCode: 200,
-			respBody: "{\"namespace\":\"{ns1}\",\"queue_id\":\"q1\",\"job\":{\"ID\":\"[a-z0-9-]{36}\",\"CreatedAt\":\"[TZ0-9:-]+\",\"UpdatedAt\":\"[TZ0-9:-]+\",\"EnqueuedAt\":\"[TZ0-9:-]+\",\"Payload\":\"InBheWxvYWQxIg==\",\"Retries\":0,\"LastError\":\"\"}}",
+			respBody: "{\"namespace\":\"{ns-http}\",\"queue_id\":\"q1\",\"job\":{\"ID\":\"[a-z0-9-]{36}\",\"CreatedAt\":\"[TZ0-9:-]+\",\"UpdatedAt\":\"[TZ0-9:-]+\",\"EnqueuedAt\":\"[TZ0-9:-]+\",\"Payload\":\"InBheWxvYWQxIg==\",\"Retries\":0,\"LastError\":\"\"}}",
 		},
 		{
 			reqMethod: "GET",
-			reqURL:    "http://example.com/metrics?namespace=%7Bns1%7D&queue_id=q1",
+			reqURL:    "http://example.com/metrics?namespace=%7Bns-http%7D&queue_id=q1",
 			respCode:  200,
-			respBody:  "{\"namespace\":\"{ns1}\",\"queue_id\":\"q1\",\"ready_total\":0,\"scheduled_total\":1,\"total\":1,\"latency\":0}\n",
+			respBody:  "{\"namespace\":\"{ns-http}\",\"queue_id\":\"q1\",\"ready_total\":0,\"scheduled_total\":1,\"total\":1,\"latency\":0}\n",
 		},
 		{
 			reqMethod: "POST",
 			reqURL:    "http://example.com/jobs",
 			reqBody: `{
-				"namespace": "{ns1}",
+				"namespace": "{ns-http}",
 				"queue_id": "q1",
 				"payload": "payload1"
 			}`,
 			respCode: 200,
-			respBody: "{\"namespace\":\"{ns1}\",\"queue_id\":\"q1\",\"job\":{\"ID\":\"[a-z0-9-]{36}\",\"CreatedAt\":\"[TZ0-9:-]+\",\"UpdatedAt\":\"[TZ0-9:-]+\",\"EnqueuedAt\":\"[TZ0-9:-]+\",\"Payload\":\"InBheWxvYWQxIg==\",\"Retries\":0,\"LastError\":\"\"}}",
+			respBody: "{\"namespace\":\"{ns-http}\",\"queue_id\":\"q1\",\"job\":{\"ID\":\"[a-z0-9-]{36}\",\"CreatedAt\":\"[TZ0-9:-]+\",\"UpdatedAt\":\"[TZ0-9:-]+\",\"EnqueuedAt\":\"[TZ0-9:-]+\",\"Payload\":\"InBheWxvYWQxIg==\",\"Retries\":0,\"LastError\":\"\"}}",
 		},
 		{
 			reqMethod: "GET",
-			reqURL:    "http://example.com/metrics?namespace=%7Bns1%7D&queue_id=q1",
+			reqURL:    "http://example.com/metrics?namespace=%7Bns-http%7D&queue_id=q1",
 			respCode:  200,
-			respBody:  "{\"namespace\":\"{ns1}\",\"queue_id\":\"q1\",\"ready_total\":1,\"scheduled_total\":1,\"total\":2,\"latency\":[0-9]+}\n",
+			respBody:  "{\"namespace\":\"{ns-http}\",\"queue_id\":\"q1\",\"ready_total\":1,\"scheduled_total\":1,\"total\":2,\"latency\":[0-9]+}\n",
 		},
 		{
 			reqMethod: "POST",
 			reqURL:    "http://example.com/jobs",
 			reqBody: `{
-				"namespace": "{ns1}",
+				"namespace": "{ns-http}",
 				"queue_id": "q1",
 				"id": "id1",
 				"payload": "payload1"
 			}`,
 			respCode: 200,
-			respBody: "{\"namespace\":\"{ns1}\",\"queue_id\":\"q1\",\"job\":{\"ID\":\"id1\",\"CreatedAt\":\"[TZ0-9:-]+\",\"UpdatedAt\":\"[TZ0-9:-]+\",\"EnqueuedAt\":\"[TZ0-9:-]+\",\"Payload\":\"InBheWxvYWQxIg==\",\"Retries\":0,\"LastError\":\"\"}}",
+			respBody: "{\"namespace\":\"{ns-http}\",\"queue_id\":\"q1\",\"job\":{\"ID\":\"id1\",\"CreatedAt\":\"[TZ0-9:-]+\",\"UpdatedAt\":\"[TZ0-9:-]+\",\"EnqueuedAt\":\"[TZ0-9:-]+\",\"Payload\":\"InBheWxvYWQxIg==\",\"Retries\":0,\"LastError\":\"\"}}",
 		},
 		{
 			// same job id
 			reqMethod: "POST",
 			reqURL:    "http://example.com/jobs",
 			reqBody: `{
-				"namespace": "{ns1}",
+				"namespace": "{ns-http}",
 				"queue_id": "q1",
 				"id": "id1",
 				"payload": "payload2"
 			}`,
 			respCode: 200,
-			respBody: "{\"namespace\":\"{ns1}\",\"queue_id\":\"q1\",\"job\":{\"ID\":\"id1\",\"CreatedAt\":\"[TZ0-9:-]+\",\"UpdatedAt\":\"[TZ0-9:-]+\",\"EnqueuedAt\":\"[TZ0-9:-]+\",\"Payload\":\"InBheWxvYWQxIg==\",\"Retries\":0,\"LastError\":\"\"}}",
+			respBody: "{\"namespace\":\"{ns-http}\",\"queue_id\":\"q1\",\"job\":{\"ID\":\"id1\",\"CreatedAt\":\"[TZ0-9:-]+\",\"UpdatedAt\":\"[TZ0-9:-]+\",\"EnqueuedAt\":\"[TZ0-9:-]+\",\"Payload\":\"InBheWxvYWQxIg==\",\"Retries\":0,\"LastError\":\"\"}}",
 		},
 		{
 			reqMethod: "GET",
-			reqURL:    "http://example.com/metrics?namespace=%7Bns1%7D&queue_id=q1",
+			reqURL:    "http://example.com/metrics?namespace=%7Bns-http%7D&queue_id=q1",
 			respCode:  200,
-			respBody:  "{\"namespace\":\"{ns1}\",\"queue_id\":\"q1\",\"ready_total\":2,\"scheduled_total\":1,\"total\":3,\"latency\":[0-9]+}\n",
+			respBody:  "{\"namespace\":\"{ns-http}\",\"queue_id\":\"q1\",\"ready_total\":2,\"scheduled_total\":1,\"total\":3,\"latency\":[0-9]+}\n",
 		},
 		{
 			reqMethod: "GET",
-			reqURL:    "http://example.com/jobs?namespace=%7Bns1%7D&job_id=id1",
+			reqURL:    "http://example.com/jobs?namespace=%7Bns-http%7D&job_id=id1",
 			respCode:  200,
-			respBody:  "{\"namespace\":\"{ns1}\",\"status\":\"ready\",\"job\":{\"ID\":\"id1\",\"CreatedAt\":\"[TZ0-9:-]+\",\"UpdatedAt\":\"[TZ0-9:-]+\",\"EnqueuedAt\":\"[TZ0-9:-]+\",\"Payload\":\"InBheWxvYWQxIg==\",\"Retries\":0,\"LastError\":\"\"}}",
+			respBody:  "{\"namespace\":\"{ns-http}\",\"status\":\"ready\",\"job\":{\"ID\":\"id1\",\"CreatedAt\":\"[TZ0-9:-]+\",\"UpdatedAt\":\"[TZ0-9:-]+\",\"EnqueuedAt\":\"[TZ0-9:-]+\",\"Payload\":\"InBheWxvYWQxIg==\",\"Retries\":0,\"LastError\":\"\"}}",
 		},
 		{
 			reqMethod: "DELETE",
-			reqURL:    "http://example.com/jobs?namespace=%7Bns1%7D&job_id=id1",
+			reqURL:    "http://example.com/jobs?namespace=%7Bns-http%7D&job_id=id1",
 			respCode:  500,
 			respBody:  "{\"error\":\"work: empty queue id\"}\n",
 		},
 		{
 			reqMethod: "DELETE",
-			reqURL:    "http://example.com/jobs?namespace=%7Bns1%7D&queue_id=q1&job_id=id1",
+			reqURL:    "http://example.com/jobs?namespace=%7Bns-http%7D&queue_id=q1&job_id=id1",
 			respCode:  200,
-			respBody:  "{\"namespace\":\"{ns1}\",\"queue_id\":\"q1\",\"job\":{\"ID\":\"id1\",\"CreatedAt\":\"[TZ0-9:-]+\",\"UpdatedAt\":\"[TZ0-9:-]+\",\"EnqueuedAt\":\"[TZ0-9:-]+\",\"Payload\":\"InBheWxvYWQxIg==\",\"Retries\":0,\"LastError\":\"\"}}",
+			respBody:  "{\"namespace\":\"{ns-http}\",\"queue_id\":\"q1\",\"job\":{\"ID\":\"id1\",\"CreatedAt\":\"[TZ0-9:-]+\",\"UpdatedAt\":\"[TZ0-9:-]+\",\"EnqueuedAt\":\"[TZ0-9:-]+\",\"Payload\":\"InBheWxvYWQxIg==\",\"Retries\":0,\"LastError\":\"\"}}",
 		},
 		{
 			reqMethod: "GET",
-			reqURL:    "http://example.com/metrics?namespace=%7Bns1%7D&queue_id=q1",
+			reqURL:    "http://example.com/metrics?namespace=%7Bns-http%7D&queue_id=q1",
 			respCode:  200,
-			respBody:  "{\"namespace\":\"{ns1}\",\"queue_id\":\"q1\",\"ready_total\":1,\"scheduled_total\":1,\"total\":2,\"latency\":[0-9]+}\n",
+			respBody:  "{\"namespace\":\"{ns-http}\",\"queue_id\":\"q1\",\"ready_total\":1,\"scheduled_total\":1,\"total\":2,\"latency\":[0-9]+}\n",
 		},
 	} {
 		var reqBody io.Reader

job.go

@@ -30,6 +30,31 @@ type Job struct {
 	Retries int64 `msgpack:"retries"`
 	// If the job previously fails, LastError will be populated with error string.
 	LastError string `msgpack:"last_error"`
+
+	// AllowPromotion controls whether Enqueue and PromoteJob may lower this
+	// job's score.
+	//
+	// When false (default), both operations use ZADD XX GT semantics: once a
+	// job is scheduled at time T, a subsequent Enqueue with score T' < T is
+	// a no-op, and PromoteJob cannot reduce the score. This preserves the
+	// "deferral" guarantee relied on by deterministic-ID jobs that may be
+	// enqueued multiple times concurrently (a dedup pattern) and prevents
+	// PromoteJob from demoting a job whose score has been bumped to
+	// now + InvisibleSec by Dequeue.
+	//
+	// When true, both operations use ZADD XX (no GT). This lets an
+	// explicit caller lower the score, for example in an opt-in retry flow
+	// where backoff should override Dequeue's InvisibleSec mark, and lets
+	// PromoteJob advance a scheduled-but-pending job to now. The caller is
+	// responsible for ensuring this is safe — typically that the job has a
+	// unique ID and is not relied on for dedup-deferral.
+	//
+	// This field is write-only at the Go API: it is persisted to Redis on
+	// Enqueue and consulted server-side by Enqueue and PromoteJob, but it
+	// is NOT rehydrated onto jobs returned by Dequeue or BulkFind — those
+	// always observe the zero value. PromoteJob reads the persisted value
+	// directly from Redis, so callers do not need to round-trip it.
+	AllowPromotion bool `msgpack:"-" json:",omitempty"`
 }
 
 // InvalidJobPayloadError wraps json or msgpack decoding error.
@@ -203,6 +228,20 @@ type Queue interface {
 	Dequeuer
 }
 
+// Requeuer is an optional interface a Queue may implement to express the
+// framework's retry move ("this job failed, put it back") distinctly from
+// Enqueue ("add a new job"). When a Queue implements Requeuer, the worker's
+// retry path calls Requeue with the computed backoff instead of Enqueue.
+//
+// RedisQueue does not implement Requeuer, so its retry behavior is unchanged
+// (Enqueue with a bumped score). A custom Queue whose Enqueue and "retry a
+// failed in-flight job" are different operations — for example one that keeps
+// waiting and in-flight jobs in separate structures — implements Requeuer so
+// the retry path is unambiguous.
+type Requeuer interface {
+	Requeue(job *Job, backoff time.Duration, opt *EnqueueOptions) error
+}
+
 // BulkEnqueuer enqueues jobs in a batch.
 type BulkEnqueuer interface {
 	BulkEnqueue([]*Job, *EnqueueOptions) error
@@ -240,3 +279,20 @@ func (opt *FindOptions) Validate() error {
 type BulkJobFinder interface {
 	BulkFind(jobIDs []string, opts *FindOptions) ([]*Job, error)
 }
+
+// PromoteOptions specifies how a job is promoted in the queue.
+type PromoteOptions struct {
+	Namespace string
+	QueueID   string
+}
+
+// Validate validates PromoteOptions.
+func (opt *PromoteOptions) Validate() error {
+	if opt.Namespace == "" {
+		return ErrEmptyNamespace
+	}
+	if opt.QueueID == "" {
+		return ErrEmptyQueueID
+	}
+	return nil
+}

middleware/concurrent/dequeuer_test.go

@@ -15,7 +15,7 @@ import (
 func TestDequeuer(t *testing.T) {
 	client := redistest.NewClient()
 	defer client.Close()
-	require.NoError(t, redistest.Reset(client))
+	require.NoError(t, redistest.Reset(client, "{ns-concurrent}"))
 
 	var called int
 	h1 := func(*work.DequeueOptions) (*work.Job, error) {
@@ -38,7 +38,7 @@ func TestDequeuer(t *testing.T) {
 	}
 
 	opt := &work.DequeueOptions{
-		Namespace:    "{ns1}",
+		Namespace:    "{ns-concurrent}",
 		QueueID:      "q1",
 		At:           time.Now(),
 		InvisibleSec: 60,
@@ -75,7 +75,7 @@ func TestDequeuer(t *testing.T) {
 
 	z, err := client.ZRangeByScoreWithScores(
 		context.Background(),
-		"{ns1}:lock:q1",
+		"{ns-concurrent}:lock:q1",
 		&redis.ZRangeBy{
 			Min: "-inf",
 			Max: "+inf",
@@ -87,7 +87,7 @@ func TestDequeuer(t *testing.T) {
 	require.EqualValues(t, opt.At.Unix()+60, z[0].Score)
 	require.EqualValues(t, opt.At.Unix()+60, z[1].Score)
 
-	require.NoError(t, client.ZRem(context.Background(), "{ns1}:lock:q1", "w1").Err())
+	require.NoError(t, client.ZRem(context.Background(), "{ns-concurrent}:lock:q1", "w1").Err())
 	optLater := *opt
 	optLater.At = opt.At.Add(10 * time.Second)
 	// worker 0 is locked already
@@ -105,7 +105,7 @@ func TestDequeuer(t *testing.T) {
 
 	z, err = client.ZRangeByScoreWithScores(
 		context.Background(),
-		"{ns1}:lock:q1",
+		"{ns-concurrent}:lock:q1",
 		&redis.ZRangeBy{
 			Min: "-inf",
 			Max: "+inf",
@@ -137,7 +137,7 @@ func TestDequeuer(t *testing.T) {
 
 	z, err = client.ZRangeByScoreWithScores(
 		context.Background(),
-		"{ns1}:lock:q1",
+		"{ns-concurrent}:lock:q1",
 		&redis.ZRangeBy{
 			Min: "-inf",
 			Max: "+inf",
@@ -155,7 +155,7 @@ func BenchmarkConcurrency(b *testing.B) {
 
 	client := redistest.NewClient()
 	defer client.Close()
-	require.NoError(b, redistest.Reset(client))
+	require.NoError(b, redistest.Reset(client, "{ns-concurrent}"))
 
 	var called int
 	h1 := func(*work.DequeueOptions) (*work.Job, error) {
@@ -178,7 +178,7 @@ func BenchmarkConcurrency(b *testing.B) {
 	}
 
 	opt := &work.DequeueOptions{
-		Namespace:    "{ns1}",
+		Namespace:    "{ns-concurrent}",
 		QueueID:      "q1",
 		At:           time.Now(),
 		InvisibleSec: 60,

middleware/concurrent/local_dequeuer_test.go

@@ -30,7 +30,7 @@ func TestLocalDequeuer(t *testing.T) {
 	}
 
 	opt := &work.DequeueOptions{
-		Namespace:    "{ns1}",
+		Namespace:    "{ns-concurrent}",
 		QueueID:      "q1",
 		At:           time.Now(),
 		InvisibleSec: 60,

middleware/discard/after_test.go

@@ -12,7 +12,7 @@ import (
 func TestAfter(t *testing.T) {
 	job := work.NewJob()
 	opt := &work.DequeueOptions{
-		Namespace: "{ns1}",
+		Namespace: "{ns-discard}",
 		QueueID:   "q1",
 	}
 	d := After(time.Minute)

middleware/discard/invalid_payload_test.go

@@ -10,7 +10,7 @@ import (
 func TestInvalidPayload(t *testing.T) {
 	job := work.NewJob()
 	opt := &work.DequeueOptions{
-		Namespace: "{ns1}",
+		Namespace: "{ns-discard}",
 		QueueID:   "q1",
 	}
 	h := InvalidPayload(func(*work.Job, *work.DequeueOptions) error {

middleware/discard/max_retry_test.go

@@ -11,7 +11,7 @@ import (
 func TestMaxRetry(t *testing.T) {
 	job := work.NewJob()
 	opt := &work.DequeueOptions{
-		Namespace: "{ns1}",
+		Namespace: "{ns-discard}",
 		QueueID:   "q1",
 	}
 	d := MaxRetry(1)

middleware/heartbeat/heartbeater.go

@@ -54,12 +54,13 @@ func Heartbeater(hopts *HeartbeaterOptions) work.HandleMiddleware {
 				}
 			}()
 
-			err := f(job, opt)
-			cancel()
-			<-done
-			job.UpdatedAt = copiedJob.UpdatedAt
-			job.EnqueuedAt = copiedJob.EnqueuedAt
-			return err
+			defer func() {
+				cancel()
+				<-done
+				job.UpdatedAt = copiedJob.UpdatedAt
+				job.EnqueuedAt = copiedJob.EnqueuedAt
+			}()
+			return f(job, opt)
 		}
 	}
 }