From 4afd8c58808c7873d7c9d08cdd6e0f5f81584b94 Mon Sep 17 00:00:00 2001 From: Paolo Di Tommaso Date: Fri, 31 Jul 2026 10:53:04 +0200 Subject: [PATCH 1/2] fix(cmd-queue): retry a command whose handler throws MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restores the fix originally released as 0.5.1 (6c7b171, #87) and dropped by the revert to 0.4.0 in #100. Same hunk, re-cut on top of the 0.4.0 tree. No VERSION or changelog change — code and tests only. The catch in CommandServiceImpl.processCommandWithHandler treated any escaping exception as a terminal command outcome: it persisted a FAILED CommandState and returned true, so RedisMessageStream.consume acked and deleted the entry. There was no retry. That conflates "the handler threw" with "the command failed", and it fails asymmetrically because the two live in different stores. Command state is in Redis; the domain work is in Postgres. When Micronaut closes the HikariCP pool while the queue is still draining, the handler throws a JDBC error and the catch records a permanent verdict using the store that still works, about a failure caused by the store that does not — while the domain entity was never transitioned. Queue empty, command FAILED, entity dangling, nothing left to advance it, polling clients hanging (seqeralabs/sched#712). Fix: log and return false. The entry stays unacked in the PEL for XAUTOCLAIM to hand to a live consumer, and since the catch never persists started(), status stays SUBMITTED so the next delivery re-enters execute(). A genuine failure is signalled by returning a FAILED CommandResult, which the terminal branch above already handles. Independent of the async/heartbeat-lease model that #100 removed: this works on the synchronous 1.5.0 stream because consume() only xacks/xdels when the consumer returns true. Unchanged: a returned FAILED CommandResult is still terminal, and an unknown command type is still failed and acked (no handler exists to retry) — both covered by existing tests. Test: the spec from #87, verified to fail against the 0.4.0 catch and pass with the fix. Co-Authored-By: Claude Opus 5 (1M context) --- .../data/command/CommandServiceImpl.java | 17 +++++++++--- .../data/command/CommandServiceTest.groovy | 27 +++++++++++++++++++ 2 files changed, 40 insertions(+), 4 deletions(-) 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 aebfe7a6..06eff28e 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 @@ -47,6 +47,9 @@ * *
  • If result is RUNNING → return false (message stays in queue for retry)
    • *
  • If result is terminal → 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)
    • * */ @Singleton @@ -290,10 +293,16 @@ private boolean processCommandWithHandler( return true; // Remove from queue - processing complete } catch (Exception e) { - // 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 + // 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); + return false; // Keep in queue - redelivered / re-polled } } 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 3bcefb20..fbcfac60 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 @@ -27,6 +27,7 @@ import spock.lang.Specification import java.time.Duration import java.time.Instant +import java.util.concurrent.atomic.AtomicInteger /** * End-to-end tests for the CommandService. @@ -172,6 +173,23 @@ 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' + } + def 'should handle unknown command type'() { given: def params = new TestParams(42, 'fast') @@ -239,6 +257,7 @@ class TestCommand implements Command { class TestCommandHandler implements CommandHandler { private Instant startTime + private final AtomicInteger flakyAttempts = new AtomicInteger() @Override String type() { 'test' } @@ -251,6 +270,14 @@ 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 == 'slow') { startTime = Instant.now() return CommandResult.running() From 53acfcdda423bacb926518bd83d852ba8acdc87b Mon Sep 17 00:00:00 2001 From: Paolo Di Tommaso Date: Fri, 31 Jul 2026 11:02:17 +0200 Subject: [PATCH 2/2] feat(cmd-queue): track command processing errors on CommandState MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Recovers the error tracking originally released as 0.6.0 (e54229e, #89), dropped by the revert to 0.4.0 in #100. No VERSION or changelog change — code and tests only. Companion to the retry fix in the parent commit: once a thrown handler is retried instead of terminal-failed, a command can retry indefinitely with nothing recording that it is happening. These fields make that visible. - errorsCount: consecutive processing errors since the last successful processing - modifiedAt: last-write timestamp - error: now also carries the message of a transient (non-terminal) processing error. It holds the most recent message, transient or terminal; a terminal failure is identified by status == FAILED, not by error being non-null. recordError is best-effort — a failed write is logged and never changes control flow, so the command is still kept in the queue and retried. The streak is reset on recovery, with a single write and only when there is something to reset, so healthy re-polls stay write-free. Backward-compatible: the new fields default to 0/null when older serialized state is read. One deliberate adaptation from e54229e, required by this tree: 0.4.x still has executeWithTimeout, which wraps a handler exception in a generic RuntimeException("Command execution failed"). #89 was written against #84, which had removed that method, so recording e.getMessage() verbatim was correct there but here would stamp every transient error on the execute() path with the same useless string. recordError now records the root cause's message via rootMessage(), which is also correct for the checkStatus() path where the exception propagates directly. Caught by #89's own test asserting error == 'Persistent boom'; it failed with 'Command execution failed' before the adaptation. Tests: the two specs from #89 plus its CommandState serialization coverage. Co-Authored-By: Claude Opus 5 (1M context) --- .../data/command/CommandServiceImpl.java | 35 ++++++++++ .../io/seqera/data/command/CommandState.java | 66 +++++++++++++++++-- .../data/command/CommandServiceTest.groovy | 24 +++++++ .../CommandStateSerializationTest.groovy | 33 ++++++++++ 4 files changed, 151 insertions(+), 7 deletions(-) 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 06eff28e..698b1e11 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 @@ -281,6 +281,10 @@ private boolean processCommandWithHandler( // 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 - will retry and call checkStatus() } @@ -302,6 +306,7 @@ private boolean processCommandWithHandler( // 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 } } @@ -342,4 +347,34 @@ private CommandResult executeWithTimeout(CommandHandler handler, throw new RuntimeException("Command execution failed", e); } } + + /** + * 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. + */ + private void recordError(CommandState state, Exception e) { + try { + store.save(state.withError(rootMessage(e))); + } catch (Exception fail) { + log.warn("Failed to record command error state: id={}", state.id(), fail); + } + } + + /** + * The most specific message available for a processing error. {@link #executeWithTimeout} + * wraps a handler exception in a generic {@code RuntimeException("Command execution failed")}, + * so the root cause's message is recorded instead — otherwise every transient error on the + * execute() path would read "Command execution failed" and the field would be useless for + * diagnosing a retry storm. The checkStatus() path throws directly, where the root cause is + * the exception itself. + */ + private static String rootMessage(Throwable e) { + Throwable root = e; + while (root.getCause() != null && root.getCause() != root) { + root = root.getCause(); + } + return root.getMessage() != null ? root.getMessage() : root.toString(); + } } 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 c922eb01..8a79a248 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,7 +14,6 @@ * limitations under the License. * */ - package io.seqera.data.command; import java.time.Instant; @@ -26,6 +25,28 @@ * 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 RUNNING (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, @@ -36,8 +57,10 @@ 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 ) { @@ -45,19 +68,22 @@ 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.SUBMITTED, params, - null, null, Instant.now(), null, null + null, null, 0, now, null, now, null ); } /** - * Transition to RUNNING status. + * Transition to RUNNING status. A successful (non-throwing) transition, so the + * consecutive-error streak is reset. */ public CommandState started() { + final Instant now = Instant.now(); return new CommandState( id, type, CommandStatus.RUNNING, params, - result, error, createdAt, Instant.now(), completedAt + result, error, 0, createdAt, now, now, completedAt ); } @@ -65,9 +91,10 @@ 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, createdAt, startedAt, Instant.now() + result, null, 0, createdAt, startedAt, now, now ); } @@ -75,9 +102,10 @@ 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, createdAt, startedAt, Instant.now() + null, error, errorsCount, createdAt, startedAt, now, now ); } @@ -85,9 +113,33 @@ 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, createdAt, startedAt, Instant.now() + 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 ); } 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 fbcfac60..8ccaad65 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 @@ -188,6 +188,26 @@ class CommandServiceTest extends Specification implements TestPropertyProvider { 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'() { @@ -278,6 +298,10 @@ class TestCommandHandler implements CommandHandler { 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.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 74544596..23946b5c 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,6 +20,7 @@ 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 @@ -172,4 +173,36 @@ class CommandStateSerializationTest extends Specification { decoded.result == null decoded.status == CommandStatus.RUNNING } + + 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' + decoded.id() == 'cmd-legacy' + decoded.status() == CommandStatus.RUNNING + 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 + } }