From 7f20ac90cefeede0bae7a1c9ebe185f8652b564d Mon Sep 17 00:00:00 2001 From: Paolo Di Tommaso Date: Fri, 31 Jul 2026 09:22:35 +0200 Subject: [PATCH] Revert lib-data-stream-redis to 1.5.0 and lib-cmd-queue-redis to 0.4.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restore both module trees to their state at 8bad0f5 ([release] lib-data-stream-redis@1.5.0, 14 May 2026) — the commit where stream was 1.5.0 and cmd-queue 0.4.0 — undoing everything released on top of them: - stream 2.0.0 / cmd-queue 0.5.0: async, non-blocking consumer processing with heartbeat lease and the poll/renew/ack/release SPI (PR #84, bdf9374), plus doc follow-ups 11fdd3a and 86ae9ac - cmd-queue 0.5.1: retry command on handler exception (PR #87, 6c7b171) - cmd-queue 0.6.0: error tracking on CommandState (PR #89, e54229e, 1f124f0) - cmd-queue 0.7.0: migration to lib-data-workqueue(-redis) and the CommandStatus SUBMITTED->PENDING / RUNNING->PROCESSING rename (PR #86, f3f4ac4); README follow-up PR #91 (e8fe916) - the parts of PR #94 (d38afb0) that touched these two modules' sources cmd-queue therefore depends on lib-data-stream-redis again. lib-data-workqueue and lib-data-workqueue-redis are left in place untouched — they carry the lease-based design forward and were never published. Two build-infra bits from PR #94 are deliberately kept rather than reverted, since they are repo-wide conventions and not module API: the io.seqera.micronaut-library-conventions plugin id (Java 25 target for Micronaut modules) and Groovy 4.0.31 for stream's test dependencies (4.0.24 cannot run on a JDK 25 toolchain). The changelogs keep a REVERTED entry recording the withdrawn versions with their PRs, and cmd-queue's notes the downgrade hazard: 0.7.0-persisted command state uses PENDING/PROCESSING with a 7-day TTL, which this 0.4.0 code cannot decode. Pre-revert state is preserved on branch archive/workqueue-pre-revert (2ecc744), which also carries the unmerged invocation-lease rework. Co-Authored-By: Claude Opus 5 (1M context) --- lib-cmd-queue-redis/README.md | 130 +----- lib-cmd-queue-redis/VERSION | 2 +- lib-cmd-queue-redis/build.gradle | 6 +- lib-cmd-queue-redis/changelog.txt | 85 +--- .../io/seqera/data/command/CommandConfig.java | 19 +- .../seqera/data/command/CommandHandler.java | 16 +- .../io/seqera/data/command/CommandQueue.java | 55 +-- .../io/seqera/data/command/CommandResult.java | 13 +- .../data/command/CommandServiceImpl.java | 141 +++--- .../io/seqera/data/command/CommandState.java | 70 +-- .../io/seqera/data/command/CommandStatus.java | 17 +- .../command/CommandQueueShowcaseTest.groovy | 4 +- .../data/command/CommandServiceTest.groovy | 60 +-- .../CommandStateSerializationTest.groovy | 62 +-- .../data/command/TestCommandConfig.java | 8 + .../seqera/data/command/TestCommandQueue.java | 17 +- .../src/test/resources/application-test.yml | 1 + .../src/test/resources/logback-test.xml | 2 +- lib-data-stream-redis/README.md | 89 +--- lib-data-stream-redis/VERSION | 2 +- lib-data-stream-redis/changelog.txt | 42 +- .../data/stream/AbstractMessageStream.java | 402 ++---------------- .../io/seqera/data/stream/MessageStream.java | 95 +---- .../data/stream/impl/LocalMessageStream.java | 58 +-- .../data/stream/impl/RedisMessageStream.java | 121 +----- .../data/stream/impl/RedisStreamConfig.java | 44 -- .../data/stream/AsyncStreamLocalTest.groovy | 253 ----------- .../data/stream/AsyncStreamRedisTest.groovy | 209 --------- .../data/stream/LocalMessageStreamTest.groovy | 82 +--- .../io/seqera/data/stream/TestStream.groovy | 2 - .../seqera/data/stream/TunableStream.groovy | 82 ---- .../seqera/data/stream/TestPlainStream.java | 1 - .../io/seqera/data/stream/TestWorkerPool.java | 36 -- 33 files changed, 303 insertions(+), 1923 deletions(-) delete mode 100644 lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/AsyncStreamLocalTest.groovy delete mode 100644 lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/AsyncStreamRedisTest.groovy delete mode 100644 lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/TunableStream.groovy delete mode 100644 lib-data-stream-redis/src/test/java/io/seqera/data/stream/TestWorkerPool.java diff --git a/lib-cmd-queue-redis/README.md b/lib-cmd-queue-redis/README.md index 438786b8..128fdb91 100644 --- a/lib-cmd-queue-redis/README.md +++ b/lib-cmd-queue-redis/README.md @@ -8,7 +8,7 @@ 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' } ``` @@ -16,9 +16,9 @@ dependencies { - 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,7 +78,7 @@ public class AsyncProcessingHandler implements CommandHandler execute(Command 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 @@ -86,7 +86,7 @@ public class AsyncProcessingHandler implements CommandHandler target, CommandConfig config, @Nullable MeterRegistry registry) { - super(target, config, registry != null - ? new MicrometerQueueMetrics(registry, "my-cmd-queue") + public MyCommandQueue(MessageStream 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` (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` (from `lib-data-store-state-redis`). Holds the full JSON state with a TTL (default 7 days). Backed by Redis or in-memory. | -| `CommandHandler` | 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 index faef31a4..1d0ba9ea 100644 --- 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 index 8bfded3f..828cdff2 100644 --- 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 index 9e2bb4b2..d90fbfca 100644 --- 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 `.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 index af64da15..6d5c07e7 100644 --- 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); } } diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandHandler.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandHandler.java index ccf55431..568704fb 100644 --- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandHandler.java +++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandHandler.java @@ -33,11 +33,11 @@ public interface CommandHandler { /** * Execute the command and return a result. - * This method runs on a virtual-thread worker (never on the queue dispatcher thread), - * so it may block for as long as needed without stalling other commands. - * For long-running or external work, return {@link CommandResult#processing()} to indicate - * the operation is in progress; {@link #checkStatus} is then called periodically until - * a terminal result is returned. + * This method is executed asynchronously via an executor service. + * If execution takes longer than 1 second, the command is marked as RUNNING and + * {@link #checkStatus} will be called periodically to check completion. + * For long-running commands, return {@link CommandResult#running()} to indicate + * the operation is in progress. * * @param command The command to execute * @return The result of the execution @@ -46,15 +46,15 @@ public interface CommandHandler { /** * Check the status of a long-running command. - * Called periodically for commands in PROCESSING state until a terminal status is returned. + * Called periodically for commands in RUNNING state until a terminal status is returned. * The command parameter provides typed access to params via {@code command.params()}. * The state parameter provides access to timing and status information. * * @param command The command being checked (provides typed params access) * @param state The current command state (timing, status info) - * @return The result indicating current status (PROCESSING to continue, or terminal status) + * @return The result indicating current status (RUNNING to continue, or terminal status) */ default CommandResult checkStatus(Command

command, CommandState state) { - return CommandResult.processing(); + return CommandResult.running(); } } diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandQueue.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandQueue.java index 8a60b89d..9544d2ef 100644 --- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandQueue.java +++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandQueue.java @@ -16,14 +16,12 @@ */ package io.seqera.data.command; -import java.time.Duration; - import io.micronaut.core.annotation.Nullable; -import io.seqera.data.workqueue.AbstractWorkQueue; -import io.seqera.data.workqueue.MessageConsumer; -import io.seqera.data.workqueue.WorkQueue; -import io.seqera.data.workqueue.metrics.NoopQueueMetrics; -import io.seqera.data.workqueue.metrics.QueueMetrics; +import io.seqera.data.stream.AbstractMessageStream; +import io.seqera.data.stream.MessageConsumer; +import io.seqera.data.stream.MessageStream; +import io.seqera.data.stream.metrics.NoopStreamMetrics; +import io.seqera.data.stream.metrics.StreamMetrics; import io.seqera.serde.encode.StringEncodingStrategy; import io.seqera.serde.moshi.MoshiEncodeStrategy; import jakarta.annotation.PreDestroy; @@ -32,33 +30,30 @@ /** * Abstract message queue for command processing. - * Extends AbstractWorkQueue to provide async, fire-and-forget command submission. + * Extends AbstractMessageStream to provide async, fire-and-forget command submission. * - *

Behaviour knobs ({@link #pollInterval()}, {@link #concurrency()}) are read from the - * supplied {@link CommandConfig}; subclasses only need to implement {@link #name()}. + * Subclasses must implement {@link #name()} and {@link #pollInterval()} + * to configure the queue behavior. */ -public abstract class CommandQueue extends AbstractWorkQueue { +public abstract class CommandQueue extends AbstractMessageStream { private static final Logger log = LoggerFactory.getLogger(CommandQueue.class); - private final CommandConfig config; - - public CommandQueue(WorkQueue target, CommandConfig config) { - this(target, config, null); + public CommandQueue(MessageStream target) { + super(target); + log.info("Created command queue - name={}", name()); } /** * Constructs a command queue with optional metrics instrumentation. * - * @param target the underlying {@link WorkQueue} - * @param config the command-queue configuration (poll interval, concurrency, …) - * @param metrics the {@link QueueMetrics} to publish to, or {@code null} for no-op + * @param target the underlying {@link MessageStream} + * @param metrics the {@link StreamMetrics} to publish to, or {@code null} for no-op */ - public CommandQueue(WorkQueue target, CommandConfig config, @Nullable QueueMetrics metrics) { + public CommandQueue(MessageStream target, @Nullable StreamMetrics metrics) { super(target, metrics); - this.config = config; log.info("Created command queue - name={}; metrics={}", - name(), metrics != null && !(metrics instanceof NoopQueueMetrics) ? "enabled" : "disabled"); + name(), metrics != null && !(metrics instanceof NoopStreamMetrics) ? "enabled" : "disabled"); } @Override @@ -72,24 +67,6 @@ protected StringEncodingStrategy createEncodingStrategy() { @Override protected abstract String name(); - /** Interval for polling the queue, from {@link CommandConfig#pollInterval()}. */ - @Override - protected Duration pollInterval() { - return config.pollInterval(); - } - - /** - * Maximum number of commands in flight on this instance at once, from - * {@link CommandConfig#concurrency()}. Handlers run on virtual threads, so this is a - * memory/heartbeat ceiling rather than a thread count; commands beyond it wait in the - * queue (backpressure). Cross-replica single-runner exclusion is provided by the - * per-message lease, so no per-command lock is required. - */ - @Override - protected int concurrency() { - return config.concurrency(); - } - /** * The name of the message stream, derived from {@link #name()}. */ diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandResult.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandResult.java index 1f0433be..5f85da2e 100644 --- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandResult.java +++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandResult.java @@ -44,22 +44,11 @@ public static CommandResult failure(String error) { return new CommandResult<>(CommandStatus.FAILED, null, error); } - /** - * Indicate that the command is still being processed (for long-running commands). - */ - public static CommandResult processing() { - return new CommandResult<>(CommandStatus.PROCESSING, null, null); - } - /** * Indicate that the command is still running (for long-running commands). - * - * @deprecated renamed to {@link #processing()} to avoid confusion with a downstream - * task's own "running" state. Behaviour is identical; will be removed in a future release. */ - @Deprecated public static CommandResult running() { - return processing(); + return new CommandResult<>(CommandStatus.RUNNING, null, null); } /** diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandServiceImpl.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandServiceImpl.java index 1082bc3e..aebfe7a6 100644 --- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandServiceImpl.java +++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandServiceImpl.java @@ -20,6 +20,9 @@ import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import io.micronaut.scheduling.TaskExecutors; import io.seqera.data.command.store.CommandStateStore; @@ -33,20 +36,17 @@ * Implementation of the command service. * Handles queue consumption and command execution with proper multi-replica support. * - *

Processing runs on the shared worker pool of the underlying message stream, so - * neither {@code execute()} nor {@code checkStatus()} blocks the dispatcher loop and no - * per-command timeout is needed. Cross-replica single-runner exclusion comes from the - * stream's per-message lease. - * *

Processing flow: *

    - *
  • If command is already PROCESSING → call checkStatus()
    • - *
  • If command is still PENDING → call execute()
    • - *
  • If result is PROCESSING → mark as PROCESSING and return false (re-polled later)
    • - *
  • If result is terminal → apply result and return true (message removed from queue)
    • - *
  • If the handler throws → return false so the message is retried; a throw is treated as - * transient, never as a terminal failure (deciding permanent failure is the domain - * layer's job, see seqeralabs/sched#712)
    • + *
  • If command is already RUNNING → call checkStatus() synchronously
    • + *
  • If command is not RUNNING → execute asynchronously with 1-second timeout: + *
      + *
    • If completes within timeout → process result immediately
      • + *
    • If times out → mark as RUNNING, retry later via queue
      • + *
    + *
    • + *
  • If result is RUNNING → return false (message stays in queue for retry)
    • + *
  • If result is terminal → return true (message removed from queue)
    • *
*/ @Singleton @@ -54,6 +54,9 @@ public class CommandServiceImpl implements CommandService { private static final Logger log = LoggerFactory.getLogger(CommandServiceImpl.class); + @Inject + private CommandConfig config; + @Inject private CommandStateStore store; @@ -62,7 +65,7 @@ public class CommandServiceImpl implements CommandService { @Inject @Named(TaskExecutors.BLOCKING) - private ExecutorService blockingExecutor; + private ExecutorService executor; private final Map> handlers = new ConcurrentHashMap<>(); @@ -75,8 +78,6 @@ public void start() { return; } started = true; - // run handlers on the shared Micronaut BLOCKING (virtual-thread) executor - queue.withHandlerExecutor(blockingExecutor); queue.addConsumer(this::processCommand); log.info("Command service started - consuming commands"); } @@ -220,11 +221,16 @@ private boolean processCommand(CommandMsg msg) { *

This helper method captures the type parameters {@code } from the * {@link CommandRegistration}, allowing type-safe interaction with the handler. * - *

Runs directly on the shared worker pool thread (no timeout, no extra executor): + *

Processing flow: *

    - *
  1. If command is already PROCESSING → call {@code checkStatus()} to poll for completion
    2. - *
  2. If command is still PENDING → call {@code execute()}
    3. - *
  3. If result status is PROCESSING → mark PROCESSING and return false (re-polled later)
    4. + *
  4. If command is already RUNNING → call {@code checkStatus()} to poll for completion
    5. + *
  5. If command is not yet RUNNING → call {@code execute()} with timeout: + *
      + *
    • If completes within timeout → process the result immediately
      • + *
    • If times out → mark as RUNNING, return false to retry later
      • + *
    + *
    6. + *
  6. If result status is RUNNING → return false (keep in queue for polling)
    7. *
  7. If result status is terminal → update state and return true (done)
    8. *
* @@ -246,29 +252,34 @@ private boolean processCommandWithHandler( final CommandHandler handler = registration.handler(); try { - // Branch on the command status. Only PENDING and PROCESSING are reachable here - // (processCommand already acked terminal states); any other value is a bug or a - // newly-added status and must fail loudly rather than be silently executed. Both - // execute() and checkStatus() run on the shared worker pool, so a slow handler - // does not block the loop. - final CommandResult result = switch (state.status()) { - case PENDING -> handler.execute(command); - case PROCESSING -> handler.checkStatus(command, state); - default -> throw new IllegalStateException("Unexpected command status: " + state.status() + " - id=" + state.id()); - }; + CommandResult result; + + // Branch based on current command status + if (state.status() == CommandStatus.RUNNING) { + // Command was previously marked as RUNNING (long-running async operation) + // Call checkStatus() to poll the external system for completion + result = handler.checkStatus(command, state); + } else { + // Command not yet running (status is SUBMITTED) + // Execute with timeout to avoid blocking the queue processor indefinitely + result = executeWithTimeout(handler, command); + + // Timeout case: execute() is still running in background thread + // Mark state as RUNNING so next delivery will call checkStatus() instead + if (result == null) { + store.save(state.started()); + return false; // Keep in queue - will retry and call checkStatus() + } + } // Handler returned a result - check if command is still in progress - if (result.status() == CommandStatus.PROCESSING) { - // Handler explicitly returned PROCESSING (e.g., async job not yet complete) - // Ensure state reflects PROCESSING status for accurate reporting - if (state.status() != CommandStatus.PROCESSING) { + if (result.status() == CommandStatus.RUNNING) { + // Handler explicitly returned RUNNING (e.g., async job not yet complete) + // Ensure state reflects RUNNING status for accurate reporting + if (state.status() != CommandStatus.RUNNING) { store.save(state.started()); - } else if (state.errorsCount() > 0) { - // Recovered after one or more transient errors — reset the streak. Single write, - // and only when there is something to reset, so healthy re-polls stay write-free. - store.save(state.clearErrors()); } - return false; // Keep in queue - re-polled and will call checkStatus() + return false; // Keep in queue - will retry and call checkStatus() } // Terminal result (SUCCEEDED, FAILED, or CANCELLED) @@ -279,31 +290,47 @@ private boolean processCommandWithHandler( return true; // Remove from queue - processing complete } catch (Exception e) { - // A thrown handler is a transient/retryable condition, NOT a terminal command - // outcome: keep the message in the queue (return false) so the stream layer retains - // its lease and re-polls it. A genuine command failure is signalled by returning a - // FAILED CommandResult (handled above), never by throwing. Persisting FAILED + acking - // here would turn a transient/infra error (e.g. the Postgres pool closing during - // shutdown) into a permanent FAILED command while the domain entity is left - // non-terminal, stranding the work. Deciding a command has *permanently* failed is - // delegated to the domain layer that owns the entity state (see seqeralabs/sched#712). - log.error("Command processing errored, will retry: id={}", msg.commandId(), e); - recordError(state, e); - return false; // Keep in queue - redelivered / re-polled + // Unexpected exception during processing - mark as FAILED + log.error("Command processing failed: id={}", msg.commandId(), e); + store.save(state.failed(e.getMessage())); + return true; // Remove from queue - no point retrying a crashed handler } } /** - * Best-effort: record a non-terminal processing error on the command state — increment the - * consecutive-error count and capture the message — for observability of a retry storm on a - * command that stays retryable. A failure to persist this must not change control flow: the - * command is kept in the queue and retried regardless. + * Execute a command handler with a timeout. + * + *

Submits the handler's {@code execute()} method to a thread pool and waits + * up to {@code config.executeTimeout()} for completion. This prevents slow handlers from + * blocking the queue processor thread. + * + *

Timeout behavior: If the handler doesn't complete within the timeout, + * this method returns {@code null} but the handler continues executing in the + * background. The caller should mark the command as RUNNING and retry later + * via {@code checkStatus()}. + * + * @param handler The command handler to execute + * @param command The command with parameters + * @param

The command parameter type + * @param The command result type + * @return The result if completed within timeout, or {@code null} if timed out + * @throws RuntimeException if the handler throws an exception */ - private void recordError(CommandState state, Exception e) { + private CommandResult executeWithTimeout(CommandHandler handler, Command

command) { + // Submit handler execution to thread pool for async execution + final Future> future = executor.submit(() -> handler.execute(command)); + try { - store.save(state.withError(e.getMessage() != null ? e.getMessage() : e.toString())); - } catch (Exception fail) { - log.warn("Failed to record command error state: id={}", state.id(), fail); + // Block until result is available or timeout expires + return future.get(config.executeTimeout().toMillis(), TimeUnit.MILLISECONDS); + } catch (TimeoutException e) { + // Handler is taking longer than allowed - let it continue in background + // Caller will mark as RUNNING and poll via checkStatus() on retry + return null; + } catch (Exception e) { + // Handler threw an exception - cancel the future and propagate + future.cancel(true); + throw new RuntimeException("Command execution failed", e); } } } diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandState.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandState.java index f4ca3898..c922eb01 100644 --- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandState.java +++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandState.java @@ -14,6 +14,7 @@ * limitations under the License. * */ + package io.seqera.data.command; import java.time.Instant; @@ -25,28 +26,6 @@ * Persistent state of a command, stored as JSON in the database. * Uses @JsonTypeInfo to preserve type information for params and result * during serialization, enabling proper deserialization without explicit type knowledge. - * - *

{@code errorsCount} counts processing errors that did not terminally fail the - * command — a handler that threw is retried (see {@code CommandServiceImpl}), so this records how - * many consecutive times it has thrown, for observability of a retry storm on an otherwise - * non-terminal command. {@code error} holds the message of the most recent error, transient or - * terminal — check {@code status == FAILED} to tell a terminal failure from a transient one, not - * {@code error != null}. {@code modifiedAt} is refreshed on every state write, giving a - * last-touched timestamp. - * - * @param id command id - * @param type command type discriminator - * @param status current lifecycle status - * @param params command parameters (polymorphic, type preserved via {@code @JsonTypeInfo}) - * @param result terminal result payload, if any (polymorphic) - * @param error message of the most recent error, transient or terminal (nullable); terminal only - * when {@code status == FAILED} - * @param errorsCount number of consecutive processing errors since the last successful - * processing; reset to 0 on any successful transition or recovery - * @param createdAt when the command was first submitted - * @param startedAt when the command first transitioned to PROCESSING (nullable) - * @param modifiedAt when the command state was last written (nullable for pre-existing records) - * @param completedAt when the command reached a terminal state (nullable) */ public record CommandState( String id, @@ -57,10 +36,8 @@ public record CommandState( @JsonTypeInfo(use = JsonTypeInfo.Id.CLASS) @Nullable Object result, @Nullable String error, - int errorsCount, Instant createdAt, @Nullable Instant startedAt, - @Nullable Instant modifiedAt, @Nullable Instant completedAt ) { @@ -68,22 +45,19 @@ public record CommandState( * Create a new submitted command state. */ public static CommandState submitted(String id, String type, Object params) { - final Instant now = Instant.now(); return new CommandState( - id, type, CommandStatus.PENDING, params, - null, null, 0, now, null, now, null + id, type, CommandStatus.SUBMITTED, params, + null, null, Instant.now(), null, null ); } /** - * Transition to PROCESSING status. A successful (non-throwing) transition, so the - * consecutive-error streak is reset. + * Transition to RUNNING status. */ public CommandState started() { - final Instant now = Instant.now(); return new CommandState( - id, type, CommandStatus.PROCESSING, params, - result, error, 0, createdAt, now, now, completedAt + id, type, CommandStatus.RUNNING, params, + result, error, createdAt, Instant.now(), completedAt ); } @@ -91,10 +65,9 @@ public CommandState started() { * Transition to SUCCEEDED status with result. */ public CommandState completed(Object result) { - final Instant now = Instant.now(); return new CommandState( id, type, CommandStatus.SUCCEEDED, params, - result, null, 0, createdAt, startedAt, now, now + result, null, createdAt, startedAt, Instant.now() ); } @@ -102,10 +75,9 @@ public CommandState completed(Object result) { * Transition to FAILED status with error. */ public CommandState failed(String error) { - final Instant now = Instant.now(); return new CommandState( id, type, CommandStatus.FAILED, params, - null, error, errorsCount, createdAt, startedAt, now, now + null, error, createdAt, startedAt, Instant.now() ); } @@ -113,33 +85,9 @@ public CommandState failed(String error) { * Transition to CANCELLED status. */ public CommandState cancelled() { - final Instant now = Instant.now(); return new CommandState( id, type, CommandStatus.CANCELLED, params, - null, null, 0, createdAt, startedAt, now, now - ); - } - - /** - * Record a non-terminal processing error: keep the current status (the command stays retryable), - * increment the consecutive-error count, capture the message, and refresh {@code modifiedAt}. - * Called when a handler throws and the command is kept in the queue for retry. - */ - public CommandState withError(String message) { - return new CommandState( - id, type, status, params, - result, message, errorsCount + 1, createdAt, startedAt, Instant.now(), completedAt - ); - } - - /** - * Clear the consecutive-error streak after a recovery, without changing status. Refreshes - * {@code modifiedAt}. {@code error} is retained as a historical marker of the last error seen. - */ - public CommandState clearErrors() { - return new CommandState( - id, type, status, params, - result, error, 0, createdAt, startedAt, Instant.now(), completedAt + null, null, createdAt, startedAt, Instant.now() ); } diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandStatus.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandStatus.java index 400c25ab..8114e6fd 100644 --- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandStatus.java +++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandStatus.java @@ -17,23 +17,16 @@ package io.seqera.data.command; -import com.fasterxml.jackson.annotation.JsonAlias; - /** * Status of a command in the queue. - * - *

Wire compatibility: {@code PENDING} and {@code PROCESSING} are the renamed forms of the - * former {@code SUBMITTED} and {@code RUNNING}. New state is serialized with the new names, - * while the legacy names are still accepted on read via {@link JsonAlias}, so command state - * persisted by earlier versions continues to deserialize. Do not remove those aliases. */ public enum CommandStatus { - /** In the queue, awaiting first processing (legacy wire name: {@code "SUBMITTED"}). */ - @JsonAlias("SUBMITTED") + /** Created, not yet submitted to queue */ PENDING, - /** Being processed by a handler (legacy wire name: {@code "RUNNING"}). */ - @JsonAlias("RUNNING") - PROCESSING, + /** In queue, waiting for pickup */ + SUBMITTED, + /** Being executed */ + RUNNING, /** Completed successfully */ SUCCEEDED, /** Completed with error */ diff --git a/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandQueueShowcaseTest.groovy b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandQueueShowcaseTest.groovy index 418a5900..08fad2cf 100644 --- a/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandQueueShowcaseTest.groovy +++ b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandQueueShowcaseTest.groovy @@ -128,8 +128,8 @@ class CommandQueueShowcaseTest extends Specification { sleep(500) def initialState = commandService.getState(commandId).orElseThrow() - then: 'command is in PROCESSING state (async processing started)' - initialState.status() == CommandStatus.PROCESSING + then: 'command is in RUNNING state (async processing started)' + initialState.status() == CommandStatus.RUNNING when: 'wait for async completion via periodic status checks' sleep(4000) diff --git a/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandServiceTest.groovy b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandServiceTest.groovy index 6cff913c..3bcefb20 100644 --- a/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandServiceTest.groovy +++ b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandServiceTest.groovy @@ -31,7 +31,7 @@ import java.time.Instant /** * End-to-end tests for the CommandService. */ -@MicronautTest(packages = ["io.seqera.data.workqueue"], transactional = false) +@MicronautTest(packages = ["io.seqera.data.stream"], transactional = false) @TestInstance(TestInstance.Lifecycle.PER_CLASS) class CommandServiceTest extends Specification implements TestPropertyProvider { @@ -158,8 +158,8 @@ class CommandServiceTest extends Specification implements TestPropertyProvider { sleep(500) def state = commandService.getState(command.id()).orElseThrow() - then: 'status is PROCESSING' - state.status() == CommandStatus.PROCESSING + then: 'status is RUNNING' + state.status() == CommandStatus.RUNNING when: 'wait for periodic checker' sleep(3000) @@ -172,43 +172,6 @@ class CommandServiceTest extends Specification implements TestPropertyProvider { result.processedValue == 99 } - def 'should retry a handler that throws instead of failing it terminally'() { - given: 'a command whose handler throws on the first attempt, then succeeds' - def params = new TestParams(7, 'flaky') - def command = new TestCommand(TsidCreator.getTsid().toLowerCase(), 'test', params) - - when: 'command is submitted' - commandService.submit(command) - - and: 'wait for the first (throwing) attempt to be retried' - sleep(3000) - def state = commandService.getState(command.id()).orElseThrow() - - then: 'the throw was treated as transient and retried to success, not persisted as FAILED' - state.status() == CommandStatus.SUCCEEDED - commandService.getResult(command.id(), TestResult).orElseThrow().message == 'Recovered' - - and: 'the consecutive-error streak is reset once the command recovers' - state.errorsCount() == 0 - } - - def 'should track consecutive errors and last message without failing a still-retryable command'() { - given: 'a handler that always throws' - def params = new TestParams(0, 'always-throw') - def command = new TestCommand(TsidCreator.getTsid().toLowerCase(), 'test', params) - - when: 'command is submitted and retried a few times' - commandService.submit(command) - sleep(2000) - def state = commandService.getState(command.id()).orElseThrow() - - then: 'the command stays retryable while the error streak and last message are recorded' - !state.status().isTerminal() - state.errorsCount() >= 1 - state.error() == 'Persistent boom' - state.modifiedAt() != null - } - def 'should handle unknown command type'() { given: def params = new TestParams(42, 'fast') @@ -276,7 +239,6 @@ class TestCommand implements Command { class TestCommandHandler implements CommandHandler { private Instant startTime - private final java.util.concurrent.atomic.AtomicInteger flakyAttempts = new java.util.concurrent.atomic.AtomicInteger() @Override String type() { 'test' } @@ -289,21 +251,9 @@ class TestCommandHandler implements CommandHandler { return CommandResult.failure('Intentional failure') } - if (params.mode == 'flaky') { - // throw on the first attempt (simulating a transient/infra error), succeed on retry - if (flakyAttempts.getAndIncrement() == 0) { - throw new RuntimeException('Transient failure') - } - return CommandResult.success(new TestResult('Recovered', params.value)) - } - - if (params.mode == 'always-throw') { - throw new RuntimeException('Persistent boom') - } - if (params.mode == 'slow') { startTime = Instant.now() - return CommandResult.processing() + return CommandResult.running() } def result = new TestResult('Processed', params.value) @@ -321,6 +271,6 @@ class TestCommandHandler implements CommandHandler { } } - return CommandResult.processing() + return CommandResult.running() } } diff --git a/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/store/CommandStateSerializationTest.groovy b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/store/CommandStateSerializationTest.groovy index 64cc001b..74544596 100644 --- a/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/store/CommandStateSerializationTest.groovy +++ b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/store/CommandStateSerializationTest.groovy @@ -20,7 +20,6 @@ import java.time.Instant import com.fasterxml.jackson.annotation.JsonTypeInfo import groovy.transform.Canonical -import io.seqera.data.command.CommandState import io.seqera.data.command.CommandStatus import io.seqera.serde.jackson.JacksonEncodingStrategy import spock.lang.Specification @@ -146,7 +145,7 @@ class CommandStateSerializationTest extends Specification { decoded.result.deleted } - def 'should handle null result for processing commands'() { + def 'should handle null result for running commands'() { given: def encoder = new JacksonEncodingStrategy>() {} def now = Instant.now() @@ -154,7 +153,7 @@ class CommandStateSerializationTest extends Specification { def state = new TypedCommandState<>( 'cmd-xyz', 'create-job', - CommandStatus.PROCESSING, + CommandStatus.RUNNING, params, null, // No result yet null, @@ -171,61 +170,6 @@ class CommandStateSerializationTest extends Specification { decoded.params instanceof CreateJobParams decoded.params.image == 'ubuntu:22.04' decoded.result == null - decoded.status == CommandStatus.PROCESSING - } - - def 'new state uses the new wire names but legacy names still decode (compatibility)'() { - given: - def encoder = new JacksonEncodingStrategy>() {} - def now = Instant.now() - def params = new CreateJobParams('ubuntu:22.04', 'sleep 60', 2, 1024) - - when: 'a PENDING / PROCESSING state is encoded' - def pendingJson = encoder.encode(new TypedCommandState<>('c1', 't', CommandStatus.PENDING, params, null, null, now, null, null)) - def processingJson = encoder.encode(new TypedCommandState<>('c2', 't', CommandStatus.PROCESSING, params, null, null, now, now, null)) - - then: 'new values are written with the NEW names' - pendingJson.contains('"PENDING"') - processingJson.contains('"PROCESSING"') - - and: 'they round-trip' - encoder.decode(pendingJson).status == CommandStatus.PENDING - encoder.decode(processingJson).status == CommandStatus.PROCESSING - - and: 'legacy entries written by earlier versions still decode to the renamed constants' - encoder.decode(pendingJson.replace('"PENDING"', '"SUBMITTED"')).status == CommandStatus.PENDING - encoder.decode(processingJson.replace('"PROCESSING"', '"RUNNING"')).status == CommandStatus.PROCESSING - } - - def 'should decode legacy JSON without error-tracking fields into the real record'() { - given: 'the encoder as wired by CommandStateStoreFactory' - def encoder = new JacksonEncodingStrategy() {} - def now = Instant.now() - and: 'old-format JSON, before errorsCount/modifiedAt existed' - def paramsClass = CreateJobParams.name - def legacyJson = """\ - { - "id": "cmd-legacy", - "type": "create-job", - "status": "RUNNING", - "params": {"@class": "${paramsClass}", "image": "alpine:latest", "command": "echo hi", "cpu": 1, "memory": 512}, - "result": null, - "error": null, - "createdAt": "${now}", - "startedAt": "${now}", - "completedAt": null - }""".stripIndent() - - when: - def decoded = encoder.decode(legacyJson) - - then: 'existing fields survive (legacy wire name "RUNNING" decodes to PROCESSING via @JsonAlias)' - decoded.id() == 'cmd-legacy' - decoded.status() == CommandStatus.PROCESSING - decoded.params() instanceof CreateJobParams - decoded.params().image == 'alpine:latest' - and: 'new fields default without a stored value — safe rolling deploy' - decoded.errorsCount() == 0 - decoded.modifiedAt() == null + decoded.status == CommandStatus.RUNNING } } diff --git a/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestCommandConfig.java b/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestCommandConfig.java index 9260d637..0e293cf8 100644 --- a/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestCommandConfig.java +++ b/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestCommandConfig.java @@ -32,6 +32,9 @@ public class TestCommandConfig implements CommandConfig { @Value("${command.poll-interval:100ms}") private Duration pollInterval; + @Value("${command.execute-timeout:1s}") + private Duration executeTimeout; + @Value("${command.state.ttl:1h}") private Duration stateTtl; @@ -40,6 +43,11 @@ public Duration pollInterval() { return pollInterval; } + @Override + public Duration executeTimeout() { + return executeTimeout; + } + @Override public Duration stateTtl() { return stateTtl; diff --git a/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestCommandQueue.java b/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestCommandQueue.java index 0b375b04..28c08010 100644 --- a/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestCommandQueue.java +++ b/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestCommandQueue.java @@ -16,8 +16,10 @@ */ package io.seqera.data.command; +import java.time.Duration; + import io.micronaut.context.annotation.Factory; -import io.seqera.data.workqueue.WorkQueue; +import io.seqera.data.stream.MessageStream; import jakarta.inject.Singleton; /** @@ -25,14 +27,19 @@ */ class TestCommandQueue extends CommandQueue { - TestCommandQueue(WorkQueue target, CommandConfig config) { - super(target, config); + TestCommandQueue(MessageStream target) { + super(target); } @Override protected String name() { return "test-command-queue"; } + + @Override + protected Duration pollInterval() { + return Duration.ofMillis(100); + } } /** @@ -42,7 +49,7 @@ protected String name() { class TestCommandQueueFactory { @Singleton - CommandQueue commandQueue(WorkQueue target, CommandConfig config) { - return new TestCommandQueue(target, config); + CommandQueue commandQueue(MessageStream target) { + return new TestCommandQueue(target); } } diff --git a/lib-cmd-queue-redis/src/test/resources/application-test.yml b/lib-cmd-queue-redis/src/test/resources/application-test.yml index 71260fa1..72717358 100644 --- a/lib-cmd-queue-redis/src/test/resources/application-test.yml +++ b/lib-cmd-queue-redis/src/test/resources/application-test.yml @@ -7,6 +7,7 @@ micronaut: # Command queue configuration command: poll-interval: 100ms + execute-timeout: 1s state: ttl: 1h diff --git a/lib-cmd-queue-redis/src/test/resources/logback-test.xml b/lib-cmd-queue-redis/src/test/resources/logback-test.xml index 1eec64a0..2140c7df 100644 --- a/lib-cmd-queue-redis/src/test/resources/logback-test.xml +++ b/lib-cmd-queue-redis/src/test/resources/logback-test.xml @@ -24,7 +24,7 @@ - + diff --git a/lib-data-stream-redis/README.md b/lib-data-stream-redis/README.md index 9caab6bf..d77925b5 100644 --- a/lib-data-stream-redis/README.md +++ b/lib-data-stream-redis/README.md @@ -1,18 +1,5 @@ # lib-data-stream-redis -> **⚠️ Deprecated.** This module is frozen and kept only for existing consumers; no further -> changes will be made here. You have two paths: -> -> - **Move to [`lib-data-workqueue`](../lib-data-workqueue/README.md) + -> [`lib-data-workqueue-redis`](../lib-data-workqueue-redis/README.md)** — the split/rename of -> this library with aligned vocabulary (`poll`→`receive`, `renew`→`renewLease`, -> `claim-timeout`→`visibility-timeout`). `workqueue 1.0.0` ≡ this library's `2.0.0` behaviour -> (async, at-least-once, heartbeat lease — handlers must be idempotent). Recommended for new -> code. See the [migration guide](../docs/superpowers/specs/2026-07-11-workqueue-rename-migration.md). -> - **Stay on `lib-data-stream-redis:1.5.x`** — the last *synchronous* release (handler runs on -> the listener thread, exactly-once-per-poll). Pin `1.5.x` if you don't want the `2.0.0` -> async/at-least-once rewrite and aren't ready to adopt the idempotency requirement. - Message streaming with Redis Streams and local implementations for persistent event processing. ## Installation @@ -21,7 +8,7 @@ Add this dependency to your `build.gradle`: ```gradle dependencies { - implementation 'io.seqera:lib-data-stream-redis:2.0.0' + implementation 'io.seqera:lib-data-stream-redis:1.5.0' } ``` @@ -155,80 +142,8 @@ class ActivityConsumer implements MessageConsumer { messageStream.consume("user-activity", new ActivityConsumer()) ``` -## Architecture - -`AbstractMessageStream` runs handlers **asynchronously and concurrently** while -guaranteeing that a given message is processed by exactly one *live* consumer at a -time. A message is owned by its consumer for as long as the handler keeps working — -independent of how long that takes — and ownership is relinquished only when the work -finishes or the consumer dies. - -``` - offer(msg) ┌────────────────────────────────┐ - │ │ AbstractMessageStream │ - ▼ │ │ - ┌──────────┐ poll (XREADGROUP / XAUTOCLAIM) │ dispatcher thread │ - │ Redis │◀──────────────────────────────────┤ • acquire a semaphore slot │ - │ stream │ │ • poll one message │ - │ (PEL, │ renew (XCLAIM … JUSTID) │ • hand it to the executor │ - │ group) │◀───────────── heartbeat daemon ───┤ (never runs it inline) │ - │ │ every claim-timeout/3 │ │ - │ │ ack (XACK + XDEL) │ worker (executor thread) │ - │ │◀──────────── on terminal ─────────┤ accept(msg): │ - └──────────┘ │ ├─ true → ack + free slot │ - ▲ │ └─ false → keep lease, │ - │ reclaimed by a peer only if the owner │ re-run after pollInterval│ - │ dies (heartbeat stops → idle > claim-timeout) via the re-poll scheduler │ - └──────────────────────────────────────────────────────────────────────────┘ -``` - -**Three mechanisms:** - -1. **Async dispatch (no head-of-line blocking).** The dispatcher thread never runs a - handler; it hands each message to a worker executor and moves on. Handlers run on the - executor supplied via `withHandlerExecutor(...)` — **mandatory, no default** (Micronaut - consumers inject the `@Named(BLOCKING)` executor). A `Semaphore` sized by - `concurrency()` bounds how many messages are in flight at once (backpressure: excess - messages stay in the stream). - -2. **Heartbeat lease (single live runner + safe long handlers).** While a message is in - flight, a daemon renews its Redis consumer-group entry (`XCLAIM … JUSTID`) every - `claim-timeout / 3`, pinning its idle time near zero so no peer's `XAUTOCLAIM` can - reclaim it — no matter how long the handler runs. If the owning process dies, the - heartbeat stops, idle time crosses `claim-timeout`, and a peer reclaims the message - (real dead-consumer failover). A `max-processing-time` safety valve stops renewing a - single invocation that runs pathologically long, without interrupting its thread. - -3. **In-process re-poll for not-yet-terminal work.** When a handler returns `false` (work - in progress), the message keeps its lease and the handler is **re-invoked in-process** - after `pollInterval` via a scheduler — Redis is not re-read. This makes the re-poll - cadence independent of `claim-timeout` (which then governs only failover). - - The re-poll is scheduled **after the handler returns** — a fixed *delay*, not a fixed - *rate*. So a handler that runs **longer than `pollInterval` never overlaps itself**: the - next call starts `pollInterval` after the previous one finished, and a given message is - processed by at most one invocation at a time (regardless of how slow the handler is). - The *only* exception is an invocation that exceeds `max-processing-time` — the safety - valve then stops renewing the lease so a concurrent reclaim becomes possible (see below), - which is why handlers must be idempotent. - -Delivery is **at-least-once** (a crash/pause beyond `claim-timeout`, or the -`max-processing-time` valve, can hand a still-running message to a peer), so consumers -must be idempotent. The in-memory `LocalMessageStream` has no pending-entries list, so it -has no lease/heartbeat (renew is a no-op); it still benefits from async, concurrent dispatch. - -### Configuration - -| Knob | Where | Default | Governs | -|---|---|---|---| -| `pollInterval()` | `AbstractMessageStream` | — (subclass) | Idle backoff **and** in-process re-poll cadence | -| `concurrency()` | `AbstractMessageStream` | `1` | Max in-flight messages (semaphore ceiling) | -| `getClaimTimeout()` | `RedisStreamConfig` | — | Dead-consumer failover window | -| `getHeartbeatInterval()` | `RedisStreamConfig` | `claim-timeout / 3` | Lease renewal cadence | -| `getMaxProcessingTime()` | `RedisStreamConfig` | `15m` | Upper bound on a single `accept()` before its lease is released | - ## Testing ```bash ./gradlew :lib-data-stream-redis:test -``` +``` \ No newline at end of file diff --git a/lib-data-stream-redis/VERSION b/lib-data-stream-redis/VERSION index 359a5b95..bc80560f 100644 --- a/lib-data-stream-redis/VERSION +++ b/lib-data-stream-redis/VERSION @@ -1 +1 @@ -2.0.0 \ No newline at end of file +1.5.0 diff --git a/lib-data-stream-redis/changelog.txt b/lib-data-stream-redis/changelog.txt index 1911cf6f..956f7603 100644 --- a/lib-data-stream-redis/changelog.txt +++ b/lib-data-stream-redis/changelog.txt @@ -1,39 +1,13 @@ # lib-data-stream-redis changelog -2.0.0 - 11 Jul 2026 -- BREAKING (SPI): MessageStream gains abstract poll/renew/ack/release plus the Lease - record. Any class IMPLEMENTING MessageStream must add them (the in-repo Redis/Local impls - are updated). consume(streamId, consumer) is RETAINED as a default implemented over the - triad, so callers of consume() remain source-compatible. -- BREAKING (subclass): the protected processMessage(...) signature dropped its trailing - AtomicInteger count parameter; subclasses overriding it must update. -- BREAKING (behaviour): handlers now run asynchronously off the listener thread (on virtual - threads / an injectable executor); delivery is at-least-once with a heartbeat lease and - in-process re-poll, so consumers MUST be idempotent. concurrency() defaults to 1, so intake - stays effectively serial until a subclass opts in to parallelism. -- Async, non-blocking consumer processing: the listener thread becomes a dispatcher that - hands each message to a virtual-thread executor and never runs a handler itself, removing - head-of-line blocking and enabling concurrent processing. Handlers run on a shared - virtual-thread executor by default; supply the Micronaut @Named(BLOCKING) executor via - AbstractMessageStream.withHandlerExecutor(...). Concurrency is bounded by a semaphore - (see concurrency()), not by the executor. -- Worker-scoped heartbeat lease: while a handler is in-flight, a daemon renews its Redis - entry (XCLAIM ... JUSTID) so an alive consumer is never reclaimed regardless of duration; - the lease lapses on crash to preserve real failover. Decouples processing time from - claim-timeout. -- Model B (hold-lease + scheduled re-poll): a not-yet-terminal command keeps its lease and - is re-invoked in-process after pollInterval instead of waiting for a Redis reclaim, - decoupling re-poll cadence from claim-timeout. -- SPI evolution on MessageStream: add lease triad poll/renew/ack/release plus Lease - record; add heartbeatInterval()/maxProcessingTime() (default null; overridden by the Redis - impl to expose the configured values) so the base derives the renew cadence from the - underlying stream's claim-timeout. consume(streamId, consumer) is retained as a default - implemented over the triad (backward compatible, additive). -- RedisStreamConfig gains getHeartbeatInterval() (default claim-timeout/3) and - getMaxProcessingTime() (default 15m) with their *Millis() variants; RedisMessageStream - wires them through so the heartbeat cadence tracks any configured claim-timeout. -- AbstractMessageStream.concurrency() defaults to 1 (a semaphore ceiling on in-flight - commands), overridable by subclasses. +REVERTED - 2.0.0 (11 Jul 2026) +- 2.0.0 was released and has been reverted; this module is back at 1.5.0. It brought async, + non-blocking consumer processing with a heartbeat lease and the poll/renew/ack/release SPI + (PR #84, commit bdf9374), plus doc-only follow-ups (11fdd3a, 86ae9ac) and the injected + handler executor from PR #94 (d38afb0). +- The lease-based design is not lost: it lives on in lib-data-workqueue / + lib-data-workqueue-redis, which supersede this module for new work. +- Sources and full changelog of the reverted work: branch archive/workqueue-pre-revert. 1.5.0 - 14 May 2026 - BREAKING (observability): rename Outcome.FAILED -> Outcome.ACTIVE; metric tag value diff --git a/lib-data-stream-redis/src/main/java/io/seqera/data/stream/AbstractMessageStream.java b/lib-data-stream-redis/src/main/java/io/seqera/data/stream/AbstractMessageStream.java index 140ac345..693f5bbf 100644 --- a/lib-data-stream-redis/src/main/java/io/seqera/data/stream/AbstractMessageStream.java +++ b/lib-data-stream-redis/src/main/java/io/seqera/data/stream/AbstractMessageStream.java @@ -20,15 +20,7 @@ import java.io.Closeable; import java.time.Duration; import java.util.Map; -import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.Semaphore; -import java.util.concurrent.ThreadFactory; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import io.micronaut.core.annotation.Nullable; @@ -78,9 +70,6 @@ * // Usage * MyMessageStream stream = new MyMessageStream(underlyingStream); * - * // Supply the handler executor (mandatory, no default) before adding consumers - * stream.withHandlerExecutor(executorService); - * * // Add consumer for a specific stream * stream.addConsumer("user-events", event -> { * processUserEvent(event); @@ -122,64 +111,10 @@ public abstract class AbstractMessageStream implements Closeable { private final StreamMetrics metrics; - private volatile Thread thread; + private Thread thread; private final String name0; - /** - * A message picked up from a stream and held while it is processed. The - * {@code streamId} + {@code leaseId} pair identifies the delivered entry; the - * {@code message} is kept so a not-yet-terminal command can be re-invoked in-process - * (Model B) without re-reading it from the stream. - */ - private record InFlight(String streamId, String leaseId, String message) { - String key() { - return streamId + '|' + leaseId; - } - } - - /** - * Leases held from pickup to terminal/crash; every entry is heartbeated by the - * daemon so an alive consumer is never reclaimed. Keyed by {@code streamId|leaseId}. - */ - private final Map inFlight = new ConcurrentHashMap<>(); - - /** - * Subset of {@link #inFlight} whose {@code accept()} invocation is running right now, - * mapped to the wall-clock millis at which that invocation started. Used by the - * heartbeat daemon to enforce {@code max-processing-time} on a single invocation. - */ - private final Map active = new ConcurrentHashMap<>(); - - /** - * Executor that runs the message handlers. Supplied by the consumer via - * {@link #withHandlerExecutor} before the first {@link #addConsumer} — there is no default, - * so {@link #startProcessing()} fails fast if it was never set. Micronaut consumers pass the - * injected {@code BLOCKING} executor. Handler concurrency is bounded by {@link #slots}, not by - * this executor, so it is never sized or shut down here. - */ - private volatile ExecutorService pool; - - /** - * Gates new intake: a permit is acquired when a lease is picked up and held for the - * whole lease lifetime (across re-polls), released on terminal ack / eviction / - * release. This bounds concurrent handlers to {@link #concurrency()} and reserves - * capacity so in-flight commands' re-polls are never starved by new intake. - */ - private volatile Semaphore slots; - - /** - * Schedules delayed re-poll re-submissions for not-yet-terminal commands (Model B). - */ - private volatile ScheduledExecutorService scheduler; - - /** - * Renews every in-flight lease on a fixed cadence so an alive consumer keeps ownership. - */ - private volatile ScheduledExecutorService heartbeat; - - private volatile boolean closed; - /** * Constructs a new stream without metrics instrumentation. Behavior is identical * to passing {@link NoopStreamMetrics#INSTANCE} to {@link #AbstractMessageStream(MessageStream, StreamMetrics)}. @@ -225,44 +160,10 @@ protected Thread createListenerThread() { /** * @return * The time interval to await before trying to read again the stream - * when no more entries are available. Also the cadence at which a - * not-yet-terminal command is re-invoked in-process (Model B). + * when no more entries are available. */ protected abstract Duration pollInterval(); - /** - * @return - * The maximum number of message handlers that may run concurrently on this - * instance (the worker pool size). Defaults to {@code 1}; subclasses may - * override to enable parallel processing. - */ - protected int concurrency() { - return 1; - } - - /** - * @return - * How often in-flight leases are renewed so an alive consumer keeps ownership - * of its message regardless of how long its handler runs. Must be shorter than - * the underlying stream's claim timeout; subclasses backed by a configuration - * should wire this to {@code claim-timeout / 3}. - */ - protected Duration heartbeatInterval() { - final Duration d = stream.heartbeatInterval(); - return d != null ? d : Duration.ofSeconds(20); - } - - /** - * @return - * The upper bound on a single {@code accept()} invocation before its lease is - * released (safety valve); it does not interrupt the handler thread. Defaults - * to {@code 15m}. - */ - protected Duration maxProcessingTime() { - final Duration d = stream.maxProcessingTime(); - return d != null ? d : Duration.ofMinutes(15); - } - /** * Adds a message to the specified stream for asynchronous processing. * @@ -317,53 +218,13 @@ public void addConsumer(String streamId, MessageConsumer consumer) { listeners.put(streamId, consumer); // bind the backlog gauge for this stream id (no-op when metrics disabled) metrics.bindBacklog(streamId, () -> stream.length(streamId)); - // finally start the dispatcher thread and its supporting executors + // finally start the listener thread if (thread == null) { - startProcessing(); + thread = createListenerThread(); } } } - /** - * Lazily create the worker pool, the re-poll scheduler, the heartbeat daemon and the - * capacity gate, then start the dispatcher thread. Invoked once, when the first - * consumer is registered. - */ - private void startProcessing() { - // a handler executor must be supplied via withHandlerExecutor() before processing starts - Objects.requireNonNull(pool, "Handler executor not set - call withHandlerExecutor() before addConsumer()"); - // 'slots' — not the executor — bounds how many commands may be in flight at once; - // the cap is a memory/heartbeat ceiling, independent of the executor's threading model. - this.slots = new Semaphore(Math.max(1, concurrency())); - this.scheduler = new ScheduledThreadPoolExecutor(1, daemonFactory(name() + "-repoll-" + count.get())); - this.heartbeat = new ScheduledThreadPoolExecutor(1, daemonFactory(name() + "-heartbeat-" + count.get())); - final long hb = heartbeatInterval().toMillis(); - this.heartbeat.scheduleAtFixedRate(this::heartbeatTick, hb, hb, TimeUnit.MILLISECONDS); - this.thread = createListenerThread(); - } - - /** - * Supply the executor used to run message handlers. Consumers must call this - * before the first {@link #addConsumer} — there is no default executor. - * Micronaut-managed consumers pass the injected {@code @Named(TaskExecutors.BLOCKING)} - * {@link ExecutorService}. The executor is never shut down by {@link #close()} - * (it is shared / container-managed). - * - * @param executor the shared handler executor; must not be {@code null} - */ - public void withHandlerExecutor(ExecutorService executor) { - this.pool = Objects.requireNonNull(executor, "Handler executor cannot be null"); - } - - private static ThreadFactory daemonFactory(String prefix) { - final AtomicInteger seq = new AtomicInteger(); - return runnable -> { - final Thread t = new Thread(runnable, prefix + "-" + seq.getAndIncrement()); - t.setDaemon(true); - return t; - }; - } - /** * Deserialize the message as string into the target message object and process it by applying * the given consumer {@link MessageConsumer}. @@ -372,40 +233,61 @@ private static ThreadFactory daemonFactory(String prefix) { * The message serialised as a string value * @param consumer * The consumer {@link MessageConsumer} that will handle the message as a object + * @param count + * An {@link AtomicInteger} counter incremented by one when this method is invoked, + * irrespective if the consumer is successful or not. * @return * The result of the consumer {@link MessageConsumer} operation. */ - protected boolean processMessage(String msg, MessageConsumer consumer) { + protected boolean processMessage(String msg, MessageConsumer consumer, AtomicInteger count) { + count.incrementAndGet(); final M decoded = encoder.decode(msg); log.trace("Message stream - receiving message={}; decoded={}", msg, decoded); return consumer.accept(decoded); } /** - * The dispatcher loop (runs on the listener thread). It never runs a handler itself: - * for every stream that has free pool capacity it polls one message (without acking) - * and submits its processing to the worker pool, then sleeps for {@link #pollInterval()} - * when nothing was polled this cycle. + * Run one consume cycle for the given stream and record the outcome on the + * {@link StreamMetrics} handle. The outcome is derived from the {@code count} + * delta (was the consumer lambda invoked?) and the return value of + * {@link MessageStream#consume}. + */ + private void consumeOne(String streamId, MessageConsumer consumer, AtomicInteger count) { + final long sample = metrics.startSample(); + final int countBefore = count.get(); + Outcome outcome = Outcome.EMPTY; + try { + final boolean accepted = stream.consume(streamId, (String msg) -> processMessage(msg, consumer, count)); + if (count.get() != countBefore) { + outcome = accepted ? Outcome.PROCESSED : Outcome.ACTIVE; + } + } + catch (Throwable t) { + outcome = Outcome.ERRORED; + throw t; + } + finally { + metrics.recordOutcome(sample, streamId, outcome); + } + } + + /** + * Process the messages as they are available from the underlying stream */ protected void processMessages() { - log.trace("Message stream - starting dispatcher thread"); + log.trace("Message stream - starting listener thread"); while (!Thread.currentThread().isInterrupted()) { try { - boolean polled = false; + final var count = new AtomicInteger(); for (Map.Entry> entry : listeners.entrySet()) { - // poll a stream only when a worker slot is free (backpressure); the - // permit is held for the whole lease lifetime so re-polls of in-flight - // commands are never starved by new intake - if (!slots.tryAcquire()) { - break; - } - // dispatchOne releases the permit itself when nothing is polled - polled = dispatchOne(entry.getKey()) || polled; + final var streamId = entry.getKey(); + final var consumer = entry.getValue(); + consumeOne(streamId, consumer, count); } // reset the attempt count because no error has been thrown attempt.reset(); - // if nothing was polled this cycle, sleep for a while before retrying - if (!polled) { + // if no message was sent, sleep for a while before retrying + if (count.get() == 0) { log.trace("Message stream - await before checking for new messages"); Thread.sleep(pollInterval().toMillis()); } @@ -421,218 +303,26 @@ protected void processMessages() { sleep(d0.toMillis()); } } - log.trace("Message stream - exiting dispatcher thread"); + log.trace("Message stream - exiting listener thread"); } /** - * Poll a single stream (a worker permit has already been acquired by the caller) and, - * if a message is available, register it as in-flight and submit it to the pool. - * If nothing is available the permit is released and {@code false} is returned. - * - * @return {@code true} if a message was polled and submitted, {@code false} otherwise - */ - private boolean dispatchOne(String streamId) { - boolean submitted = false; - try { - final MessageStream.Lease lease = stream.poll(streamId); - if (lease == null) { - metrics.recordOutcome(metrics.startSample(), streamId, Outcome.EMPTY); - return false; - } - final var e = new InFlight(streamId, lease.id(), lease.message()); - // Guard against self-reclaim: if the heartbeat falls behind by more than the - // claim-timeout, this instance's own poll() (XAUTOCLAIM) can re-deliver an entry it - // is already processing. The reclaim only refreshed the lease idle time, so keep the - // live in-flight entry and drop the duplicate — otherwise a second handler runs - // concurrently and its permit leaks (the original remove() returns null). - if (inFlight.putIfAbsent(e.key(), e) != null) { - return false; // 'submitted' stays false → finally releases this permit - } - submitRun(e); - submitted = true; - return true; - } - finally { - // the permit is held only once the lease is in flight; release it on an empty - // poll or an exception so the single acquire in the dispatcher stays balanced - if (!submitted) { - slots.release(); - } - } - } - - /** - * Submit the processing of an in-flight lease to the worker pool. Swallows the - * rejection that occurs when the pool is being shut down. - */ - private void submitRun(InFlight e) { - try { - pool.execute(() -> run(e)); - } - catch (RejectedExecutionException ex) { - log.debug("Message stream - worker pool rejected task for entry={} (shutting down)", e.key()); - } - } - - /** - * Runs a single {@code accept()} invocation on a worker thread. On {@code true} - * (terminal) it acks the message and drops the lease; on {@code false} (Model B, - * not-yet-terminal) it keeps the lease in-flight and schedules the next invocation - * after {@link #pollInterval()} — strictly serial per command, since the next - * invocation is scheduled only after this one returned. - */ - private void run(InFlight e) { - final boolean accepted = invokeHandler(e); - if (accepted) { - acknowledge(e); - } - else if (shouldRepoll(e)) { - scheduleRepoll(e); - } - } - - /** - * Run one {@code accept()} invocation on the worker thread, recording the metrics - * outcome. Returns {@code true} for a terminal result, {@code false} for - * not-yet-terminal or an error (both keep the lease for a later re-poll). - */ - private boolean invokeHandler(InFlight e) { - final MessageConsumer consumer = listeners.get(e.streamId()); - final long sample = metrics.startSample(); - boolean accepted = false; - Outcome outcome = Outcome.ACTIVE; - active.put(e.key(), System.currentTimeMillis()); - try { - accepted = processMessage(e.message(), consumer); - outcome = accepted ? Outcome.PROCESSED : Outcome.ACTIVE; - } - catch (Throwable t) { - outcome = Outcome.ERRORED; - log.error("Message stream - error processing entry={} - cause: {}", e.key(), t.getMessage(), t); - } - finally { - active.remove(e.key()); - metrics.recordOutcome(sample, e.streamId(), outcome); - } - return accepted; - } - - /** Terminal result: acknowledge the message and release its lease. */ - private void acknowledge(InFlight e) { - try { - stream.ack(e.streamId(), e.leaseId()); - } - catch (Throwable t) { - log.error("Message stream - error acking entry={} - cause: {}", e.key(), t.getMessage(), t); - } - finally { - releaseLease(e.key()); - } - } - - /** Whether a not-yet-terminal command should be re-polled: still owned and not shutting down. */ - private boolean shouldRepoll(InFlight e) { - return !closed && inFlight.containsKey(e.key()); - } - - /** - * Keep the lease (the heartbeat keeps renewing it, so no reclaim/migration) and schedule - * the next in-process invocation after {@link #pollInterval()} — strictly serial, since - * it is scheduled only after the previous invocation returned. - */ - private void scheduleRepoll(InFlight e) { - try { - scheduler.schedule(() -> submitRun(e), pollInterval().toMillis(), TimeUnit.MILLISECONDS); - } - catch (RejectedExecutionException ex) { - log.debug("Message stream - re-poll scheduler rejected entry={} (shutting down)", e.key()); - } - } - - /** - * Drop a lease from the in-flight set and free its capacity permit. This pair is the - * single invariant "a permit is held iff its key is in-flight"; returns {@code true} if - * this call performed the removal (so callers can log only a real eviction). - */ - private boolean releaseLease(String key) { - if (inFlight.remove(key) != null) { - slots.release(); - return true; - } - return false; - } - - /** - * Heartbeat tick: renew every in-flight lease so an alive consumer keeps ownership, - * and release the lease of any single invocation that has exceeded - * {@link #maxProcessingTime()} (safety valve; does not interrupt the handler thread). - */ - private void heartbeatTick() { - final long now = System.currentTimeMillis(); - final long maxMillis = maxProcessingTime().toMillis(); - for (InFlight e : inFlight.values()) { - final String key = e.key(); - final long start = active.getOrDefault(key, now); - if (now - start > maxMillis) { - // a single invocation is hung beyond the bound: stop renewing so the - // lease becomes reclaimable, and free its capacity permit - if (releaseLease(key)) { - log.warn("Message stream - releasing lease of hung entry={} after {} - reclaimable after claim timeout", - key, Duration.ofMillis(now - start)); - } - } - else { - try { - stream.renew(e.streamId(), e.leaseId()); - } - catch (Throwable t) { - // swallow transient errors; the next tick retries - log.warn("Message stream - error renewing lease for entry={} - cause: {}", key, t.getMessage()); - } - } - } - } - - /** - * Shutdown orderly the stream: stop the dispatcher, cancel pending re-polls, drain - * the worker pool so active handlers finish and ack, release any remaining leases so - * they are redelivered, and finally stop the heartbeat daemon. + * Shutdown orderly the stream */ @Override public void close() { if (thread == null) { return; } - closed = true; - // 1. stop the dispatcher + // interrupt the thread thread.interrupt(); + // wait for the termination try { thread.join(1_000); } catch (Exception e) { log.debug("Unexpected error while terminating {} - cause: {}", name0, e.getMessage()); } - // 2. cancel pending scheduled re-polls - if (scheduler != null) { - scheduler.shutdownNow(); - } - // 3. the handler executor is shared / container-managed — not shut down here; - // any active handler finishes on its own (short-lived) and acks - // 4. release any lease still held so it is redelivered without waiting for lapse - for (InFlight e : inFlight.values()) { - if (inFlight.remove(e.key()) != null) { - try { - stream.release(e.streamId(), e.leaseId()); - } - catch (Throwable t) { - log.debug("Message stream - error releasing entry={} on shutdown - cause: {}", e.key(), t.getMessage()); - } - } - } - // 5. stop the heartbeat daemon last (any remaining leases lapse -> peers reclaim) - if (heartbeat != null) { - heartbeat.shutdownNow(); - } } public int length(String streamId) { diff --git a/lib-data-stream-redis/src/main/java/io/seqera/data/stream/MessageStream.java b/lib-data-stream-redis/src/main/java/io/seqera/data/stream/MessageStream.java index 3af991e2..73dea79c 100644 --- a/lib-data-stream-redis/src/main/java/io/seqera/data/stream/MessageStream.java +++ b/lib-data-stream-redis/src/main/java/io/seqera/data/stream/MessageStream.java @@ -17,8 +17,6 @@ package io.seqera.data.stream; -import java.time.Duration; - /** * Interface for a distributed message stream that supports real-time event processing. * @@ -152,98 +150,7 @@ public interface MessageStream { * {@code false} if no message was available or processing failed * @see MessageConsumer#accept(Object) */ - default boolean consume(String streamId, MessageConsumer consumer) { - final Lease lease = poll(streamId); - if (lease == null) { - return false; - } - final boolean accepted = consumer.accept(lease.message()); - if (accepted) { - ack(streamId, lease.id()); - } - else { - release(streamId, lease.id()); - } - return accepted; - } - - /** - * A single delivered message paired with the token needed to renew, acknowledge - * or release it. The {@code id} is the stream-implementation specific handle - * (e.g. the Redis stream entry id) that identifies the delivered entry within - * its stream. - * - * @param the type of the delivered message - * @param id the implementation specific identifier of the delivered entry - * @param message the delivered message payload - */ - record Lease(String id, M message) {} - - /** - * Reads one message (either newly delivered or reclaimed from a stalled consumer) - * without acknowledging it. The caller becomes responsible for - * eventually calling {@link #ack(String, String)} once processing terminates, or - * {@link #release(String, String)} to hand it back for later redelivery. - * - * @param streamId the unique identifier of the source stream; must not be null or empty - * @return a {@link Lease} for the delivered message, or {@code null} if none is available - */ - Lease poll(String streamId); - - /** - * Resets the idle time of the given lease (heartbeat), so that an alive consumer - * keeps ownership of a message for as long as its handler runs. Implementations - * without a pending-entries list have no lease semantics and treat this as a no-op. - * - * @param streamId the unique identifier of the stream; must not be null or empty - * @param leaseId the identifier of the lease to renew - */ - void renew(String streamId, String leaseId); - - /** - * Acknowledges terminal processing of the given lease, removing the message from - * the stream so that it is never redelivered. - * - * @param streamId the unique identifier of the stream; must not be null or empty - * @param leaseId the identifier of the lease to acknowledge - */ - void ack(String streamId, String leaseId); - - /** - * Releases the given lease without acknowledging it, so that the message becomes - * available for redelivery later (a nack; used on shutdown). Implementations - * without a pending-entries list re-offer the message. - * - * @param streamId the unique identifier of the stream; must not be null or empty - * @param leaseId the identifier of the lease to release - */ - void release(String streamId, String leaseId); - - /** - * How often an in-flight lease must be renewed to retain ownership, so an alive - * consumer is never reclaimed by a peer while its handler is still running. The - * value is the implementation's own setting (e.g. {@code claim-timeout / 3} for a - * Redis consumer group) and MUST be shorter than the reclaim window. Returns - * {@code null} when the implementation has no lease concept (e.g. in-memory), in - * which case the caller uses its own default. - * - * @return the heartbeat interval, or {@code null} if the implementation has no lease - */ - default Duration heartbeatInterval() { - return null; - } - - /** - * Upper bound on a single {@code accept()} invocation before its lease is released - * (safety valve); it does not interrupt the handler thread. Returns {@code null} - * when the implementation has no lease concept, in which case the caller uses its - * own default. - * - * @return the maximum single-invocation processing time, or {@code null} - */ - default Duration maxProcessingTime() { - return null; - } + boolean consume(String streamId, MessageConsumer consumer); /** * Returns the approximate number of messages currently in the specified stream. diff --git a/lib-data-stream-redis/src/main/java/io/seqera/data/stream/impl/LocalMessageStream.java b/lib-data-stream-redis/src/main/java/io/seqera/data/stream/impl/LocalMessageStream.java index d678b347..0578e487 100644 --- a/lib-data-stream-redis/src/main/java/io/seqera/data/stream/impl/LocalMessageStream.java +++ b/lib-data-stream-redis/src/main/java/io/seqera/data/stream/impl/LocalMessageStream.java @@ -22,10 +22,12 @@ import io.micronaut.context.annotation.Requires; import io.seqera.activator.redis.RedisActivator; +import io.seqera.data.stream.MessageConsumer; import io.seqera.data.stream.MessageStream; import jakarta.inject.Singleton; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static io.seqera.data.stream.impl.SleepHelper.sleep; /** * In-memory implementation of {@link MessageStream} using Java {@link LinkedBlockingQueue} @@ -87,51 +89,33 @@ public void offer(String streamId, String message) { /** * {@inheritDoc} - * - *

Reads one message off the local queue. There is no pending-entries list, so - * the lease id is simply the message value itself (used to re-offer it on release). */ @Override - public Lease poll(String streamId) { + public boolean consume(String streamId, MessageConsumer consumer) { final var message = delegate .get(streamId) .poll(); if (message == null) { - return null; + return false; } - return new Lease<>(message, message); - } - - /** - * {@inheritDoc} - * - *

No pending-entries list ⇒ no lease semantics ⇒ no-op. - */ - @Override - public void renew(String streamId, String leaseId) { - // no-op: the local queue has no pending-entries list - } - - /** - * {@inheritDoc} - * - *

The message was already removed from the queue on {@link #poll(String)}, - * so acknowledgment is a no-op (the entry is simply dropped). - */ - @Override - public void ack(String streamId, String leaseId) { - // no-op: the entry was removed from the queue on poll - } - /** - * {@inheritDoc} - * - *

Re-offers the message onto the queue so it is redelivered later, mimicking the - * behavior of a Redis stream pending entry that is not acknowledged. - */ - @Override - public void release(String streamId, String leaseId) { - offer(streamId, leaseId); + Throwable error = null; + boolean result = false; + try { + result = consumer.accept(message); + } + catch (Throwable e) { + result = false; + log.debug("Failed to consume message from stream={} - cause: {}", streamId, e.getMessage(), e); + } + finally { + if (!result) { + // add again message not consumed to mimic the behavior or redis stream + sleep(1_000); + offer(streamId, message); + } + } + return result; } /** diff --git a/lib-data-stream-redis/src/main/java/io/seqera/data/stream/impl/RedisMessageStream.java b/lib-data-stream-redis/src/main/java/io/seqera/data/stream/impl/RedisMessageStream.java index f0064991..4a3e4b30 100644 --- a/lib-data-stream-redis/src/main/java/io/seqera/data/stream/impl/RedisMessageStream.java +++ b/lib-data-stream-redis/src/main/java/io/seqera/data/stream/impl/RedisMessageStream.java @@ -37,7 +37,6 @@ import redis.clients.jedis.StreamEntryID; import redis.clients.jedis.exceptions.JedisDataException; import redis.clients.jedis.params.XAutoClaimParams; -import redis.clients.jedis.params.XClaimParams; import redis.clients.jedis.params.XReadGroupParams; import redis.clients.jedis.resps.StreamEntry; @@ -143,117 +142,33 @@ public void offer(String streamId, String message) { /** * {@inheritDoc} - * - *

Reads one entry (a reclaimed stalled one via {@code XAUTOCLAIM}, otherwise a - * newly delivered one via {@code XREADGROUP >}) without acking it. - * The returned lease id is the Redis {@link StreamEntryID} of the delivered entry. */ @Override - public Lease poll(String streamId) { + public boolean consume(String streamId, MessageConsumer consumer) { try (Jedis jedis = pool.getResource()) { + String msg; + final long begin = System.currentTimeMillis(); StreamEntry entry = claimMessage(jedis, streamId); if (entry == null) { entry = readMessage(jedis, streamId); } - if (entry == null) { - return null; + if (entry != null && consumer.accept(msg = entry.getFields().get(DATA_FIELD))) { + final var tx = jedis.multi(); + // acknowledge the entry has been processed so that it cannot be claimed anymore + tx.xack(streamId, config.getDefaultConsumerGroupName(), entry.getID()); + final var delta = System.currentTimeMillis() - begin; + if (delta > config.getConsumerWarnTimeoutMillis()) { + log.warn("Redis message stream - consume processing took {} - offending entry={}; message={}", + Duration.ofMillis(delta), entry.getID(), msg); + } + // this remove permanently the entry from the stream + tx.xdel(streamId, entry.getID()); + tx.exec(); + return true; } - return new Lease<>(entry.getID().toString(), entry.getFields().get(DATA_FIELD)); - } - } - - /** - * {@inheritDoc} - * - *

Resets the idle time of the entry to zero by re-claiming it to this same - * consumer with a {@code min-idle} of {@code 0} using {@code XCLAIM … JUSTID}, - * so an alive consumer keeps ownership of the message regardless of how long the - * handler runs. - */ - @Override - public void renew(String streamId, String leaseId) { - try (Jedis jedis = pool.getResource()) { - jedis.xclaimJustId( - streamId, - config.getDefaultConsumerGroupName(), - consumerName, - 0L, - XClaimParams.xClaimParams(), - new StreamEntryID(leaseId)); - } - } - - /** - * {@inheritDoc} - * - *

Acknowledges the entry ({@code XACK}) and permanently removes it from the - * stream ({@code XDEL}) atomically so it can neither be claimed nor redelivered. - */ - @Override - public void ack(String streamId, String leaseId) { - final var id = new StreamEntryID(leaseId); - try (Jedis jedis = pool.getResource()) { - final var tx = jedis.multi(); - // acknowledge the entry has been processed so that it cannot be claimed anymore - tx.xack(streamId, config.getDefaultConsumerGroupName(), id); - // this removes permanently the entry from the stream - tx.xdel(streamId, id); - tx.exec(); - } - } - - /** - * {@inheritDoc} - * - *

No-op: the entry remains in the pending-entries list and becomes reclaimable - * by a peer consumer once its idle time exceeds {@code claim-timeout}. - */ - @Override - public void release(String streamId, String leaseId) { - // no-op: entry stays in the PEL, reclaimable after claim-timeout - } - - /** - * {@inheritDoc} - * - *

Derived from the configured {@code claim-timeout} so an alive consumer's lease - * is renewed well before a peer could reclaim it. - */ - @Override - public Duration heartbeatInterval() { - return config.getHeartbeatInterval(); - } - - /** - * {@inheritDoc} - */ - @Override - public Duration maxProcessingTime() { - return config.getMaxProcessingTime(); - } - - /** - * {@inheritDoc} - */ - @Override - public boolean consume(String streamId, MessageConsumer consumer) { - final long begin = System.currentTimeMillis(); - final Lease lease = poll(streamId); - if (lease == null) { - return false; - } - if (consumer.accept(lease.message())) { - ack(streamId, lease.id()); - final var delta = System.currentTimeMillis() - begin; - if (delta > config.getConsumerWarnTimeoutMillis()) { - log.warn("Redis message stream - consume processing took {} - offending entry={}; message={}", - Duration.ofMillis(delta), lease.id(), lease.message()); + else { + return false; } - return true; - } - else { - release(streamId, lease.id()); - return false; } } diff --git a/lib-data-stream-redis/src/main/java/io/seqera/data/stream/impl/RedisStreamConfig.java b/lib-data-stream-redis/src/main/java/io/seqera/data/stream/impl/RedisStreamConfig.java index dc6c19b0..56c312cb 100644 --- a/lib-data-stream-redis/src/main/java/io/seqera/data/stream/impl/RedisStreamConfig.java +++ b/lib-data-stream-redis/src/main/java/io/seqera/data/stream/impl/RedisStreamConfig.java @@ -93,48 +93,4 @@ default long getClaimTimeoutMillis() { default long getConsumerWarnTimeoutMillis() { return getConsumerWarnTimeout().toMillis(); } - - /** - * Returns how often in-flight leases are renewed (heartbeated) to keep them - * from being reclaimed by peer consumers while a handler is still running. - * Must be shorter than {@link #getClaimTimeout()}; defaults to {@code claim-timeout / 3} - * so that up to two consecutive misses are tolerated. - * - * @return the heartbeat interval duration - */ - default Duration getHeartbeatInterval() { - return getClaimTimeout().dividedBy(3); - } - - /** - * Returns the heartbeat interval in milliseconds for convenience. - * This is a derived value from {@link #getHeartbeatInterval()}. - * - * @return the heartbeat interval in milliseconds - */ - default long getHeartbeatIntervalMillis() { - return getHeartbeatInterval().toMillis(); - } - - /** - * Returns the upper bound on a single {@code accept()} invocation before its - * lease is released (safety valve). This bounds one handler invocation, not the - * total lease lifetime; past this bound the heartbeat daemon stops renewing the - * lease so it becomes reclaimable. Defaults to {@code 15m}. - * - * @return the maximum single-invocation processing time duration - */ - default Duration getMaxProcessingTime() { - return Duration.ofMinutes(15); - } - - /** - * Returns the maximum processing time in milliseconds for convenience. - * This is a derived value from {@link #getMaxProcessingTime()}. - * - * @return the maximum processing time in milliseconds - */ - default long getMaxProcessingTimeMillis() { - return getMaxProcessingTime().toMillis(); - } } diff --git a/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/AsyncStreamLocalTest.groovy b/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/AsyncStreamLocalTest.groovy deleted file mode 100644 index dc493245..00000000 --- a/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/AsyncStreamLocalTest.groovy +++ /dev/null @@ -1,253 +0,0 @@ -/* - * Copyright 2026, Seqera Labs - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ - -package io.seqera.data.stream - -import java.time.Duration -import java.util.concurrent.ConcurrentLinkedQueue -import java.util.concurrent.CountDownLatch -import java.util.concurrent.TimeUnit -import java.util.concurrent.atomic.AtomicInteger - -import io.seqera.data.stream.impl.LocalMessageStream -import io.seqera.random.LongRndKey -import spock.lang.Specification -import spock.util.concurrent.PollingConditions - -/** - * Async-processing behaviour of {@link AbstractMessageStream} exercised over the - * in-memory {@link LocalMessageStream} backend, so these run WITHOUT Docker. - * - * Covers spec §10 tests: 1 (non-blocking), 2 (concurrency), 5 (re-poll cadence), - * 6 (serial per command), 8 (backpressure), 9 (concurrency==1 default). - * - * @author Paolo Di Tommaso - */ -class AsyncStreamLocalTest extends Specification { - - // §10.1 — a slow handler on stream A must not delay a fast handler on stream B - def 'should not block a fast stream behind a slow one' () { - given: - def target = new LocalMessageStream() - def stream = new TunableStream(target, concurrency: 2, pollInterval: Duration.ofMillis(100)) - def idA = "stream-${LongRndKey.rndHex()}" - def idB = "stream-${LongRndKey.rndHex()}" - def slowDone = new CountDownLatch(1) - def fastDone = new CountDownLatch(1) - - when: - stream.addConsumer(idA, { msg -> Thread.sleep(3_000); slowDone.countDown(); true }) - stream.addConsumer(idB, { msg -> fastDone.countDown(); true }) - and: - stream.offer(idA, 'slow') - stream.offer(idB, 'fast') - - then: - // the fast handler completes well before the slow one finishes - fastDone.await(2, TimeUnit.SECONDS) - slowDone.count == 1 - - cleanup: - stream.close() - } - - // §10.2 — N messages with a slow handler complete in ~max(handler), not ~sum - def 'should process messages concurrently' () { - given: - def target = new LocalMessageStream() - def stream = new TunableStream(target, concurrency: 4, pollInterval: Duration.ofMillis(100)) - def id = "stream-${LongRndKey.rndHex()}" - def done = new CountDownLatch(4) - - when: - stream.addConsumer(id, { msg -> Thread.sleep(500); done.countDown(); true }) - def t0 = System.currentTimeMillis() - 4.times { stream.offer(id, "msg-$it".toString()) } - - then: - done.await(5, TimeUnit.SECONDS) - def elapsed = System.currentTimeMillis() - t0 - // 4 x 500ms serial would be ~2000ms; concurrent should be well under that - elapsed < 1_500 - - cleanup: - stream.close() - } - - // §10.5 — a not-yet-terminal command is re-invoked at ~pollInterval (Model B) - def 'should re-poll a not-yet-terminal command at poll interval' () { - given: - def poll = Duration.ofMillis(300) - def target = new LocalMessageStream() - def stream = new TunableStream(target, concurrency: 1, pollInterval: poll) - def id = "stream-${LongRndKey.rndHex()}" - def timestamps = new ConcurrentLinkedQueue() - - when: - // record the wall-clock of each invocation; stay non-terminal for 5 calls, then ack - stream.addConsumer(id, { msg -> - timestamps.add(System.currentTimeMillis()) - return timestamps.size() >= 5 - }) - stream.offer(id, 'running') - - then: - new PollingConditions(timeout: 10).eventually { - assert timestamps.size() == 5 - } - and: - def times = timestamps.toList() - def gaps = (1..= 150 && it <= 1_500 } - - cleanup: - stream.close() - } - - // §10.6 — never two concurrent accept() invocations for the same command - def 'should invoke a command strictly serially across re-polls' () { - given: - def target = new LocalMessageStream() - def stream = new TunableStream(target, concurrency: 4, pollInterval: Duration.ofMillis(150)) - def id = "stream-${LongRndKey.rndHex()}" - def inProgress = new AtomicInteger() - def maxConcurrent = new AtomicInteger() - def calls = new AtomicInteger() - - when: - stream.addConsumer(id, { msg -> - def now = inProgress.incrementAndGet() - maxConcurrent.accumulateAndGet(now, Math::max) - Thread.sleep(100) - inProgress.decrementAndGet() - return calls.incrementAndGet() >= 4 - }) - stream.offer(id, 'running') - - then: - new PollingConditions(timeout: 10).eventually { - assert calls.get() >= 4 - } - and: - // one message => the same lease is never processed by two workers at once - maxConcurrent.get() == 1 - - cleanup: - stream.close() - } - - // §10.8 — with pool size K and more than K ready messages, at most K run at once - def 'should bound concurrent handlers by the pool size (backpressure)' () { - given: - def target = new LocalMessageStream() - def stream = new TunableStream(target, concurrency: 2, pollInterval: Duration.ofMillis(100)) - def id = "stream-${LongRndKey.rndHex()}" - def inProgress = new AtomicInteger() - def maxConcurrent = new AtomicInteger() - def done = new CountDownLatch(6) - - when: - stream.addConsumer(id, { msg -> - def now = inProgress.incrementAndGet() - maxConcurrent.accumulateAndGet(now, Math::max) - Thread.sleep(300) - inProgress.decrementAndGet() - done.countDown() - true - }) - 6.times { stream.offer(id, "msg-$it".toString()) } - - then: - done.await(10, TimeUnit.SECONDS) - maxConcurrent.get() <= 2 - - cleanup: - stream.close() - } - - // §10.9 — default concurrency is 1: at most one handler runs at a time - def 'should run at most one handler with the default concurrency' () { - given: - def target = new LocalMessageStream() - // default TunableStream -> concurrency 1 - def stream = new TunableStream(target, pollInterval: Duration.ofMillis(100)) - def id = "stream-${LongRndKey.rndHex()}" - def inProgress = new AtomicInteger() - def maxConcurrent = new AtomicInteger() - def done = new CountDownLatch(4) - - when: - stream.addConsumer(id, { msg -> - def now = inProgress.incrementAndGet() - maxConcurrent.accumulateAndGet(now, Math::max) - Thread.sleep(150) - inProgress.decrementAndGet() - done.countDown() - true - }) - 4.times { stream.offer(id, "msg-$it".toString()) } - - then: - done.await(10, TimeUnit.SECONDS) - maxConcurrent.get() == 1 - - cleanup: - stream.close() - } - - // self-reclaim: if the heartbeat falls behind, this instance's own poll() (XAUTOCLAIM) can - // re-deliver an entry it is still processing. That duplicate must NOT start a second handler - // or leak a permit (regression for the concurrency()>1 permit-leak / double-run). - def 'self-reclaim of an in-flight entry does not double-run the handler'() { - given: 'a backing stream that re-delivers the SAME lease id twice, then nothing' - def deliveries = new AtomicInteger(0) - def acks = new AtomicInteger(0) - def target = [ - init : { String q -> }, - offer : { String q, String m -> }, - poll : { String q -> deliveries.getAndIncrement() < 2 ? new MessageStream.Lease('dup-id', 'payload') : null }, - renew : { String q, String id -> }, - ack : { String q, String id -> acks.incrementAndGet() }, - release: { String q, String id -> }, - length : { String q -> 0 } - ] as MessageStream - def stream = new TunableStream(target, concurrency: 2, pollInterval: Duration.ofMillis(50)) - def runs = new AtomicInteger(0) - def gate = new CountDownLatch(1) - - when: 'the handler blocks, so the entry stays in flight across the duplicate delivery' - stream.addConsumer('q1', { msg -> runs.incrementAndGet(); gate.await(5, TimeUnit.SECONDS); true } as MessageConsumer) - and: 'wait until the duplicate delivery has been attempted, then let a stray 2nd run surface' - new PollingConditions(timeout: 3).eventually { deliveries.get() >= 2 } - sleep(300) - - then: 'the handler ran exactly once despite the duplicate delivery' - runs.get() == 1 - - when: 'the handler completes' - gate.countDown() - - then: 'the entry is acked exactly once (no permit leak / double-run)' - new PollingConditions(timeout: 5).eventually { acks.get() == 1 } - - cleanup: - gate.countDown() - stream.close() - } - -} diff --git a/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/AsyncStreamRedisTest.groovy b/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/AsyncStreamRedisTest.groovy deleted file mode 100644 index 0c17f1fa..00000000 --- a/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/AsyncStreamRedisTest.groovy +++ /dev/null @@ -1,209 +0,0 @@ -/* - * Copyright 2026, Seqera Labs - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ - -package io.seqera.data.stream - -import java.time.Duration -import java.util.concurrent.CountDownLatch -import java.util.concurrent.TimeUnit -import java.util.concurrent.atomic.AtomicInteger - -import io.micronaut.context.ApplicationContext -import io.seqera.data.stream.impl.RedisMessageStream -import io.seqera.fixtures.redis.RedisTestContainer -import io.seqera.random.LongRndKey -import spock.lang.Specification -import spock.util.concurrent.PollingConditions - -/** - * Testcontainers-backed verification of the async lease model against a real Redis - * (consumer-group PEL semantics). Covers spec §10 tests: 3 (no reclaim of live work, - * single- and two-instance), 4 (crash failover), 7 (max-processing-time safety valve). - * - * The test config sets {@code claim-timeout = 1s}; the streams below heartbeat every - * 300ms so an alive owner keeps its lease. - * - * @author Paolo Di Tommaso - */ -class AsyncStreamRedisTest extends Specification implements RedisTestContainer { - - private ApplicationContext newContext() { - // all contexts read the same redis.host/redis.port system properties set by the - // RedisTestContainer trait, so they share one Redis but each gets its own - // RedisMessageStream bean (distinct consumer name) => independent instances - return ApplicationContext.run('test', 'redis') - } - - // §10.3 (single instance) — a handler running longer than claim-timeout is not - // reclaimed by this instance's own poll while it is alive/heartbeated - def 'should not reclaim live work within a single instance' () { - given: - def ctx = newContext() - def target = ctx.getBean(RedisMessageStream) - // concurrency 2 keeps the dispatcher polling while the one message is in-flight - def stream = new TunableStream(target, - concurrency: 2, - pollInterval: Duration.ofMillis(200), - heartbeatInterval: Duration.ofMillis(300)) - def id = "stream-${LongRndKey.rndHex()}" - def calls = new AtomicInteger() - def done = new CountDownLatch(1) - - when: - // handler runs ~3s (>> claim-timeout 1s); the lease is heartbeated so it is never - // reclaimed and the dispatcher's own poll never re-delivers it - stream.addConsumer(id, { msg -> - calls.incrementAndGet() - Thread.sleep(3_000) - done.countDown() - true - }) - stream.offer(id, 'long-running') - - then: - done.await(8, TimeUnit.SECONDS) - and: - // give any spurious re-delivery a chance to show up, then assert single execution - sleep 1_000 - calls.get() == 1 - - cleanup: - stream.close() - ctx.stop() - } - - // §10.3 (two instances) — a live, heartbeated owner is not reclaimed by a peer - def 'should not reclaim live work across two instances' () { - given: - def ctxA = newContext() - def ctxB = newContext() - def targetA = ctxA.getBean(RedisMessageStream) - def targetB = ctxB.getBean(RedisMessageStream) - def streamA = new TunableStream(targetA, - concurrency: 1, pollInterval: Duration.ofMillis(200), heartbeatInterval: Duration.ofMillis(300)) - def streamB = new TunableStream(targetB, - concurrency: 1, pollInterval: Duration.ofMillis(200), heartbeatInterval: Duration.ofMillis(300)) - def id = "stream-${LongRndKey.rndHex()}" - // shared across both instances: total number of times the message is processed - def calls = new AtomicInteger() - def done = new CountDownLatch(1) - - when: - def handler = { msg -> - calls.incrementAndGet() - Thread.sleep(3_000) // > claim-timeout, but heartbeated -> no reclaim by peer - done.countDown() - true - } - streamA.addConsumer(id, handler) - streamB.addConsumer(id, handler) - streamA.offer(id, 'once') - - then: - done.await(8, TimeUnit.SECONDS) - and: - sleep 1_500 // longer than claim-timeout, let any duplicate reclaim surface - calls.get() == 1 - - cleanup: - streamA.close() - streamB.close() - ctxA.stop() - ctxB.stop() - } - - // §10.4 — a non-heartbeating (crashed) owner's message is reclaimed by a peer after - // claim-timeout and processed there - def 'should fail over to a peer when the owner stops heartbeating' () { - given: - def ctxDead = newContext() - def ctxLive = newContext() - def targetDead = ctxDead.getBean(RedisMessageStream) - def targetLive = ctxLive.getBean(RedisMessageStream) - // 'dead' owner: picks up the message and hangs, and never heartbeats (interval 1h) - // so its lease idle-time grows and becomes reclaimable after claim-timeout - def streamDead = new TunableStream(targetDead, - concurrency: 1, pollInterval: Duration.ofMillis(200), heartbeatInterval: Duration.ofHours(1)) - def streamLive = new TunableStream(targetLive, - concurrency: 1, pollInterval: Duration.ofMillis(200), heartbeatInterval: Duration.ofMillis(300)) - def id = "stream-${LongRndKey.rndHex()}" - def hang = new CountDownLatch(1) - def deadStarted = new CountDownLatch(1) - def processedByLive = new CountDownLatch(1) - - when: 'only the dead owner is consuming, so it is guaranteed to pick up the message' - streamDead.addConsumer(id, { msg -> deadStarted.countDown(); hang.await(); true }) - streamDead.offer(id, 'orphan') - - then: 'the dead owner picks it up and then hangs without heartbeating' - deadStarted.await(5, TimeUnit.SECONDS) - - when: 'a live peer joins the group' - streamLive.addConsumer(id, { msg -> processedByLive.countDown(); true }) - - then: 'it reclaims the orphaned entry after the claim timeout and processes it' - processedByLive.await(8, TimeUnit.SECONDS) - - cleanup: - hang.countDown() - streamDead.close() - streamLive.close() - ctxDead.stop() - ctxLive.stop() - } - - // §10.7 — a single invocation exceeding max-processing-time has its lease released - // (stops being renewed), so the message is reclaimed and re-delivered - def 'should release the lease of an invocation exceeding max-processing-time' () { - given: - def ctx = newContext() - def target = ctx.getBean(RedisMessageStream) - def stream = new TunableStream(target, - concurrency: 2, - pollInterval: Duration.ofMillis(200), - heartbeatInterval: Duration.ofMillis(300), - maxProcessingTime: Duration.ofSeconds(1)) - def id = "stream-${LongRndKey.rndHex()}" - def calls = new AtomicInteger() - def hang = new CountDownLatch(1) - def redelivered = new CountDownLatch(1) - - when: - stream.addConsumer(id, { msg -> - def n = calls.incrementAndGet() - if (n == 1) { - // first invocation hangs past max-processing-time (1s) -> lease released - hang.await() - return true - } - // the re-delivered invocation completes normally - redelivered.countDown() - return true - }) - stream.offer(id, 'hung') - - then: 'the hung invocation is evicted and the message is re-delivered' - redelivered.await(10, TimeUnit.SECONDS) - calls.get() >= 2 - - cleanup: - hang.countDown() - stream.close() - ctx.stop() - } - -} diff --git a/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/LocalMessageStreamTest.groovy b/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/LocalMessageStreamTest.groovy index c0662fda..82662565 100644 --- a/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/LocalMessageStreamTest.groovy +++ b/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/LocalMessageStreamTest.groovy @@ -69,19 +69,14 @@ class LocalMessageStreamTest extends Specification { then: stream.consume(id1, { it-> it=='alpha'}) and: - // the default consume() does not catch handler exceptions: it propagates and, - // since poll() already removed 'delta' and release() is not reached, it is dropped - try { - stream.consume(id1, { it-> throw new RuntimeException("Oops")}) - assert false - } - catch (RuntimeException e) { - assert e.message == 'Oops' - } + !stream.consume(id1, { it-> throw new RuntimeException("Oops")}) and: - // next message is 'gamma' as expected ('delta' was dropped on the throw) + // next message is 'gamma' as expected stream.consume(id1, { it-> it=='gamma'}) and: + // now the errored message is available again + stream.consume(id1, { it-> it=='delta'}) + and: !stream.consume(id1, { it-> assert false /* <-- this should not be invoked */ }) when: @@ -112,71 +107,4 @@ class LocalMessageStreamTest extends Specification { stream.length(id1) == 2 } - // §10.10 — Local backend: poll returns a lease; renew is a no-op; release re-offers - def 'should poll and release re-offering the message' () { - given: - def id1 = "stream-${LongRndKey.rndHex()}" - def stream = new LocalMessageStream() - stream.init(id1) - stream.offer(id1, 'alpha') - - when: 'poll takes the message off the queue (lease id == message value)' - def lease = stream.poll(id1) - then: - lease != null - lease.message() == 'alpha' - stream.length(id1) == 0 - - when: 'renew is a no-op and does not throw nor alter the queue' - stream.renew(id1, lease.id()) - then: - stream.length(id1) == 0 - - when: 'release re-offers the message for later redelivery' - stream.release(id1, lease.id()) - then: - stream.length(id1) == 1 - stream.poll(id1).message() == 'alpha' - } - - def 'should ack by dropping the polled message' () { - given: - def id1 = "stream-${LongRndKey.rndHex()}" - def stream = new LocalMessageStream() - stream.init(id1) - stream.offer(id1, 'alpha') - - when: - def lease = stream.poll(id1) - stream.ack(id1, lease.id()) - then: - // ack is a no-op (already removed on poll) and nothing is redelivered - stream.length(id1) == 0 - stream.poll(id1) == null - } - - // §10.11 — default consume() acks on true (message gone) / releases on false (redelivered) - def 'should ack on true and release on false via default consume()' () { - given: - def id1 = "stream-${LongRndKey.rndHex()}" - def stream = new LocalMessageStream() - stream.init(id1) - stream.offer(id1, 'keep-me') - - when: 'consumer returns false -> message is released and stays available' - def r1 = stream.consume(id1, { it -> false }) - then: - !r1 - stream.length(id1) == 1 - - when: 'consumer returns true -> message is acked and removed' - def r2 = stream.consume(id1, { it -> it == 'keep-me' }) - then: - r2 - stream.length(id1) == 0 - and: - // nothing left to consume - !stream.consume(id1, { it -> assert false /* not invoked */ }) - } - } diff --git a/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/TestStream.groovy b/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/TestStream.groovy index 05ab00ce..e897684b 100644 --- a/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/TestStream.groovy +++ b/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/TestStream.groovy @@ -34,12 +34,10 @@ class TestStream extends AbstractMessageStream { TestStream(MessageStream target) { super(target) - withHandlerExecutor(TestWorkerPool.INSTANCE) } TestStream(MessageStream target, StreamMetrics metrics) { super(target, metrics) - withHandlerExecutor(TestWorkerPool.INSTANCE) } static TestStream withRegistry(MessageStream target, MeterRegistry registry) { diff --git a/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/TunableStream.groovy b/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/TunableStream.groovy deleted file mode 100644 index 16c3c31b..00000000 --- a/lib-data-stream-redis/src/test/groovy/io/seqera/data/stream/TunableStream.groovy +++ /dev/null @@ -1,82 +0,0 @@ -/* - * Copyright 2026, Seqera Labs - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ - -package io.seqera.data.stream - -import java.time.Duration - -import io.seqera.serde.encode.StringEncodingStrategy - -/** - * A {@link AbstractMessageStream} used by the async-processing tests. It carries a - * String payload (identity encoding) and exposes the async knobs — concurrency, - * poll interval, heartbeat interval and max-processing-time — as constructor options - * so each test can tune them independently. - * - * @author Paolo Di Tommaso - */ -class TunableStream extends AbstractMessageStream { - - private final int workers - private final Duration pollDelay - private final Duration hbInterval - private final Duration maxProcTime - - TunableStream(Map opts = [:], MessageStream target) { - super(target) - withHandlerExecutor(TestWorkerPool.INSTANCE) - this.workers = (opts.concurrency ?: 1) as int - this.pollDelay = (opts.pollInterval ?: Duration.ofSeconds(1)) as Duration - this.hbInterval = (opts.heartbeatInterval ?: Duration.ofSeconds(20)) as Duration - this.maxProcTime = (opts.maxProcessingTime ?: Duration.ofMinutes(15)) as Duration - } - - @Override - protected StringEncodingStrategy createEncodingStrategy() { - return new StringEncodingStrategy() { - @Override - String encode(String message) { return message } - @Override - String decode(String encoded) { return encoded } - } - } - - @Override - protected String name() { - return 'tunable-stream' - } - - @Override - protected Duration pollInterval() { - return pollDelay - } - - @Override - protected int concurrency() { - return workers - } - - @Override - protected Duration heartbeatInterval() { - return hbInterval - } - - @Override - protected Duration maxProcessingTime() { - return maxProcTime - } -} diff --git a/lib-data-stream-redis/src/test/java/io/seqera/data/stream/TestPlainStream.java b/lib-data-stream-redis/src/test/java/io/seqera/data/stream/TestPlainStream.java index 71370eee..040a25b8 100644 --- a/lib-data-stream-redis/src/test/java/io/seqera/data/stream/TestPlainStream.java +++ b/lib-data-stream-redis/src/test/java/io/seqera/data/stream/TestPlainStream.java @@ -33,7 +33,6 @@ public class TestPlainStream extends AbstractMessageStream { public TestPlainStream(MessageStream target) { super(target); - withHandlerExecutor(TestWorkerPool.INSTANCE); } @Override diff --git a/lib-data-stream-redis/src/test/java/io/seqera/data/stream/TestWorkerPool.java b/lib-data-stream-redis/src/test/java/io/seqera/data/stream/TestWorkerPool.java deleted file mode 100644 index 3ad8ad77..00000000 --- a/lib-data-stream-redis/src/test/java/io/seqera/data/stream/TestWorkerPool.java +++ /dev/null @@ -1,36 +0,0 @@ -/* - * Copyright 2026, Seqera Labs - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ - -package io.seqera.data.stream; - -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; - -/** - * Shared daemon handler executor for tests. {@link AbstractMessageStream} no longer ships a - * built-in default executor (handlers must be supplied via {@code withHandlerExecutor}), so the - * test fixtures inject this one. Daemon threads so it never keeps the test JVM alive. - */ -public final class TestWorkerPool { - private TestWorkerPool() {} - - public static final ExecutorService INSTANCE = Executors.newCachedThreadPool(r -> { - Thread t = new Thread(r, "test-handler"); - t.setDaemon(true); - return t; - }); -}