Recover interrupted MR consolidation jobs

[nforro]

Move cancelled, stale, and orphaned MR consolidation jobs into durable Redis recovery storage so redeployments and crashes do not lose work.

Use renewable heartbeats and payload ownership checks to prevent healthy long-running jobs from being recovered concurrently or cleaned up by an older worker. Recovery entries are prioritized and preserved across collisions, including during mixed-version rollouts. Custom hash-backed polling uses explicit shutdown recovery instead of synthetic Redis list entries.

Fixes https://redhat.atlassian.net/browse/PACKIT-5212.

RELEASE NOTES BEGIN

Interrupted MR consolidation jobs, including jobs affected by production redeployments
or worker crashes, are now recovered and retried instead of being lost.

RELEASE NOTES END

[qodo-for-packit]

PR Summary by Qodo

Recover interrupted MR consolidation jobs

🐞 Bug fix 🧪 Tests 🕐 20-40 Minutes

AI Description

• Preserve cancelled and stale consolidation jobs in durable Redis recovery slots.
• Prioritize recovered work over newly pending jobs during subsequent polling.
• Cover cancellation, atomic recovery, and recovery-first scheduling with unit tests.

Diagram

graph TD
  Pending[("Pending Slot")] --> Poller["Queue Poller"] --> Active[("Active Slot")] --> Worker["Consolidation Worker"] --> Completed["Job Complete"]
  Active -->|timeout| Sweep["Stale Sweep"] -->|compare and move| Recovery[("Recovery Slot")]
  Worker -->|cancelled| Recovery -->|priority poll| Poller

Loading

High-Level Assessment

The existing Redis hash and atomic Lua operations are the best fit for this targeted recovery behavior. A Redis Streams or separate-queue redesign could provide native delivery semantics, but would require broader migration and consumer changes without clear benefit for this fix.

Files changed (4) +152 / -57

Bug fix (3) +79 / -32

mr_consolidation_agent.pyRequeue active jobs when consolidation tasks are cancelled +4/-6

Requeue active jobs when consolidation tasks are cancelled

• Cancellation handling now moves the active job into recovery storage before propagating 'CancelledError'. This preserves work across graceful pod shutdowns and redeployments.

ymir/agents/mr_consolidation_agent.py

tasks.pyExpose the active-job recovery helper +1/-0

Expose the active-job recovery helper

• Re-exports 'requeue_active_job' from the shared merge queue module for agent and test consumers.

ymir/agents/tasks.py

merge_queue.pyAdd durable, prioritized recovery for active jobs +74/-26

Add durable, prioritized recovery for active jobs

• Adds an atomic active-to-recovery transition and prioritizes recovery entries when selecting the next job. Stale active jobs now use compare-and-move recovery instead of deletion, preserving their payload while avoiding races with completion or replacement.

ymir/common/merge_queue.py

Tests (1) +73 / -25

test_mr_consolidation.pyTest interrupted-job recovery and prioritized retries +73/-25

Test interrupted-job recovery and prioritized retries

• Extends the Redis test double to simulate recovery moves and recovery-first polling. Adds coverage for active-job requeueing and verifies cancellation stores jobs for retry instead of losing them.

ymir/agents/tests/unit/test_mr_consolidation.py

[qodo-for-packit]

Code Review by Qodo

🐞 Bugs (0) 📘 Rule violations (0) 📜 Skill insights (0)

⚠️ 4 lower-priority findings omitted to fit the comment size limit; re-run the review or view the findings in the Qodo portal.

1. Rolling upgrades can discard jobs ✓ Resolved 🐞 Bug ≡ Correctness

Description

_CONDITIONAL_HDEL_LUA unconditionally writes a stale active payload into the single recovery
field, replacing any payload already stored there. During a rolling upgrade, an older poller can
promote a pending job while that package and branch already has recovery work because only the new
picker recognizes recovery fields, and recovering the newly active job then erases the first
interrupted job.

Code

ymir/common/merge_queue.py[R337-340]

+    redis.call('HDEL', KEYS[1], ARGV[1])
+    redis.call('HDEL', KEYS[1], ARGV[4])
+    redis.call('HSET', KEYS[1], ARGV[3], ARGV[2])
+    return 1

Relevance

●●● Strong

Recent accepted precedent prioritizes atomic Redis handling of consolidation races and conflict
states; this overwrite can lose recovery work.

PR-#806

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The architecture permits recovery to coexist with other slots, and the new picker explicitly
prioritizes recovery because previous pollers did not understand it. Both recovery scripts use an
unconditional HSET against the single recovery key, so an existing payload is replaced rather than
retained when an active/recovery collision occurs during rollout.

docs/mr_consolidation_architecture.md[70-73]
ymir/common/merge_queue.py[145-160]
ymir/common/merge_queue.py[223-235]
ymir/common/merge_queue.py[333-340]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
The stale-job recovery script overwrites an existing recovery payload. Mixed-version workers can create an active job alongside recovery work because the previous picker ignores recovery fields, causing durable work to be lost during rollout.

## Fix Focus Areas
- ymir/common/merge_queue.py[145-160]
- ymir/common/merge_queue.py[223-260]
- ymir/common/merge_queue.py[333-353]
- ymir/common/merge_queue.py[432-444]

## Recommended Fix
Make recovery storage capable of retaining every interrupted payload rather than using an unconditional write to one field. Update both cancellation and stale recovery moves atomically, and update the picker to consume all retained recovery entries before pending work without allowing mixed-version workers to overwrite them.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

2. Shutdown delays newly picked jobs ✓ Resolved 🐞 Bug ☼ Reliability

Description

run_task_loop skips all handling of an orphaned poll result when repush_on_shutdown is false,
even though the consolidation poll may already have promoted its returned payload into the active
slot. If shutdown wins the poll race just before pick_next_job returns, no processing task exists
to invoke its cancellation recovery, so the job cannot be picked on the next poll and waits for the
stale sweep instead.

Code

ymir/common/base_utils.py[R293-297]

+                if repush_on_shutdown:
+                    await fix_await(redis_conn.rpush(source_queue, payload))
+                    task_loop_logger.info("Re-pushed orphaned poll result to %s on shutdown", source_queue)
+                else:
+                    task_loop_logger.info("Skipping generic orphan requeue for custom poller")

Relevance

●●● Strong

Recent accepted precedents support fixing shutdown orphan races; this directly loses promoted
recovery jobs and contradicts the PR intent.

PR-#675

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The poller returns only after pick_next_job has promoted the job and serialized it, while
_race_shutdown can classify that still-running poll as orphaned. The new false branch then only
logs and neither launches process_task nor calls requeue_active_job, despite the picker having
stored the job as active with a heartbeat.

ymir/agents/mr_consolidation_agent.py[1851-1864]
ymir/common/base_utils.py[211-237]
ymir/common/base_utils.py[290-297]
ymir/common/merge_queue.py[145-159]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
An in-flight consolidation poll can promote a job after shutdown starts, but `repush_on_shutdown=False` drops the poll result without invoking the custom durable recovery path. The promoted job remains active until stale recovery runs.

## Fix Focus Areas
- ymir/common/base_utils.py[290-297]
- ymir/agents/mr_consolidation_agent.py[1975-1982]

## Recommended Fix
Add a custom orphan-result recovery callback to `run_task_loop`, and have the consolidation agent use it to parse the returned payload and conditionally move the matching active job to recovery. Keep generic list re-pushing disabled for this hash-backed queue.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

3. A stalled Redis call strands jobs ✗ Dismissed 🐞 Bug ☼ Reliability

Description

The separately spawned heartbeat_task remains active while complete_job and requeue_active_job
await the timeout-free redis_conn, because it is drained only in finally. If that connection
stalls after workflow completion or during shutdown, the other Redis connection keeps renewing the
lease so stale recovery cannot take over, and shutdown also waits indefinitely for the processing
task.

Code

ymir/agents/mr_consolidation_agent.py[1949]

+                await complete_job(redis_conn, job.package, job.target_branch, payload)

Relevance

●●● Strong

Recent accepted precedent prioritizes cancellation cleanup and preventing Redis-related shutdown
leaks.

PR-#675
PR-#806

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The heartbeat is created independently and is not drained until the final cleanup, after both
completion and cancellation recovery calls. The main Redis client is opened without a socket
timeout, while shutdown waits for cancelled processing tasks to finish; consequently a blocked
terminal call can coexist with continuing heartbeat refreshes and prevent both stale recovery and
process termination.

ymir/agents/mr_consolidation_agent.py[1925-1971]
ymir/agents/mr_consolidation_agent.py[1844-1849]
ymir/common/base_utils.py[45-63]
ymir/common/base_utils.py[251-264]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
Terminal completion and recovery operations use the timeout-free main Redis connection while the independent heartbeat task remains running. A stalled operation can therefore keep its job leased forever and block graceful shutdown.

## Fix Focus Areas
- ymir/agents/mr_consolidation_agent.py[1925-1971]
- ymir/common/base_utils.py[45-63]

## Recommended Fix
Stop and drain the heartbeat before attempting terminal completion or requeue operations, and execute those ownership mutations through a Redis connection with a finite socket timeout. Preserve the conditional payload checks so a timed-out or retried mutation remains ownership-safe.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

View high (10)

4. Stalled Redis runs jobs twice ✓ Resolved 🐞 Bug ☼ Reliability

Description

heartbeat awaits refresh_active_heartbeat without a request timeout, while the shared Redis
client explicitly has no socket timeout. If that request stalls until the stored heartbeat becomes
stale, another worker can recover and execute the job while the original workflow continues
indefinitely without detecting its lost lease.

Code

ymir/agents/mr_consolidation_agent.py[R1871-1874]

+                        owned = await refresh_active_heartbeat(
+                            redis_conn,
+                            job.package,
+                            job.target_branch,

Relevance

●●● Strong

Clear lease-safety reliability risk; recent history accepts Redis race and network-stall
protections.

PR-#806
PR-#610

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The new heartbeat loop cannot evaluate its exception or ownership branches until
refresh_active_heartbeat returns, and the repository configures the underlying Redis client with
an unlimited socket timeout. Stale sweeping relies on heartbeat age, so a blocked refresh does not
prevent another worker from recovering the active entry.

ymir/agents/mr_consolidation_agent.py[1865-1898]
ymir/agents/mr_consolidation_agent.py[1927-1933]
ymir/common/base_utils.py[44-59]
ymir/common/merge_queue.py[404-445]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
The consolidation heartbeat can block indefinitely because its Redis refresh has no request timeout and the Redis client uses `socket_timeout=None`. While blocked, the active lease can expire and another worker can recover the same job without the original workflow being cancelled.

## Fix Focus Areas
- ymir/agents/mr_consolidation_agent.py[1865-1888]
- ymir/common/base_utils.py[44-59]

## Recommended Fix
Give heartbeat refreshes a finite I/O deadline that expires before the active lease can become stale. Treat expiry as a refresh failure and preserve the existing elapsed-lease logic, using a timeout mechanism or dedicated Redis connection that does not leave a cancelled Redis command corrupting shared connection state.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

5. Network partitions run jobs twice ✓ Resolved 🐞 Bug ☼ Reliability

Description

heartbeat starts failed_since at the first failed refresh and waits a full stale threshold from
that point, even though the stored lease has already aged by one heartbeat interval. When one worker
loses Redis connectivity but continues its external workflow, another replica can recover and start
the job at the lease threshold while the original workflow remains active for up to another
heartbeat interval.

Code

ymir/agents/mr_consolidation_agent.py[R1884-1886]

+                        if failed_since is None:
+                            failed_since = time.monotonic()
+                        elif time.monotonic() - failed_since >= stale_threshold.total_seconds():

Relevance

●●● Strong

Accepted race-condition precedents show reviewers prioritize lease ownership and duplicate-work
prevention in Redis workflows.

PR-#610
PR-#540

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The worker sleeps before refreshing and only records failed_since after that first failed attempt,
then permits failures for the entire stale threshold. Meanwhile, the sweep measures staleness from
the last heartbeat stored in Redis and moves the job to recovery as soon as that heartbeat exceeds
the same threshold, so recovery precedes the original worker's abort by approximately one heartbeat
interval.

ymir/agents/mr_consolidation_agent.py[1866-1890]
ymir/common/merge_queue.py[263-293]
ymir/common/merge_queue.py[404-444]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
Heartbeat failure timing begins at the first failed refresh rather than the last successful lease refresh. This allows the stale sweep to recover a job before its disconnected owner stops processing it.

## Fix Focus Areas
- ymir/agents/mr_consolidation_agent.py[1866-1890]

## Recommended Fix
Track monotonic time from the last successful heartbeat, initialized when processing starts, and abort the workflow before the Redis heartbeat can exceed the stale threshold. Include a safety margin for scheduling delay so another replica cannot recover the job while the original workflow is still running.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

6. Recovered consolidations run twice ✓ Resolved 🐞 Bug ☼ Reliability

Description

heartbeat only logs and returns when refresh_active_heartbeat reports lost ownership, leaving
the enclosing run_workflow task running. When stale recovery has already promoted the payload to
another worker, the original and recovered workflows can perform consolidation concurrently until
the original eventually finishes.

Code

ymir/agents/mr_consolidation_agent.py[R1882-1885]

+                            job.package,
+                            job.target_branch,
+                        )
+                        return

Relevance

●●● Strong

Accepted race-condition prevention precedents support cancelling or stopping work after ownership
loss to prevent duplicate processing.

PR-#610

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The heartbeat branch returns independently from the processing coroutine, while process_task
continues awaiting run_workflow and only attempts completion afterward. The stale sweep can
meanwhile move the payload to recovery, and pick_next_job explicitly promotes recovery entries for
another worker.

ymir/agents/mr_consolidation_agent.py[1861-1887]
ymir/agents/mr_consolidation_agent.py[1888-1919]
ymir/common/merge_queue.py[145-160]
ymir/common/merge_queue.py[404-445]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
The heartbeat coroutine stops silently after detecting that its active payload is no longer owned, while the associated consolidation workflow continues running. This allows a recovered worker and the original worker to process the same consolidation concurrently.

## Fix Focus Areas
- ymir/agents/mr_consolidation_agent.py[1861-1887]
- ymir/agents/mr_consolidation_agent.py[1888-1919]

## Recommended Fix
Propagate definitive ownership loss from the heartbeat coroutine to the processing task and cancel or abort `run_workflow` immediately. Also stop processing before lease expiration when heartbeat refresh failures persist for the stale threshold, while preserving the existing ownership-checked requeue behavior during cleanup.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

7. Slow consolidation jobs run twice ✓ Resolved 🐞 Bug ≡ Correctness

Description

sweep_stale_active_jobs treats a job’s one-time activated_at timestamp as proof that its worker
is dead and atomically moves the matching payload from :active to :recovery without checking
worker ownership, liveness, or a renewable lease. When a live workflow exceeds
STALE_ACTIVE_THRESHOLD_HOURS during build polling or retries, another replica can promote and
execute the recovered payload while the original task continues, and the original worker’s
conditional completion cannot remove the recovered or re-promoted copy.

Code

ymir/common/merge_queue.py[R378-381]

+                field,
+                value,
+                recovery_field,
+            )

Relevance

●●● Strong

Accepted concurrency and recovery-race findings are recent; this directly identifies duplicate
execution from lease-free stale recovery.

PR-#610
PR-#675
PR-#806

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The poller invokes the stale sweep before selecting work, and the sweep moves every active payload
older than the threshold into recovery by comparing only the stored payload rather than worker
ownership or liveness. Because activated_at is set only during promotion and never refreshed, a
workflow awaiting build tooling or retrying builds can exceed the timeout while still running; with
two replicas and recovery entries immediately eligible for promotion, another replica can execute
the same payload. complete_job removes only a still-matching active payload, so once the original
entry has been recovered or re-promoted, the original worker cannot remove that duplicate when it
finishes.

ymir/common/merge_queue.py[180-201]
ymir/common/merge_queue.py[372-381]
ymir/agents/mr_consolidation_agent.py[1825-1830]
ymir/agents/mr_consolidation_agent.py[1842-1890]
ymir/agents/mr_consolidation_agent.py[795-816]
ymir/agents/build_agent.py[50-57]
openshift/deployment-mr-consolidation-agent-c9s.yml[4-14]
ymir/agents/mr_consolidation_agent.py[1826-1843]
ymir/common/merge_queue.py[141-154]
ymir/common/merge_queue.py[271-288]
ymir/common/merge_queue.py[343-381]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description

`sweep_stale_active_jobs` uses the original `activated_at` timestamp as a non-renewable lease, so it can move a still-running consolidation job into `:recovery`. Another replica can then promote and execute the same payload concurrently, while the original worker’s later completion cannot remove the recovered or re-promoted copy.

## Fix Focus Areas

- ymir/common/merge_queue.py[343-381]
- ymir/common/merge_queue.py[180-201]
- ymir/agents/mr_consolidation_agent.py[1842-1890]

## Recommended Fix

Add a per-execution ownership token and renewable heartbeat or lease timestamp, and refresh it atomically while `process_task` or `run_workflow` is running. Make the stale sweep compare-and-move only jobs whose heartbeat has expired, and keep completion and cancellation conditional on the same ownership token so an active worker cannot be recovered by another replica. Add a test proving that a job running beyond the stale threshold is not recovered while its heartbeats continue.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

8. A crash can strand consolidation work ✗ Dismissed 🐞 Bug ☼ Reliability

Description

pick_next_job promotes the original payload to :active before persisting activated_at through
a second Redis operation. If the process exits between those operations, the stale sweep permanently
skips the timestamp-less active entry and later jobs for that package and branch cannot be promoted.

Code

ymir/common/merge_queue.py[R186-189]

+    updated = await fix_await(
+        redis_conn.eval(
+            _UPDATE_ACTIVE_LUA,
+            1,

Relevance

●●● Strong

Matches accepted shutdown-recovery concerns; separate promotion and timestamp writes leave a
crash-induced blocking state.

PR-#675

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The promotion script stores the original queued value at line 152, while the activation timestamp is
written only by the later compare-and-set at lines 183-195. activated_at defaults to None, and
the sweep explicitly skips such active records, so interruption in this gap leaves an entry that
blocks that package and branch indefinitely.

ymir/common/merge_queue.py[141-153]
ymir/common/merge_queue.py[180-201]
ymir/common/models.py[930-935]
ymir/common/merge_queue.py[350-358]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
Promotion and activation timestamp persistence use separate Redis operations, so a crash between them leaves an active job that the stale sweep will never recover.

## Fix Focus Areas
- ymir/common/merge_queue.py[141-201]
- ymir/common/merge_queue.py[350-358]

## Recommended Fix
Promote the selected queued job with its updated `active` and `activated_at` values in one atomic Lua operation. Select and serialize the updated candidate before a compare-and-move script, retrying selection when the queued value or active slot changed concurrently.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

9. Users can queue duplicate recovery work ✓ Resolved 🐞 Bug ≡ Correctness

Description

submit_merge_job checks only the pending and active fields, while pick_next_job now treats the
recovery field as queued work. When a label-triggered submission arrives after interruption moved a
job to recovery, strict mode accepts another pending job and both consolidation jobs subsequently
run.

Code

ymir/common/merge_queue.py[R138-141]

+for _, suffix in ipairs({':recovery', ':pending'}) do
+    for i = 1, #fields, 2 do
+        local field = fields[i]
+        local value = fields[i + 1]

Relevance

●●● Strong

PR #806 accepted atomic conflict prevention for consolidation submissions; recovery must participate
in duplicate detection.

PR-#806

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The submission Lua script checks :pending and, in strict mode, :active, but never :recovery;
the changed picker explicitly scans :recovery before :pending. Thus an interrupted job can
coexist with a newly accepted strict submission and both are eligible to run.

ymir/common/merge_queue.py[42-56]
ymir/common/merge_queue.py[67-117]
ymir/common/merge_queue.py[133-149]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
Strict merge-job submission does not recognize an existing recovery entry, allowing a duplicate pending job to be queued behind interrupted work.

## Fix Focus Areas
- ymir/common/merge_queue.py[42-117]
- ymir/common/merge_queue.py[133-149]

## Recommended Fix
Pass the package-and-branch recovery key into the submission Lua script and treat it like an active entry in strict mode. Preserve the documented auto-mode behavior that permits a pending job behind existing work, and add coverage for strict and automatic submissions while recovery exists.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

10. Older workers can delete recovered jobs ✓ Resolved 🐞 Bug ≡ Correctness

Description

sweep_stale_active_jobs now atomically moves a timed-out active payload to recovery, allowing a
second worker to promote and process it, while complete_job still unconditionally deletes the
package/branch active field without proving that the completing worker owns it. When the original
worker resumes after the six-hour sweep, its completion can delete the replacement worker's active
entry; a later cancellation of that replacement then has no entry to recover.

Code

ymir/common/merge_queue.py[R342-345]

+                field,
+                value,
+                recovery_field,
+            )

Relevance

●●● Strong

Recent accepted Redis race fixes require compare-and-update/delete semantics when stale workers
overlap replacements.

PR-#675
PR-#806

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The new stale sweep moves an active record into recovery, and the queue poller subsequently promotes
recovery records into the same active field. The existing completion operation is an unconditional
HDEL, so it cannot distinguish the original worker from the replacement that now owns that field.

ymir/common/merge_queue.py[133-149]
ymir/common/merge_queue.py[175-196]
ymir/common/merge_queue.py[248-269]
ymir/common/merge_queue.py[336-345]
ymir/agents/mr_consolidation_agent.py[1826-1829]
ymir/agents/mr_consolidation_agent.py[1882-1891]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

Issue description
A worker which was superseded after stale recovery can delete the active state belonging to its replacement because completion and cancellation requeue identify work only by package and branch.

Fix Focus Areas
- ymir/common/merge_queue.py[175-196]
- ymir/common/merge_queue.py[226-245]
- ymir/common/merge_queue.py[248-269]
- ymir/common/merge_queue.py[336-345]

Recommended Fix
Carry an immutable expected active payload or a dedicated execution token from promotion into task processing. Change completion to an atomic compare-and-delete and cancellation requeue to an atomic compare-and-move, so either operation changes Redis only when the stored active value still belongs to that worker. Add a concurrency test where stale recovery promotes a replacement before the original worker finishes.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

11. Shutdowns can leave jobs unrecovered ✓ Resolved 🐞 Bug ☼ Reliability

Description

The new except asyncio.CancelledError handler wraps run_workflow but not the normal-path `else:
await complete_job(...), because an else` suite is outside the protected try body. If shutdown
cancellation arrives while that Redis deletion is in progress, the recovery move is skipped and the
task loop only re-pushes the inert mr_consolidation sentinel payload; the hash entry is either
absent or remains stale until the six-hour sweep.

Code

ymir/agents/mr_consolidation_agent.py[R1878-1880]

+                # Preserve the job for the next pod instead of letting the
+                # generic task-loop cleanup erase the hash entry.
+                await requeue_active_job(redis_conn, job.package, job.target_branch)

Relevance

●●● Strong

PR #675 accepted fixes for cancellation and shutdown orphan-requeue paths, including failures around
protected cleanup.

PR-#675

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The new recovery action is limited to the CancelledError handler, whereas normal completion executes
in the try statement's else suite. The generic task loop re-pushes its source payload on shutdown,
but this agent explicitly supplies a sentinel source queue that its poller never reads, so that
fallback cannot restore hash-backed consolidation work.

ymir/agents/mr_consolidation_agent.py[1827-1840]
ymir/agents/mr_consolidation_agent.py[1878-1891]
ymir/common/merge_queue.py[248-258]
ymir/common/base_utils.py[244-271]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

Issue description
Cancellation during normal completion bypasses the newly added recovery handler because the completion await is in the try statement's else suite rather than its protected body.

Fix Focus Areas
- ymir/agents/mr_consolidation_agent.py[1878-1891]
- ymir/agents/mr_consolidation_agent.py[1827-1840]

Recommended Fix
Restructure task processing so cancellation handling also encloses normal completion cleanup. On cancellation during completion, conditionally move the still-owned active entry to recovery; if completion already succeeded, the conditional move should safely do nothing. Add a test that cancels an in-progress completion and verifies no job is left solely in the sentinel queue or stale active storage.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

12. Shutdown requeues another worker's job ✓ Resolved 🐞 Bug ☼ Reliability

Description

process_task calls requeue_active_job using only the package and branch, and its Lua script
moves whichever payload currently occupies that active field without verifying ownership. When a
live job is swept as stale and promoted to another pod before the original pod is cancelled, the
original handler moves the replacement back to recovery and leaves it eligible for another execution
after the replacement finishes.

Code

ymir/agents/mr_consolidation_agent.py[R1878-1880]

+                # Preserve the job for the next pod instead of letting the
+                # generic task-loop cleanup erase the hash entry.
+                await requeue_active_job(redis_conn, job.package, job.target_branch)

Relevance

●●● Strong

Recent shutdown and Redis race fixes were accepted; ownership validation prevents stale workers
requeueing replacements.

PR-#675
PR-#806

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

Stale sweeping can move a job while its original worker is still alive, and recovery promotion can
then place a refreshed copy under the same active key. Because cancellation unconditionally moves
the current value at that key, an old pod can requeue the new pod's work; normal completion only
deletes active, so the recovery copy remains queued.

ymir/agents/mr_consolidation_agent.py[1822-1832]
ymir/agents/mr_consolidation_agent.py[1877-1891]
ymir/common/merge_queue.py[212-245]
ymir/common/merge_queue.py[307-345]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
Cancellation requeues by package and branch alone, so a stale worker can move a replacement worker's active payload instead of its own.

## Fix Focus Areas
- ymir/agents/mr_consolidation_agent.py[1822-1889]
- ymir/common/merge_queue.py[212-245]
- ymir/common/merge_queue.py[307-345]

## Recommended Fix
Pass the cancelled job's serialized active payload or an immutable execution token to `requeue_active_job`. Change the Lua script to compare the current active value or token before moving it, return false on mismatch, and add a concurrency test where stale recovery is promoted before the original worker is cancelled.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

13. Recovered jobs run repeatedly ✓ Resolved 🐞 Bug ≡ Correctness

Description

pick_next_job accepts the new :recovery suffix in Lua, but its Python follow-up removes only
:pending, so the refreshed payload is written under a stray :recovery:active field. Every
recovered job leaves that field after completion; once it becomes stale, the sweep treats it as
another active job and queues the same payload again under increasingly nested recovery keys.

Code

ymir/common/merge_queue.py[R142-145]

+        if string.sub(field, -#suffix) == suffix then
+            local prefix = string.sub(field, 1, #field - #suffix)
+            local active_key = prefix .. ':active'
+            if redis.call('HEXISTS', hash, active_key) == 0 then

Relevance

●●● Strong

Accepted recovery-queue race and lifecycle bugs are prioritized; suffix mismatch deterministically
leaves stale active fields.

PR-#610
PR-#806

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The Lua promotion strips whichever selected suffix matched and creates <package>:<branch>:active,
while the Python code strips only :pending before writing the refreshed timestamp. Completion
deletes only the canonical active field, and the sweep processes every field ending in :active,
proving that the stray recovery field survives and later re-enters the queue.

ymir/common/merge_queue.py[138-179]
ymir/common/merge_queue.py[202-233]
ymir/common/merge_queue.py[283-321]
ymir/common/models.py[916-940]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
Recovery promotion creates the normal active field in Lua, but the Python follow-up derives a different field because it strips only `:pending`. This leaves an orphan ending in `:active`, which the stale sweep eventually requeues and executes repeatedly.

## Fix Focus Areas
- ymir/common/merge_queue.py[138-179]
- ymir/agents/tests/unit/test_mr_consolidation.py[264-284]

## Recommended Fix
Derive the active key correctly for both `:pending` and `:recovery` selections, preferably by returning the active key from the Lua script or by explicitly stripping the matched queue suffix. Extend the recovery-pick test to assert that only the canonical package/branch `:active` field exists and no `:recovery:active` field is created.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

14. Payload-only pollers can lose work ✓ Resolved 🐞 Bug ≡ Correctness ⭐ New

Description

run_task_loop maps any non-tuple polling result to source_queue = b"" without requiring
recovery_fn. On shutdown, a payload-only custom poller that omits the callback reaches the normal
RPUSH branch and stores consumed work under the empty key rather than a queue read by workers.

Code

ymir/common/base_utils.py[R246-251]

+        if isinstance(result, tuple):
+            source_queue, payload = result
+        else:
+            # Custom pollers may return only the payload when there is no
+            # source list to requeue into; recovery_fn owns shutdown handling.
+            source_queue, payload = b"", result

Relevance

●●● Strong

Payload-only shutdown handling can RPUSH to an unread empty key; shutdown recovery issues in this
utility were accepted.

PR-#675

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The new result normalization assigns an empty source queue to bare payloads. Both active-task and
orphaned-poll shutdown paths call rpush(source_queue, payload) when no recovery callback is
supplied, so this accepted configuration writes the task to b"".

ymir/common/base_utils.py[246-251]
ymir/common/base_utils.py[273-286]
ymir/common/base_utils.py[302-311]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
Payload-only poll results have no source queue, but `run_task_loop` permits them without a recovery callback and may subsequently push them to the empty Redis key during shutdown.

## Fix Focus Areas
- ymir/common/base_utils.py[246-251]
- ymir/common/base_utils.py[273-286]
- ymir/common/base_utils.py[302-311]

## Recommended Fix
Reject payload-only poll results when `recovery_fn` is absent, or make the poll result type explicitly distinguish recoverable custom payloads and require their recovery callback before starting processing.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

15. Automatic submissions scan every job ✓ Resolved 🐞 Bug ➹ Performance ⭐ New

Description

_SUBMIT_JOB_LUA calls HKEYS to search for recovery variants before checking whether the
submission mode is strict, although automatic mode never uses the recovery-conflict result. Every
automatic submission therefore traverses the entire consolidation hash, making submission cost grow
with all pending, active, heartbeat, and recovery fields.

Code

ymir/common/merge_queue.py[R61-64]

+local recovery_prefix_exists = redis.call('HEXISTS', hash, recovery) == 1
+if not recovery_prefix_exists then
+    local fields = redis.call('HKEYS', hash)
+    for _, field in ipairs(fields) do

Relevance

●●● Strong

The full HKEYS scan is unnecessary in automatic mode and creates avoidable hash-wide submission
overhead.

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The script executes HKEYS whenever the canonical recovery field is absent, before the strict-mode
condition. The result is consumed only by the later mode == 'strict' condition, making the full
scan unnecessary for automatic submissions.

ymir/common/merge_queue.py[55-73]
ymir/common/merge_queue.py[110-134]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
The submission Lua script performs a global recovery-field scan for automatic submissions even though only strict submissions reject recovery work.

## Fix Focus Areas
- ymir/common/merge_queue.py[61-73]

## Recommended Fix
Move recovery existence detection inside the strict-mode branch. Check the exact active and recovery fields first, and scan for suffixed recovery fields only when the mode is strict and no canonical recovery field exists.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

16. Shutdowns leave orphaned Redis entries ✓ Resolved 🐞 Bug ☼ Reliability

Description

process_task moves its active payload into recovery on cancellation, but run_task_loop
subsequently RPUSHes the same task using the synthetic source queue. Each cancelled deployment
therefore also creates an unconsumed mr_consolidation list entry beside the recoverable hash
entry, and those entries accumulate because the consolidation poller reads only the hash-backed
queue.

Code

ymir/agents/mr_consolidation_agent.py[R1951-1954]

+                # Preserve the job for the next pod instead of letting the
+                # generic task-loop cleanup erase the hash entry.
+                await drain(workflow_task)
+                await requeue_active_job(redis_conn, job.package, job.target_branch, payload)

Relevance

●●● Strong

Accepted precedents consistently address shutdown requeue reliability and duplicate or orphaned
Redis task handling.

PR-#675
PR-#806

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The changed cancellation handler now writes the job to recovery, while the generic task loop
independently re-pushes every cancelled task to its recorded source queue. This poller records a
sentinel queue name, and the consolidation implementation consumes a Redis hash rather than that
list.

ymir/agents/mr_consolidation_agent.py[1855-1864]
ymir/agents/mr_consolidation_agent.py[1951-1954]
ymir/common/base_utils.py[244-263]
ymir/common/merge_queue.py[10-14]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
The custom MR-consolidation poller uses a synthetic queue name solely to satisfy `run_task_loop` bookkeeping. After `process_task` has atomically moved a cancelled active entry into hash-backed recovery storage, the generic loop still RPUSHes that same payload to the synthetic Redis list, which no worker consumes.

## Fix Focus Areas
- ymir/agents/mr_consolidation_agent.py[1951-1954]

## Recommended Fix
Add an opt-out or callback to `run_task_loop` for custom pollers whose cancellation handler owns durable recovery, and enable it for MR consolidation. The loop must not perform its generic list `RPUSH` when this poller has already requeued the active job into recovery storage.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

Context sources

⚠️ Tickets: not configured — ticket URL found in PR but could not be fetched — check ticket provider credentials

✅ Compliance rules (platform): 8 rules

Review mode: 🧠 Deep: This push introduces substantial, bug-dense recovery, lease/heartbeat, atomic Redis, shutdown, and task-loop logic across multiple independent runtime paths with significant failure-mode and data-loss risk.

Tip of the day

💡 Did you know, you can enable the Remediation agent and Qodo fixes findings in a dedicated fix PR

More tips ↗ | Customize Qodo ↗ | Qodo docs ↗

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit 5aae1d7

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit 7c5c27b

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit 94b77dc

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit aa70456

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit 1d57a3e

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit 081e05c

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit b0e7358

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit 1275f23

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit 31940a5

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit e0708d7

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit 1bb1d76

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit 8856b85

[nforro]

/agentic_review

[qodo-for-packit]

Code review by qodo was updated up to the latest commit 58dc7b7