Redis production hardening

README.md

@@ -25,6 +25,9 @@ correctness model, then the network/protocol layer, and only later transparent
 Redis proxying and persistence. The current repo is already useful as a compact
 Go reference for reliable queue internals.
 
+Moxy is `v0.1.0-alpha` software. It is useful for learning, experiments, and
+small reliability-model prototypes, but it is not production-hardened yet.
+
 ## The Pain
 
 Plain Redis list consumption often starts with `LPOP`. It is fast, simple, and
@@ -115,6 +118,8 @@ to engines. Queue backends own task storage; the core engine owns lease metadata
 expiration scheduling.
 
 See [ARCHITECTURE.md](ARCHITECTURE.md) for the deeper system notes.
+See [docs/redis-production-caveats.md](docs/redis-production-caveats.md) before
+using the Redis backend for anything beyond experimentation.
 
 ## Internal Commands
 
@@ -167,6 +172,7 @@ It is deterministic, easy to test, and useful for validating lease behavior.
 
 - `moxy:{queue}:ready`
 - `moxy:{queue}:processing`
+- `moxy:{queue}:dead`
 
 `Acquire` uses `LMOVE ready processing RIGHT LEFT`. `Complete` and `Requeue` use Lua
 scripts so finding a task by ID and removing or moving it happens atomically inside
@@ -267,7 +273,7 @@ implemented yet:
 - crash recovery
 - Redis Streams
 - distributed coordination
-- user-facing network protocol commands
+- additional Redis-compatible commands beyond `PING` and `MOXY.*`
 
 ## Roadmap
 

docs/adr/0001-redis-lists-vs-streams.md

@@ -0,0 +1,46 @@
+# ADR 0001: Redis Lists vs Streams
+
+## Status
+
+Accepted for the current alpha backend.
+
+## Context
+
+Moxy's first Redis backend uses Redis lists plus `LMOVE` and small Lua scripts.
+The backend models tasks moving through explicit lifecycle storage:
+
+```text
+READY -> PROCESSING -> ACK/REQUEUE
+```
+
+Redis Streams are also a valid Redis-native way to model reliable work queues.
+
+## Decision
+
+Keep the current Redis backend on lists for now:
+
+- Ready tasks live in a Redis list.
+- Processing tasks live in a Redis list.
+- `LMOVE` moves one task from ready to processing.
+- Lua scripts perform bounded atomic transitions for ACK, requeue, and
+  dead-letter moves.
+
+## Why Lists First
+
+Lists are the simplest explicit baseline. They make `READY -> PROCESSING ->
+ACK/REQUEUE` easy to reason about, map cleanly to the current `queue.Backend`
+abstraction, and are a good educational first backend for Moxy's lease model.
+
+## Streams Alternative
+
+Redis Streams provide consumer groups, pending entries, `XACK`, `XAUTOCLAIM`,
+and built-in tools for inspecting and reclaiming stuck messages.
+
+Streams are more Redis-native for reliable work queues and may be a better fit
+for production-oriented deployments.
+
+## Consequences
+
+The lists backend stays small and understandable, but Moxy must own more queue
+semantics in code and Lua. A future `RedisStreamsQueue` backend can be added
+without changing `core.Engine` if it satisfies the same backend interface.

docs/redis-production-caveats.md

@@ -0,0 +1,72 @@
+# Redis Production Caveats
+
+Moxy is currently `v0.1.0-alpha`. It is useful for learning, experimentation,
+and validating queue semantics, but it is not production-hardened yet.
+
+## Delivery Semantics
+
+Moxy currently provides at-least-once delivery, not exactly-once delivery. A task
+can be delivered more than once after worker crashes, lease expiration,
+replication failover, or client retry behavior. Workers should be idempotent.
+
+## Redis Persistence
+
+Moxy cannot provide stronger durability than the configured Redis persistence.
+
+- With AOF `appendfsync everysec`, Redis can lose a small recent write window
+  during a crash.
+- With AOF `appendfsync always`, Redis fsyncs every write and is safer, but
+  slower.
+- RDB-only or loosely configured persistence can lose more queue state than a
+  durable queue workload usually expects.
+
+Choose Redis persistence settings based on the durability required by the queue.
+
+## Redis Eviction
+
+Queue Redis instances should not use cache-style eviction policies such as
+`allkeys-lru` for queue data. Evicting queue keys can silently drop ready,
+processing, or dead-lettered work.
+
+Prefer `noeviction` and a persistence-oriented Redis configuration for queue
+workloads.
+
+## Replication And Failover
+
+Redis replication and failover can still produce duplicates or replayed work.
+Writes acknowledged by a primary may not be present on a promoted replica, and
+clients may retry operations around failover boundaries.
+
+Moxy expects workers to treat task handling as idempotent.
+
+## Redis Cluster Key Slots
+
+Redis Cluster requires all keys touched by one atomic operation to hash to the
+same slot. Moxy queue keys use a shared hash tag per logical queue:
+
+```text
+moxy:{<queueName>}:ready
+moxy:{<queueName>}:processing
+moxy:{<queueName>}:dead
+```
+
+The content inside braces must be identical for all keys belonging to the same
+logical queue.
+
+## Lua Scripts
+
+Redis executes Lua scripts atomically and blocks other server activity while a
+script runs. Moxy scripts should remain short and bounded to queue transition
+work such as ACK, requeue, and dead-letter moves.
+
+## Dead-Letter Queues
+
+Dead-letter queue support is being added so tasks that expire too many times can
+move out of the retry loop instead of requeueing forever. This is a baseline
+safety feature, not full crash recovery.
+
+## Streams
+
+Redis Streams are a valid alternative for reliable work queues. Consumer groups,
+pending entries, `XACK`, and reclaim operations may make Streams a better fit for
+some production systems. Moxy may explore a separate Redis Streams backend later.

internal/command/handler_test.go

