Replace lib-data-stream-redis with lib-data-workqueue

[pditommaso]

Moves the job queues from the message stream API onto the work queue API of libseqera: lib-data-stream-redis:1.2.0 → lib-data-workqueue:2.0.0 + lib-data-workqueue-redis:2.0.0.

lib-data-workqueue-redis is lib-data-stream-redis 1.5.0 re-landed on lease semantics — identical Redis stream layout (xadd/xreadGroup/xautoclaim, xack+xdel on settle, same data field). Queue keys (jobs-pending/v2, jobs-queue/v1) and the consumer group are left untouched, so no migration is needed and messages queued before the upgrade are still delivered afterwards.

Changes

build.gradle	swap the stream artifact for the two work-queue artifacts
service/data/workqueue/BaseWorkQueue	was BaseMessageStream, now on AbstractWorkQueue; same Moshi encoding strategy
service/data/workqueue/RedisWorkQueueConfigBean	was RedisStreamConfigBean, now implements RedisWorkQueueConfig
JobPendingQueue / JobProcessingQueue	extend BaseWorkQueue; queue ids unchanged
JobManager	(job) -> boolean becomes (job, lease) -> Decision.ACK/RETRY
docs/configuration.md	"Message stream" section becomes "Work queue"

launchJob/processJob are unchanged and still return a boolean, mapped onto ACK/RETRY at the addConsumer site — so JobManager keeps its own admission control and the maxRunningJobs cap is unaffected.

Why: work queue vs message stream

	message stream (before)	work queue (after)
Slow consumer	An entry's idle clock only resets on delivery/claim, so a job outlasting claim-timeout can be claimed and run concurrently by another replica. The timeout had to exceed worst-case processing time — hence 45s in prod.	A lease is renewed at visibility-timeout/4 while the owner is alive (one XCLAIM JUSTID per queue per tick), holding the idle clock near zero. The timeout now only governs recovery from a dead owner.
Shutdown	thread.interrupt() then join(1s) — severs a consumer mid-processing, and an interrupt can hand a RESP-desynced connection back to the Jedis pool.	Cooperative: a closing flag checked at loop head, waits (default 10s) for the current cycle, never interrupts. A processJob cycle completes its notifyJobCompletion/cleanup before teardown. awaitQuiescent(timeout) available for a bounded drain.
Retry pacing	Any delivery counted as progress, so a cycle that only returned false skipped the poll-interval sleep.	Only ACK/DEFERRED count as progress, so a queue that is merely retrying (job still running, pending queue full) paces at its poll interval.
Ownership safety	—	Renewal does an XPENDING ownership check first: an entry that drifted to another consumer is dropped and logged rather than seized back — the residual duplicate window is observable, not silent.

Not used by this PR, but now available: MessageConsumer.ready() (an admission gate checked before claiming — the natural home for the maxRunningJobs check, which currently claims a message and then refuses it), MessageLease/DEFERRED with retryAfter(delay), and lease/renewal/saturation metrics. Still at-least-once: renewal narrows the duplicate window, it does not close it.

Rollout

No Redis migration. Entries queued by the old code are read as-is (the integration test logs consume group=wave-message-stream already exists), and entries left pending by a terminating pod are reclaimed after the visibility timeout — in-flight jobs resume. Mixed-version replicas are safe both ways: an entry held by a new replica is renewed so an old replica's 45s-min-idle claim never sees it; an entry held by an old replica may be claimed after 45s idle, exactly as today.

Config keys are renamed wave.message-stream.* → wave.work-queue.* (claim-timeout → visibility-timeout), so the matching rename in platform-deployment (prod + stage, 45s) must land with this. That change is prepared but not yet pushed.

⚠️ The one hazard is rollback: revert the image with the new ConfigMap in place and the old code finds no message-stream.claim-timeout, falling back to a 5s claim timeout with no renewal. Keeping both key sets in the ConfigMap for one release makes it order-independent in both directions. Running pods are unaffected either way — the ConfigMap is mounted via subPath (never updated in place) and the Deployment has no config checksum annotation, so config is only read at container start.

Also note the cooperative close() can take up to 10s per queue, sequentially across two beans, against the default 30s terminationGracePeriodSeconds. In practice the dispatcher exits at its next loop-head check unless mid-consume; raising the grace period would add headroom.

The consumer group keeps defaulting to wave-message-stream deliberately — it is a wire identifier now, and renaming it would strand the pending-entry list.

Testing

• compileGroovy clean
• io.seqera.wave.service.job.* — 39 tests, 0 failures
• BuildStoreRedisTest (7) and RegistryControllerRedisTest (2) green against real Redis; logs confirm RedisWorkQueue reusing the existing consumer group and picking up the config bean (lease renewal period=1250ms)

Full suite not run locally — left to CI.

🤖 Generated with Claude Code

[pditommaso]

@claude review

[claude]

Commenting rather than approving — not because I found a defect, but because the safety of this swap rests on lib-data-workqueue:2.0.0 semantics I could not read from this checkout (no network access, the artifacts aren't in the local Gradle cache). The in-repo diff itself is clean and reads correctly.

What I verified in the repo

• No remaining references to io.seqera.data.stream, MessageStream or wave.message-stream anywhere under src/ — the only survivors are the renamed docs section — so dropping lib-data-stream-redis leaves nothing dangling.
• No wave.message-stream.* keys in application*.yml (main or test); the defaults only ever lived in the @Value annotations, so RedisWorkQueueConfigBean is a faithful 1:1 replacement of RedisStreamConfigBean.
• Queue ids are byte-identical (jobs-pending/v2, jobs-queue/v1) and the consumer group default stays wave-message-stream — the "no migration" claim holds at the level I can check, and the javadoc note explaining why that default looks stale is the right thing to have written down.
• The exception paths in launchJob/processJob still return true → ACK, so a job that throws is still consumed rather than looped on. Unchanged from before.
• CI is green, including Build Wave (25).

What the correctness of this turns on, which I can't see

1. length() vs. in-flight entries. launchJob0 gates admission on processingQueue.length() < config.maxRunningJobs — that only works if length() counts leased-but-unsettled entries (the PEL), which it did under the old xack+xdel-on-settle layout. If AbstractWorkQueue.length() in 2.0.0 reports only unclaimed entries, every running job disappears from the count and maxRunningJobs silently stops capping anything. The PR body says the Redis layout is identical, which implies this is fine — worth a one-line confirmation since it is the load-bearing assumption for the whole job pipeline.

2. Delivery-attempt cap / dead-lettering. processJob returns false → RETRY on every poll for the entire life of a running job, and poll-interval defaults to 1s. A 30-minute build settles on the order of 1800 RETRYs against one entry. If 2.0.0 has any max-delivery-count that drops or dead-letters an entry, the job is stranded: no notifyJobCompletion, no jobService.cleanup, and the build hangs until its await timeout. Nothing in the description suggests such a cap exists ("still at-least-once"), but "unbounded retries are fine" is exactly the kind of thing a 2.0 of a queue library might have changed.

3. When a RETRY becomes visible again. The description's retry-pacing row implies a retried entry comes back on the next poll cycle (~1s), not after visibility-timeout. If it's the latter, job-completion detection latency goes from ~1s to the configured 45s in prod — a large behaviour change that wouldn't show up in the local test runs listed.

If those three are as the description says, this is a straight approve from me.

Release ordering

The wave.message-stream.* → wave.work-queue.* rename is silent in both directions: Micronaut ignores unknown keys, so an unrenamed ConfigMap doesn't fail, it just drops prod/stage from 45s to the 5s default. The description already flags this and says the platform-deployment change is prepared but not pushed — worth making that a hard merge dependency rather than a follow-up.

For the rollback direction specifically, accepting the old key as a nested fallback for one release makes it order-independent both ways with no coordination at all:

@Value('${wave.work-queue.visibility-timeout:${wave.message-stream.claim-timeout:5s}}')
Duration visibilityTimeout

Same for consumer-warn-timeout / consumer-group-name. Optional, but it removes the one hazard the description calls out.

Nits (non-blocking)

• JobManagerTest exercises launchJob/processJob directly and returns booleans, so nothing asserts the new boolean → ACK/RETRY mapping at the addConsumer site. It's one line per queue, but it's the line that decides whether a message is deleted, and it's currently uncovered.
• The PR body's changes table says docs/configuration.md; the file actually touched is docs/install/reference.md.

I could not run the suite or inspect the library jars in this environment, so treat the above as a read of the diff plus repo-wide greps, not as verification of the runtime behaviour.

[pditommaso]

Verified the three open questions against the lib-data-workqueue:2.0.0 source directly:

1. length() counts leased-but-unsettled entries — it's XLEN, and an entry only leaves the stream on ACK (xack+xdel together). Running/RETRY'd entries stay counted, so the maxRunningJobs cap still works. Identical to the old stream.
2. No delivery-count cap / dead-lettering — RETRY does no Redis call, there's no attempt counter, and the only pruning (age backstop) is liveness-gated and just re-recovers the entry rather than dropping it. wave uses synchronous ACK/RETRY (no DEFERRED/liveness), so leases settle within the consume() call and nothing accumulates. A long build is not stranded; still at-least-once.
3. RETRY re-visibility is the visibility timeout (45s in prod), not ~1s — but the old stream behaved identically (claim-timeout was also 45s), so completion-detection latency is unchanged. The retry-pacing row is about the dispatcher loop no longer busy-spinning, not per-entry redelivery timing.

So all three are as the description says — no behaviour change from the swap.

Separately actioning the two nits (mapping-site test coverage; PR body should read docs/install/reference.md, not docs/configuration.md).

[pditommaso]

@munishchouhan can you please have a review at this PR ?

[munishchouhan]

bindLiveness is never called, so the lease age backstop prunes every lease

JobManager.init() accepts the lease and discards it:

pendingQueue.addConsumer((job, lease)-> launchJob(job) ? Decision.ACK : Decision.RETRY)
processingQueue.addConsumer((job, lease)-> processJob(job) ? Decision.ACK : Decision.RETRY)

With no probe bound, Lease.isOwnerAlive() returns false unconditionally, so the backstop in RedisWorkQueue:463 collapses to a plain age check:

if (!lease.isHeldForRelease() && lease.ageNanos() > maxLeaseAgeNanos && !lease.isOwnerAlive())

MessageLease#bindLiveness states it directly: "Without a bound probe the age backstop applies unconditionally."

Effect: a consumer cycle outliving maxLeaseAge (3 × visibility timeout → 15s on the 5s default) stops being renewed; one visibility timeout later another replica can claim the entry while launchJob/processJob is still running. Better than the 45s pre-PR window, but the timeout still governs slow owners, not only dead ones — which is the claim in the description.

Fix:

pendingQueue.addConsumer((job, lease)-> {
    lease.bindLiveness(() -> true)
    launchJob(job) ? Decision.ACK : Decision.RETRY
})

() -> true is accurate for a synchronous consumer: the lease is registered immediately before accept() and unregistered on every return path, so a still-registered lease means the handler is still running.

Trade-off to confirm before applying: this renews for as long as the handler runs, so a permanently hung handler would hold the entry indefinitely instead of being recovered after maxLeaseAge. Fine provided the k8s/Docker calls in jobService.status() and dispatcher.launchJob() are timeout-bounded.