Repository navigation
perf(enrichment): back off long-stuck custody deferrals on the C3 consumer route - #24377
Conversation
…sumer route #24264 bounded long-stuck custody deferrals (redirect_unresolved, conversation_missing, open_live_target, pending_live_bridge) to an hourly poll after a 6h grace, but only in the sync handler's run(). Prod serves enrichment through the C3 consumer (ENRICHMENT_ROUTE=enrichment_queue), whose claim_key/defer loop re-armed the same rows every 30-120s. The consumer now passes the custody cause through Result.defer_cause, and FirestoreEnrichmentOutbox.defer records deferred_since with the same episode rule as the sync handler. A completed callback ends the episode. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
There was a problem hiding this comment.
3 issues found across 6 files
Confidence score: 3/5
- In
outbox_firestore.py,_casusesref.update(new), so omittingdeferred_sincefrom the ack data leaves it stored in Firestore; a later deferral may continue the old episode. Explicitly delete the field and test that a deferral after ack starts a new episode. test_workers_consumer_contract.pysays it checks that metrics keep the reason, but it never checks a metric. Rename the test or add aCONSUMER_RESULTS_TOTALassertion forreason='not_due'.
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="backend/tests/unit/test_workers_consumer_contract.py">
<violation number="1" location="backend/tests/unit/test_workers_consumer_contract.py:111">
P3: The test name claims metrics keep the reason, but the body never checks a metric. Rename it to describe only the outbox assertion, or add a `CONSUMER_RESULTS_TOTAL` check for `reason='not_due'`.</violation>
</file>
<file name="backend/database/outbox_firestore.py">
<violation number="1" location="backend/database/outbox_firestore.py:307">
P2: Dropping `deferred_since` from the returned dict does not delete it in Firestore. `_cas` writes with `ref.update(new)`, which merges, so the stored `deferred_since` survives a successful ack. A later deferral with the same cause would then inherit the old episode and move to hourly polling early. Delete the field explicitly (for example with `firestore.DELETE_FIELD`) or confirm the emulator conformance test covers this.</violation>
</file>
<file name="backend/tests/unit/test_outbox_firestore.py">
<violation number="1" location="backend/tests/unit/test_outbox_firestore.py:206">
P3: This test checks that the ack removes `deferred_since` but never checks the behavior the PR relies on: the next deferral after the ack opens a new episode. Add a deferral after the ack and assert that its `deferred_since` is the new time.</violation>
</file>
Reply with feedback, questions, or to request a fix.
View guided diff | Turn on auto-fix | Re-trigger cubic
| def finish(old): | ||
| if not _holds(old, claim): | ||
| return old, (False, None) | ||
| old = {key: value for key, value in old.items() if key != 'deferred_since'} # A completed callback ends it. |
There was a problem hiding this comment.
P2: Dropping deferred_since from the returned dict does not delete it in Firestore. _cas writes with ref.update(new), which merges, so the stored deferred_since survives a successful ack. A later deferral with the same cause would then inherit the old episode and move to hourly polling early. Delete the field explicitly (for example with firestore.DELETE_FIELD) or confirm the emulator conformance test covers this.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/database/outbox_firestore.py, line 307:
<comment>Dropping `deferred_since` from the returned dict does not delete it in Firestore. `_cas` writes with `ref.update(new)`, which merges, so the stored `deferred_since` survives a successful ack. A later deferral with the same cause would then inherit the old episode and move to hourly polling early. Delete the field explicitly (for example with `firestore.DELETE_FIELD`) or confirm the emulator conformance test covers this.</comment>
<file context>
@@ -304,6 +304,7 @@ def ack(self, claim: Claim, reason: str, *, now: float) -> bool:
def finish(old):
if not _holds(old, claim):
return old, (False, None)
+ old = {key: value for key, value in old.items() if key != 'deferred_since'} # A completed callback ends it.
summary_seen = old.get('summary_seen_by') == claim.lease_token
if old['version'] == version:
</file context>
| assert outbox.calls[1][-1] == result.reason.value | ||
|
|
||
|
|
||
| def test_defer_cause_is_what_the_outbox_records_and_metrics_keep_the_reason(): |
There was a problem hiding this comment.
P3: The test name claims metrics keep the reason, but the body never checks a metric. Rename it to describe only the outbox assertion, or add a CONSUMER_RESULTS_TOTAL check for reason='not_due'.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/tests/unit/test_workers_consumer_contract.py, line 111:
<comment>The test name claims metrics keep the reason, but the body never checks a metric. Rename it to describe only the outbox assertion, or add a `CONSUMER_RESULTS_TOTAL` check for `reason='not_due'`.</comment>
<file context>
@@ -108,6 +108,13 @@ def test_run_push_settles_the_row_with_the_claim_lease_token(result, expected_ca
assert outbox.calls[1][-1] == result.reason.value
+def test_defer_cause_is_what_the_outbox_records_and_metrics_keep_the_reason():
+ outbox = FakeOutbox()
+ result = Result(Outcome.DEFERRED, ConsumerReason.NOT_DUE, available_at=2_000.0, defer_cause='redirect_unresolved')
</file context>
| def test_defer_cause_is_what_the_outbox_records_and_metrics_keep_the_reason(): | |
| def test_defer_cause_is_what_the_outbox_records(): |
| claim = _claim(rows, now=later + 40) | ||
| _enqueue(later + 1, now=later + 40) # an append during processing: ack re-arms instead of retiring the row | ||
| assert outbox.ack(claim, 'processed', now=later + 41) | ||
| assert 'deferred_since' not in state_rows[KEY] |
There was a problem hiding this comment.
P3: This test checks that the ack removes deferred_since but never checks the behavior the PR relies on: the next deferral after the ack opens a new episode. Add a deferral after the ack and assert that its deferred_since is the new time.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/tests/unit/test_outbox_firestore.py, line 206:
<comment>This test checks that the ack removes `deferred_since` but never checks the behavior the PR relies on: the next deferral after the ack opens a new episode. Add a deferral after the ack and assert that its `deferred_since` is the new time.</comment>
<file context>
@@ -170,6 +170,42 @@ def test_defer_mints_a_new_wake_and_release_keeps_it(rows):
+ claim = _claim(rows, now=later + 40)
+ _enqueue(later + 1, now=later + 40) # an append during processing: ack re-arms instead of retiring the row
+ assert outbox.ack(claim, 'processed', now=later + 41)
+ assert 'deferred_since' not in state_rows[KEY]
+
+
</file context>
| assert 'deferred_since' not in state_rows[KEY] | |
| assert 'deferred_since' not in state_rows[KEY] | |
| outbox.defer(_claim(rows, now=later + 50), later + 60, 'not_due', now=later + 50) # same cause, new episode after the ack | |
| assert state_rows[KEY]['deferred_since'] == later + 50 and state_rows[KEY]['due_at'] == later + 60 |
#24367 bumped backend/config/plan_catalog.json to catalog_revision 13 and left the storefront_v2 goldens at 12, so test_storefront_presentation fails on main. Regenerated with UPDATE_STOREFRONT_GOLDENS=1; the only change in all 52 files is catalog_revision 12 -> 13. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
UPDATE_STOREFRONT_GOLDENS=1 also writes the app/test/fixtures/storefront_v2 mirror. Same single change as the backend goldens: catalog_revision 12 -> 13. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
Round 2 (Luna, delta) raised one P2 in Lanes don't touch firmware, so I've raised the finding with the rollout lead and tracked it in anx for the firmware owner. It doesn't block this PR. |
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Summary
#24264 made long-stuck custody deferrals (
redirect_unresolved,conversation_missing,open_live_target,pending_live_bridge) poll hourly after a 6h grace. It only did this in the sync handler'srun(). Prod serves enrichment through the C3 consumer (ENRICHMENT_ROUTE=enrichment_queue). There,FirestoreEnrichmentOutbox.claim_key+deferstill re-arm the same rows every 30-120s.This PR brings the same episode rule to the consumer route:
Result.defer_causecarries the domain cause.EnrichmentConsumersets it from the callback'sdefer_reason,run_pushpasses it tooutbox.deferinstead ofnot_due, and metrics keepConsumerReason.NOT_DUE.FirestoreEnrichmentOutbox.deferrecordsdeferred_sincewith the sharedcustody_deferred_since/custody_defer_untilhelpers, sodue_at,task_due_atandwake_check_aftermove tonow + 1honce a cause has held the row for 6h.ack) dropsdeferred_since, so a new deferral starts a new episode, as in the sync handler.Custody, sweep repair and the eventual resume are unchanged. The only difference is that a long-stuck row is checked hourly instead of every minute or two. The PG and provider-STT outboxes store the cause string where they used to store
not_due; both columns are plain text.Measurement (prod, ro-prod, aggregates only, 2026-11-02 ~10:40Z)
outbox_firestore:claim_key81.1k,outbox_firestore:defer75.6k writes/h (about 157k/h together).sync_reprocess_outboxrows withdeferred_reason == not_due: 7,554. Rows waking within 2 min (the hot loop): 1,800 are ≥6h old and about 170 are newer. 4.1k old rows already wake more than 1h out.utils/enrichment/callback.py) is inSTUCK_DEFER_REASONS.deferred_since, so each episode starts at the row's first deferral after deploy. The drop appears about 6h after the deploy.omi_firestore_billed_writes_by_caller_total{caller=~"database.outbox_firestore:(claim_key|defer)"}from deploy+6h onward.Rollout
Ships on by default. It reuses #24264's constants, so there is no new flag. Deploy:
backend-integration(the prod/v2/enrichment/runtarget,API_WORKERSrole). The lead batches it into a train.Tests
backend/test.shwith the targeted list (outbox_firestore, enrichment_route, workers_consumer_contract, sync_enrichment_custody, sync_reprocess_coalesce, enrichment_callback_parity, sync_processing_hold, sync_pending_live_bridge, live_demand_consumers, feedback_media_consumers): all pass.make preflight: passed.🤖 Generated with Claude Code