From a93e0ef78894f8a4b8b035d8c79727a6e8a72a47 Mon Sep 17 00:00:00 2001 From: Paolo Di Tommaso Date: Fri, 31 Jul 2026 11:02:17 +0200 Subject: [PATCH] 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. VERSION is deliberately untouched. Companion to the retry fix in #102: 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. Module suite 18/18 green. Co-Authored-By: Claude Opus 5 (1M context) --- lib-cmd-queue-redis/changelog.txt | 14 ++++ .../data/command/CommandServiceImpl.java | 35 ++++++++++ .../io/seqera/data/command/CommandState.java | 66 +++++++++++++++++-- .../data/command/CommandServiceTest.groovy | 24 +++++++ .../CommandStateSerializationTest.groovy | 33 ++++++++++ 5 files changed, 165 insertions(+), 7 deletions(-) diff --git a/lib-cmd-queue-redis/changelog.txt b/lib-cmd-queue-redis/changelog.txt index 9f893f41..a816e073 100644 --- a/lib-cmd-queue-redis/changelog.txt +++ b/lib-cmd-queue-redis/changelog.txt @@ -1,6 +1,20 @@ # lib-cmd-queue-redis changelog 0.4.1 - 31 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 the fix below): 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. +- This restores the error tracking originally released as 0.6.0 (#89, e54229e), which was dropped + by the revert to 0.4.0 below. It is the companion to the retry fix below: once a thrown handler + is retried rather than terminal-failed, a command can retry indefinitely, and these fields are + what make that visible. - 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 unacked so the stream layer re-polls it — instead of persisting FAILED and acking (which removed the 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 + } }