Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

SonarCloud S1172 fixed — d986c3823

_run_batch_stages no longer takes the unused task_instance param. When I extracted it from _process_file_batch_core, task_instance's only use (deriving celery_task_id) stayed in the caller, so the extracted helper received it but never used it. Removed the param + updated both call sites; _process_file_batch_core keeps it (it reads .request.id). Routing test mocks with *a, unaffected; ruff clean.

Review round addressed — f5c6b98d5

Thanks @muhammad-ali-e — excellent review, especially the strand-on-failure analysis. All 15 handled.

Decision on the strand windows (Critical #69 + High #74 + High #81): reaper-as-safety-net (confirmed with the author). Fixed everything catchable in-process; the un-catchable windows are now documented as a hard reaper dependency for PR 3 rather than "future work". 2c is gated off (resolve_transport→celery), so none of these can bite until the reaper recovery net exists.

Failure handling

• fixing document service #69 — run_batch_with_barrier wraps both work + decrement in the abort, so a decrement-side failure (guard / DB / last-batch callback dispatch) tears down in-body instead of stranding. + test.
• Renaming and removal unused constants #79 — _abort_barrier_in_body logs inside the suppression (no more false "torn down").
• changed exception handler to a hook #74 / Removed existing rows before altering the table #81 — documented the SIGKILL-mid-work and post-commit-callback-dispatch windows as a hard reaper dependency for PR 3.
• Authorization/Permissions Adapter #86 — finalise cleanup split into two independent try/excepts with distinct logs; the dedup leak documented as a PR-3 reaper dependency.

Typing / clarity

• fix: Prompt studio index and run issues #1 BarrierContext(TypedDict); connectors #3 renamed backend→transport (collided with QueueBackend); Edit adapter fix and optimizations #27 is_pg_transport() predicate; Feature/app deployment docker fixes #20 extracted _dispatch_pg(); Reverted the condition to filter based on created_by #35 normalize_transport() at the general worker entry; Settings Modal in Prompt Studio #94 log on missing header queue; Prompt Studio - Summarize, Output Analyzer and other UI Improvements #9/[WIP] [FEAT] V1 - Time based trials  #13 born-stale comment + kwargs-not-args docstring.

Tests (#37/#41) — last-batch self-chains callback to PG + cleans up barrier/dedup; decrement-failure aborts; PG-branch mid-loop dispatch-failure deletes row; header args/queue/pre-existing-kwargs preservation. (Claim race already covered at the claim_batch level — run_batch_with_barrier decrements iff the claim won.)

Verification: 137 barrier/dedup/routing tests green; bootstrap clean under WORKER_BARRIER_BACKEND=pg; ruff clean. Still inert — net behaviour on the default path: NONE.

The reaper-recovery dependency (sweep stranded barriers → mark ERROR / re-drive, reclaim pg_batch_dedup) is now an explicit gate on enabling pg_queue_execution_enabled in PR 3.

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant O as Orchestrator
    participant PB as PgBarrier.enqueue()
    participant PQ as PG Queue
    participant W as Worker (process_file_batch)
    participant D as _barrier_pg_decrement
    participant CB as Callback Task

    O->>PB: "create_chord_execution(transport=pg_queue)"
    PB->>PB: "UPSERT pg_barrier_state(remaining=N) + DELETE pg_batch_dedup"
    loop for each header task
        PB->>PQ: "dispatch(task + _barrier_context, backend=PG)"
    end
    PQ->>W: deliver process_file_batch(_barrier_context)
    W->>W: claim_batch: True or False(skip redelivery)
    W->>W: "_run_batch_stages() -> result"
    W->>D: _barrier_pg_decrement(result)
    D->>D: "UPDATE remaining-=1, append result"
    alt "remaining > 0"
        D-->>W: pending
    else "remaining == 0"
        D->>CB: "_fire_barrier_callback -> dispatch(backend=PG)"
        D->>D: DELETE pg_barrier_state + pg_batch_dedup
        CB-->>O: callback fires
    end
    alt exception
        W->>W: "_abort_barrier_in_body -> barrier_pg_abort + clear_execution_batches"
    end

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant O as Orchestrator
    participant PB as PgBarrier.enqueue()
    participant PQ as PG Queue
    participant W as Worker (process_file_batch)
    participant D as _barrier_pg_decrement
    participant CB as Callback Task

    O->>PB: "create_chord_execution(transport=pg_queue)"
    PB->>PB: "UPSERT pg_barrier_state(remaining=N) + DELETE pg_batch_dedup"
    loop for each header task
        PB->>PQ: "dispatch(task + _barrier_context, backend=PG)"
    end
    PQ->>W: deliver process_file_batch(_barrier_context)
    W->>W: claim_batch: True or False(skip redelivery)
    W->>W: "_run_batch_stages() -> result"
    W->>D: _barrier_pg_decrement(result)
    D->>D: "UPDATE remaining-=1, append result"
    alt "remaining > 0"
        D-->>W: pending
    else "remaining == 0"
        D->>CB: "_fire_barrier_callback -> dispatch(backend=PG)"
        D->>D: DELETE pg_barrier_state + pg_batch_dedup
        CB-->>O: callback fires
    end
    alt exception
        W->>W: "_abort_barrier_in_body -> barrier_pg_abort + clear_execution_batches"
    end

_{Reviews (3): Last reviewed commit: "UN-3563 fix greptile #2069: mid-loop ded..." | Re-trigger Greptile}

@greptile-apps — both findings fixed (a2d7ba8c9)

Thanks for the 4/5. Both were real (in the gated PG path):

Issue 1 — dedup markers orphaned on mid-loop PG dispatch failure. Correct: with the barrier row deleted, an earlier header's in-flight barrier_pg_abort returns already_aborted and never reaches the clear_execution_batches inside it. Fixed — the mid-loop handler now calls clear_execution_batches(execution_id) directly on the PG path after _delete_barrier. + test asserting an already-claimed marker is reclaimed (not orphaned).

Issue 2 — PG callback dropped fairness headers. Correct — the callback was always dispatched at default priority, diverging from the Celery path. Went with your "extend the descriptor to carry the FairnessKey value" suggestion via a small _fairness_from_headers() that reconstructs the FairnessKey from the stored x-fairness-key headers and passes it to _dispatch_pg, so the PG callback rides the producer's org/priority. + tests for the fairness-carried and no-fairness (None-safe) paths.

75 barrier/dedup tests green; ruff + ruff-format clean; bootstrap clean under WORKER_BARRIER_BACKEND=pg. Still gated/inert by default.

Both fixes look solid. The implementation is clean and the tests are thorough.

Issue 1 (mid-loop dedup cleanup): The clear_execution_batches call in the mid-loop handler is correctly scoped to the is_pg branch, and test_pg_mid_loop_dispatch_failure_deletes_row_and_clears_dedup makes the right assertion: a pre-committed marker at index 0 gets reclaimed along with the barrier row. The invariant is closed.

Issue 2 (PG callback fairness): _fairness_from_headers is clean. The DEFAULT_PRIORITY fallback on pipeline_priority=payload.get("pipeline_priority", DEFAULT_PRIORITY) correctly handles headers produced before pipeline_priority was a field. Note: WorkloadType(payload["workload_type"]) is a hard lookup with no fallback — that's the right call here (a malformed stored header is a programming error, not a runtime condition), so the absence of a try/except is intentional and correct. Both test paths (fairness_carried / without_fairness_passes_none) pin the contract precisely.

75 tests, both fixes targeted. LGTM — good to merge when ready.

SonarCloud S3776 fixed — 90c61ce02

PgBarrier.enqueue cognitive complexity 17 → under 15. Extracted the per-header dispatch loop into PgBarrier._dispatch_headers — the deeply-nested for → try/except → if/else → if (PG-vs-celery branch + mid-loop teardown + PG dedup-clear) was the driver; that nesting was added across the last two review rounds. Behaviour identical, just relocated. radon: enqueue C(11)→B(6); ruff C901 passes at the 15 threshold; 75 barrier/dedup tests green.

@greptile-apps — "test passes for the wrong reason" fixed (9e078fb94)

Spot on. The pre-seeded claim_batch("exec-midfail", 0) marker was wiped by enqueue's UPSERT block (the reuse-reset DELETE FROM pg_batch_dedup) before the dispatch loop ran, so the mid-loop clear_execution_batches deleted 0 rows — the count == 0 assertion passed on the UPSERT, not the guard under test.

Fixed per your suggestion: the first dispatch side-effect now calls claim_batch("exec-midfail", 0) before returning its handle — inserting the marker after the UPSERT cleared stale ones (a fast PG consumer claiming its batch). The second dispatch fails, and the mid-loop clear_execution_batches is now what removes the marker.

Verified it's load-bearing: with the mid-loop clear no-op'd the marker orphans (count=1); with it enabled, count=0. 57 barrier/dedup tests green, ruff clean.

The fix is exactly right and the reasoning is sound. The ordering invariant now correctly models what happens at runtime:

1. UPSERT clears stale dedup rows (reuse-reset)
2. PG consumer races in and claims its batch (the side-effect on the first dispatch)
3. Loop aborts mid-flight → clear_execution_batches removes the marker the consumer inserted

The old test was asserting count == 0 after a code path that never actually exercised clear_execution_batches — the UPSERT block had already wiped the row before the guard had a chance to run. The fix makes the marker insertion temporally live (post-UPSERT), so the only way count == 0 can hold is if clear_execution_batches did its job.

The "verify it's load-bearing" check (no-op the mid-loop clear → count=1 orphan) is exactly the right validation criterion — it confirms causality, not just coincidence. Clean.

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud