seqeralabs · pditommaso · Jul 31, 2026 · Jul 31, 2026
diff --git a/lib-cmd-queue-redis/README.md b/lib-cmd-queue-redis/README.md
@@ -8,17 +8,17 @@ Add this dependency to your `build.gradle`:
 
 ```gradle
 dependencies {
-    implementation 'io.seqera:lib-cmd-queue-redis:0.7.0'
+    implementation 'io.seqera:lib-cmd-queue-redis:0.4.0'
 }
 ```
 
 ## Features
 
 - Fire-and-forget command submission
 - Typed parameters and results with JSON serialization
-- Status transitions: `PENDING` → `PROCESSING` → `SUCCEEDED`/`FAILED`/`CANCELLED`
-- Non-blocking, concurrent handler execution on virtual threads (no per-command timeout)
-- Periodic status checking for async commands via in-process re-polling
+- Status transitions: `SUBMITTED` → `RUNNING` → `SUCCEEDED`/`FAILED`/`CANCELLED`
+- Automatic timeout handling for long-running commands
+- Periodic status checking for async commands
 - Command cancellation support
 - Persistent storage using Redis or in-memory backend
 
@@ -78,15 +78,15 @@ public class AsyncProcessingHandler implements CommandHandler<ProcessingParams,
     public CommandResult<ProcessingResult> execute(Command<ProcessingParams> command) {
         // Start async job
         externalService.startJob(command.id(), command.params());
-        return CommandResult.processing();  // checkStatus() will be called later
+        return CommandResult.running();  // checkStatus() will be called later
     }
 
     @Override
     public CommandResult<ProcessingResult> checkStatus(Command<ProcessingParams> command, CommandState state) {
         var status = externalService.getStatus(command.id());
         if (status.isComplete()) return CommandResult.success(status.getResult());
         if (status.isFailed()) return CommandResult.failure(status.getError());
-        return CommandResult.processing();  // Still processing, check again later
+        return CommandResult.running();  // Still running, check again later
     }
 }
 ```
@@ -121,131 +121,41 @@ commandService.stop();
 ## Metrics (optional)
 
 Since `0.4.0`, `CommandQueue` exposes a second constructor that forwards an optional
-[`QueueMetrics`](https://github.com/seqeralabs/libseqera/tree/master/lib-data-workqueue)
-handle to the underlying `AbstractWorkQueue`. Subclasses that want to publish
-Micrometer metrics construct a `MicrometerQueueMetrics` from a `MeterRegistry` and pass
+[`StreamMetrics`](https://github.com/seqeralabs/libseqera/tree/master/lib-data-stream-redis)
+handle to the underlying `AbstractMessageStream`. Subclasses that want to publish
+Micrometer metrics construct a `MicrometerStreamMetrics` from a `MeterRegistry` and pass
 it through:
 
 ```java
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micronaut.core.annotation.Nullable;
-import io.seqera.data.workqueue.metrics.MicrometerQueueMetrics;
+import io.seqera.data.stream.metrics.MicrometerStreamMetrics;
 
 public class MyCommandQueue extends CommandQueue {
 
     @Inject
-    public MyCommandQueue(WorkQueue<String> target, CommandConfig config, @Nullable MeterRegistry registry) {
-        super(target, config, registry != null
-                ? new MicrometerQueueMetrics(registry, "my-cmd-queue")
+    public MyCommandQueue(MessageStream<String> target, @Nullable MeterRegistry registry) {
+        super(target, registry != null
+                ? new MicrometerStreamMetrics(registry, "my-cmd-queue")
                 : null);
     }
 
     @Override protected String name() { return "my-cmd-queue"; }
+    @Override protected Duration pollInterval() { return Duration.ofSeconds(1); }
 }
 ```
 
 The 1-arg constructor is unchanged: existing subclasses continue to compile and run
-with no metrics. See [`lib-data-workqueue`](../lib-data-workqueue/README.md) for the
-list of published meters (`seqera.workqueue.entries`, `seqera.workqueue.messages`,
-`seqera.workqueue.processing`) and their tags.
-
-## Architecture
-
-Under the hood the module splits a command into two independently stored parts:
-a lightweight **message** that flows through a queue, and the **full state**
-(params, result, status, timings) that lives in a persistent store. The queue is
-just transport; the store is the source of truth.
-
-```
-   submit(command)
-        │  persist PENDING state + enqueue CommandMsg (fire-and-forget)
-        ▼
-  ┌──────────────┐  save() ┌──────────────────────────────────────────────┐
-  │ CommandState │◀────────┤              CommandServiceImpl                │
-  │    store     │  find() │  processCommand(msg): load state, then         │──▶ execute()
-  │ (Redis/mem)  │────────▶│  dispatch the handler to a worker pool         │    checkStatus()
-  │              │         │  (off the dispatcher thread; virtual threads): │
-  └──────────────┘         │    • terminal     → ack (remove from queue)    │
-        ▲                  │    • processing() → keep lease, re-poll after  │
-        │ getState/Result  │                     pollInterval (in-process)  │
-        │                  └───────────────────────┬────────────────────────┘
-        │                       submit(msg)         │  addConsumer(processCommand)
-        │                                           ▼
-        │                     ┌────────────────────────────────────────────┐
-        └─────────────────────│  CommandQueue (Redis work queue / in-mem.)  │
-                              │  = AbstractWorkQueue: dispatcher + worker   │
-                              │  pool + heartbeat lease → exactly one live  │
-                              │  runner per command, no timeout             │
-                              └────────────────────────────────────────────┘
-```
-
-### Components
-
-| Component | Role |
-|-----------|------|
-| `CommandService` | Public facade: `submit`, `getState`, `getResult`, `cancel`, `registerHandler`, `start`/`stop`. |
-| `CommandQueue` | Abstract `AbstractWorkQueue<CommandMsg>` (from `lib-data-workqueue-redis`). Carries only `CommandMsg` (id + type), Moshi-encoded. Backed by a Redis work queue or an in-memory queue. |
-| `CommandStateStore` | Abstract `AbstractStateStore<CommandState>` (from `lib-data-store-state-redis`). Holds the full JSON state with a TTL (default 7 days). Backed by Redis or in-memory. |
-| `CommandHandler<P,R>` | User code: `execute()` runs the work; optional `checkStatus()` polls a long-running/external job. |
-| `CommandState` | Persisted record (params + result via `@JsonTypeInfo`, status, timings). The source of truth. |
-| `CommandMsg` | Minimal queue pointer — just `commandId` + `type`; the payload is looked up from the store on delivery. |
-
-Backend selection is automatic: when a `RedisActivator` bean is present both the
-queue and the store use Redis; otherwise they fall back to in-memory
-implementations (useful for tests and single-node setups).
-
-### Submit path
-
-`submit()` is fire-and-forget: it persists a `PENDING` `CommandState` to the
-store, then enqueues a `CommandMsg`, and returns the command id immediately. No
-handler runs on the caller's thread.
-
-### Processing loop
-
-`start()` supplies the shared Micronaut `BLOCKING` (virtual-thread) executor to the
-queue and registers `processCommand` as the consumer. The queue's dispatcher thread
-never runs a handler itself: it hands each delivered `CommandMsg` to the executor and
-returns immediately, so a slow handler never blocks intake. The consumer returns a
-boolean:
-
-- **`true`** → terminal; the message is acknowledged and removed.
-- **`false`** → not yet terminal; the command **keeps its lease** and `processCommand`
-  is re-invoked in-process after `pollInterval` (the poll loop for long-running commands).
-
-For each delivery, `processCommand` loads the state and decides:
-
-1. **State missing or already terminal** → `true`; nothing to do (another replica finished it, or it was cancelled).
-2. **No handler registered** → mark `FAILED`, `true`.
-3. **State is `PENDING`** → run `handler.execute()`. Terminal result → apply, `true`;
-   `processing()` → mark `PROCESSING`, `false` (re-polled after `pollInterval`).
-4. **State is `PROCESSING`** → run `handler.checkStatus()`. Terminal → `true`;
-   `processing()` → `false` (re-polled again).
-
-A quick command finishes in one delivery; a slow or external one flips to `PROCESSING`
-and is driven to completion by repeated `checkStatus()` calls at `pollInterval`
-cadence. Handler exceptions transition the command to `FAILED` and ack. There is no
-per-command timeout and no per-command lock — the handler runs to completion on a
-virtual thread, and the underlying work queue's per-message lease guarantees a single
-concurrent runner across replicas (see
-[`lib-data-workqueue-redis`](../lib-data-workqueue-redis/README.md)).
-
-### Multi-replica behaviour
-
-The work queue's per-message lease (a heartbeated Redis consumer-group entry) ensures a
-command is processed by exactly one live replica at a time; if that replica dies, the
-lease lapses and a peer reclaims the command. The store is the shared source of truth,
-so the terminal-state check keeps processing idempotent. Delivery is **at-least-once**,
-so `execute()`/`checkStatus()` should be idempotent.
+with no metrics. See [`lib-data-stream-redis`](../lib-data-stream-redis/README.md) for the
+list of published meters (`seqera.stream.entries`, `seqera.stream.messages`,
+`seqera.stream.processing`) and their tags.
 
 ## Command Status Flow
 
 ```
-submit() ──▶ PENDING ──pickup──▶ PROCESSING ─┬─success──▶ SUCCEEDED
-                                             ├─error────▶ FAILED
-                                             └─cancel───▶ CANCELLED
-
-(new state persists as PENDING/PROCESSING; legacy SUBMITTED/RUNNING entries still decode.
-Upgrading across this rename requires a zero-overlap rollout — see changelog 0.7.0.)
+submit() ──▶ SUBMITTED ──pickup──▶ RUNNING ─┬─success──▶ SUCCEEDED
+                                            ├─error────▶ FAILED
+                                            └─cancel───▶ CANCELLED
 ```
 
 ## Testing

diff --git a/lib-cmd-queue-redis/VERSION b/lib-cmd-queue-redis/VERSION
@@ -1 +1 @@
-0.7.0
+0.4.0
diff --git a/lib-cmd-queue-redis/build.gradle b/lib-cmd-queue-redis/build.gradle
@@ -32,10 +32,8 @@ dependencies {
     // JSON serialization
     implementation project(':lib-serde-jackson')
 
-    // Work queue - 'api' because CommandQueue publicly extends AbstractWorkQueue and exposes
-    // WorkQueue / MessageConsumer from this module in its public API (consumers subclass it).
-    api project(':lib-data-workqueue')
-    implementation project(':lib-data-workqueue-redis')
+    // Message stream
+    implementation project(':lib-data-stream-redis')
     implementation project(':lib-serde-moshi')
 
     // State store

diff --git a/lib-cmd-queue-redis/changelog.txt b/lib-cmd-queue-redis/changelog.txt
@@ -1,76 +1,19 @@
 # lib-cmd-queue-redis changelog
 
-0.7.0 - 17 Jul 2026
-- Adopt the lib-data-workqueue / lib-data-workqueue-redis modules; no longer depends on
-  lib-data-stream-redis (which remains available as a standalone module).
-- MessageStream -> WorkQueue, AbstractMessageStream -> AbstractWorkQueue, StreamMetrics ->
-  QueueMetrics, NoopStreamMetrics -> NoopQueueMetrics; package io.seqera.data.stream.* ->
-  io.seqera.data.workqueue.* (MessageConsumer name unchanged, new package). No behavioural change.
-- Add CommandConfig.concurrency() (default 1000) to make the per-instance in-flight ceiling
-  configurable.
-- CommandQueue now takes a CommandConfig in its constructor and reads pollInterval() /
-  concurrency() from it directly; subclasses only implement name() (no more per-subclass
-  pollInterval/concurrency wiring). Constructor change: (target, config[, metrics]).
-- Rename CommandStatus.SUBMITTED -> PENDING and RUNNING -> PROCESSING (the latter to avoid
-  confusion with a downstream task's own "running" state). The dead, never-persisted PENDING
-  value is removed.
-- Wire compatibility: new state is serialized with the new names; legacy SUBMITTED/RUNNING
-  entries still decode via @JsonAlias, so command state persisted by earlier versions is read
-  correctly.
-- MIGRATION: upgrade with a ZERO-OVERLAP deploy — old and new replicas must never run
-  against the same Redis at once. Scaling to replicas=1 is NOT sufficient: a default k8s
-  rolling update still surges to 2 (maxSurge=1), so old and new briefly coexist. An old
-  replica cannot deserialize the new PROCESSING (hard failure) and decodes the new PENDING
-  to the removed dead constant (retries forever). Use a Recreate strategy (or maxSurge=0 /
-  maxUnavailable=1) so the old pod is gone before the new one starts.
-- ROLLBACK WARNING: new state persists with a 7-day TTL. Rolling back to <=0.6.0 while any
-  PENDING/PROCESSING entry is still live re-triggers the same break on the old code — wait out
-  the TTL (or flush the affected keys) before downgrading.
-- CommandResult.running() is deprecated in favour of CommandResult.processing() (identical
-  behaviour); the switch on command status now fails loud on any unexpected value.
-- Source-breaking for downstream references to CommandStatus.SUBMITTED/RUNNING (e.g. sched):
-  update to PENDING/PROCESSING. The running() helper remains (deprecated) so handler code compiles.
-- Requires lib-data-workqueue 1.0.0+ / lib-data-workqueue-redis 1.0.0+
-
-0.6.0 - 16 Jul 2026
-- Add error-tracking fields to CommandState for observability of a retry storm on a command that
-  stays retryable (i.e. errors that do not terminally fail it, since a thrown handler is now
-  retried — see 0.5.1): errorsCount (count of consecutive processing errors since the last
-  successful processing) and modifiedAt (last-write timestamp). The existing `error` field now
-  also holds the message of a transient (non-terminal) processing error — it carries the most
-  recent error message, transient or terminal; the terminal failure is identified by
-  status == FAILED, not by error being non-null. CommandServiceImpl increments the count /
-  records the message on each caught handler exception (best-effort — a failed record never
-  changes control flow), and resets the streak on any successful transition or recovery.
-  Backward-compatible: the new fields default to 0/null when an older serialized CommandState
-  is read.
-
-0.5.1 - 16 Jul 2026
-- Fix: a handler that throws is no longer treated as a terminal command failure. The catch in
-  CommandServiceImpl.processCommandWithHandler now returns false — keeping the message leased so
-  the stream layer re-polls it — instead of persisting FAILED and acking (which removed the
-  message from the queue). A transient/infra error (e.g. the Postgres connection pool closing
-  during pod shutdown) previously became a permanent FAILED command while the domain entity was
-  left non-terminal, stranding the work with an empty queue and no retry (seqeralabs/sched#712).
-  A genuine command failure must be signalled by returning a FAILED CommandResult, never by
-  throwing; deciding a command has *permanently* failed is delegated to the domain layer that
-  owns the entity state.
-
-0.5.0 - 11 Jul 2026
-- BREAKING (API): CommandConfig.executeTimeout() is removed — the per-command execute-timeout
-  mechanism no longer exists (handlers run asynchronously, so nothing blocks the loop).
-  Downstream CommandConfig implementations that @Override executeTimeout() must drop the
-  override, and any `<prefix>.command-queue.execute-timeout` config key becomes inert.
-- BREAKING (behaviour): commands are processed asynchronously on a shared worker pool, off the
-  dispatcher thread, with at-least-once delivery via the underlying per-message lease.
-  Handlers (execute()/checkStatus()) MUST be idempotent and thread-safe — they can be
-  re-invoked and, across replicas, may run at-least-once. CommandServiceImpl injects
-  @Named(TaskExecutors.BLOCKING) ExecutorService (provided by Micronaut by default).
-- Remove executeWithTimeout / the BLOCKING future.get(timeout) crutch and the per-command
-  distributed lock: cross-replica single-runner is provided by the message-level lease.
-- Adopt the shared async worker-pool model (lease-based dispatch + heartbeat + Model B
-  re-poll); CommandQueue sets concurrency() to bound in-flight commands.
-- Requires lib-data-stream-redis 2.0.0+
+REVERTED - 0.5.0, 0.5.1, 0.6.0, 0.7.0 (13 - 18 Jul 2026)
+- These releases have been reverted; this module is back at 0.4.0 and to its
+  lib-data-stream-redis dependency:
+  - 0.5.0 - async, non-blocking processing with heartbeat lease (PR #84, bdf9374)
+  - 0.5.1 - retry command on handler exception instead of terminal-failing it (PR #87, 6c7b171)
+  - 0.6.0 - error tracking on CommandState: errorsCount, error, modifiedAt (PR #89, e54229e)
+  - 0.7.0 - migrate to lib-data-workqueue(-redis); CommandStatus SUBMITTED -> PENDING and
+    RUNNING -> PROCESSING (PR #86, f3f4ac4)
+- DOWNGRADE WARNING: 0.7.0 persists command state with the renamed PENDING/PROCESSING values
+  and a 7-day TTL. This 0.4.0 code cannot decode them (PROCESSING fails deserialization;
+  PENDING was a dead constant here), so if 0.7.0 ever ran against a given Redis, wait out the
+  TTL or flush the affected command-state keys before rolling back onto it.
+- Sources and full changelog of the reverted work - including the unreleased invocation-lease
+  rework (2ecc744) that was never merged: branch archive/workqueue-pre-revert.
 
 0.4.0 - 13 May 2026
 - Add CommandQueue(MessageStream, StreamMetrics) constructor for optional Micrometer instrumentation

diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandConfig.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandConfig.java
@@ -37,20 +37,19 @@ default Duration pollInterval() {
     }
 
     /**
-     * TTL (Time-To-Live) for command state records in the persistent store.
-     * Commands expire and are removed after this duration.
+     * Timeout for synchronous command execution.
+     * If execute() takes longer than this, the command is marked as RUNNING
+     * and checkStatus() will be called on subsequent queue deliveries.
      */
-    default Duration stateTtl() {
-        return Duration.ofDays(7);
+    default Duration executeTimeout() {
+        return Duration.ofSeconds(1);
     }
 
     /**
-     * Maximum number of commands that may be in flight on a single instance at once.
-     * Handlers run on virtual threads, so this is a memory/heartbeat ceiling (the underlying
-     * work queue's in-flight semaphore), not a thread-pool size; commands beyond it wait in
-     * the queue (backpressure). Effective minimum is 1.
+     * TTL (Time-To-Live) for command state records in the persistent store.
+     * Commands expire and are removed after this duration.
      */
-    default int concurrency() {
-        return 1000;
+    default Duration stateTtl() {
+        return Duration.ofDays(7);
     }
 }