@@ -3,6 +3,7 @@ package command
 import (
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/an8kk/moxy/internal/core"
 	"github.com/an8kk/moxy/internal/queue"
@@ -138,6 +139,37 @@ func TestStatsReportsQueueState(t *testing.T) {
 	if stats.Stats.Ready != 1 || stats.Stats.Processing != 1 || stats.Stats.ActiveLeases != 1 {
 		t.Fatalf("stats = %+v, want ready=1 processing=1 active=1", stats.Stats)
 	}
+	if stats.Stats.Dead != 0 {
+		t.Fatalf("dead count = %d, want 0", stats.Stats.Dead)
+	}
+}
+
+func TestStatsReportsDeadCount(t *testing.T) {
+	svc := service.NewWithConfig(func(queueName string) queue.Backend {
+		return queue.NewMemoryQueue()
+	}, service.ServiceConfig{
+		Engine: core.EngineConfig{MaxAttempts: 1},
+	})
+	handler := NewHandler(svc)
+
+	if _, err := handler.Handle(Command{Name: "MOXY.ENQUEUE", Args: []string{"jobs", "payload"}}); err != nil {
+		t.Fatalf("enqueue returned error: %v", err)
+	}
+	fetch, err := handler.Handle(Command{Name: "MOXY.FETCH", Args: []string{"jobs", "1"}})
+	if err != nil {
+		t.Fatalf("fetch returned error: %v", err)
+	}
+	if _, err := svc.ReapExpired(fetch.ExpiresAt.Add(time.Nanosecond)); err != nil {
+		t.Fatalf("reap expired returned error: %v", err)
+	}
+
+	stats, err := handler.Handle(Command{Name: "MOXY.STATS", Args: []string{"jobs"}})
+	if err != nil {
+		t.Fatalf("stats returned error: %v", err)
+	}
+	if stats.Stats.Dead != 1 {
+		t.Fatalf("dead count = %d, want 1", stats.Stats.Dead)
+	}
 }
 
 func newTestHandler() *Handler {

internal/core/engine.go

@@ -22,16 +22,24 @@ const requeueRetryDelay = time.Second
 type Stats struct {
 	Ready          int
 	Processing     int
+	Dead           int
 	ActiveLeases   int
 	ExpirationHeap int
 }
 
+// EngineConfig controls lease expiration behavior.
+type EngineConfig struct {
+	MaxAttempts       int
+	RequeueRetryDelay time.Duration
+}
+
 // Engine owns all mutable in-memory queue, lease, and expiration state.
 type Engine struct {
 	mu          sync.Mutex
 	ready       queue.Backend
 	leases      map[string]*Lease
 	expirations expirationHeap
+	config      EngineConfig
 }
 
 // NewEngine creates an empty in-memory lease engine.
@@ -41,17 +49,31 @@ func NewEngine() *Engine {
 
 // NewEngineWithBackend creates a single-queue lease engine over the provided backend.
 func NewEngineWithBackend(backend queue.Backend) *Engine {
-	return newEngine(backend)
+	return newEngine(backend, EngineConfig{})
+}
+
+// NewEngineWithBackendAndConfig creates a lease engine with explicit expiration config.
+func NewEngineWithBackendAndConfig(backend queue.Backend, config EngineConfig) *Engine {
+	return newEngine(backend, config)
 }
 
-func newEngine(backend queue.Backend) *Engine {
+func newEngine(backend queue.Backend, configs ...EngineConfig) *Engine {
+	var config EngineConfig
+	if len(configs) > 0 {
+		config = configs[0]
+	}
+
 	expirations := expirationHeap{}
 	heap.Init(&expirations)
+	if config.RequeueRetryDelay <= 0 {
+		config.RequeueRetryDelay = requeueRetryDelay
+	}
 
 	return &Engine{
 		ready:       backend,
 		leases:      make(map[string]*Lease),
 		expirations: expirations,
+		config:      config,
 	}
 }
 
@@ -141,10 +163,11 @@ func (e *Engine) ReapExpired(now time.Time) (int, error) {
 			continue
 		}
 
-		if err := e.ready.Requeue(lease.Task.ID); err != nil {
+		err := e.expireLease(lease)
+		if err != nil {
 			heap.Push(&e.expirations, expirationItem{
 				LeaseID:   item.LeaseID,
-				ExpiresAt: now.Add(requeueRetryDelay),
+				ExpiresAt: now.Add(e.config.RequeueRetryDelay),
 			})
 			return requeued, err
 		}
@@ -155,6 +178,18 @@ func (e *Engine) ReapExpired(now time.Time) (int, error) {
 	return requeued, nil
 }
 
+func (e *Engine) expireLease(lease *Lease) error {
+	if e.shouldDeadLetter(lease) {
+		return e.ready.DeadLetter(lease.Task.ID, "max attempts exceeded")
+	}
+
+	return e.ready.Requeue(lease.Task.ID)
+}
+
+func (e *Engine) shouldDeadLetter(lease *Lease) bool {
+	return e.config.MaxAttempts > 0 && lease.Task.Attempts+1 >= e.config.MaxAttempts
+}
+
 // Stats returns counts for tests and diagnostics.
 func (e *Engine) Stats() Stats {
 	e.mu.Lock()
@@ -164,6 +199,7 @@ func (e *Engine) Stats() Stats {
 	return Stats{
 		Ready:          queueStats.Ready,
 		Processing:     queueStats.Processing,
+		Dead:           queueStats.Dead,
 		ActiveLeases:   len(e.leases),
 		ExpirationHeap: e.expirations.Len(),
 	}
@@ -184,8 +220,9 @@ func cloneLease(lease *Lease) *Lease {
 
 func cloneTask(task Task) Task {
 	return Task{
-		ID:      task.ID,
-		Payload: cloneBytes(task.Payload),
+		ID:       task.ID,
+		Payload:  cloneBytes(task.Payload),
+		Attempts: task.Attempts,
 	}
 }