Fix UniqueViolation in equivalent_identifiers_refresh cache rebuild (PP-4786)

[dbernstein]

Description

Routes all recursiveequivalentscache inserts in the equivalent_identifiers_refresh task through a shared _insert_cache_rows helper that uses PostgreSQL INSERT ... ON CONFLICT (parent_identifier_id, identifier_id) DO NOTHING, so a concurrently-committed cache row no longer aborts the task. Also collapses the per-parent deletes into a single sorted WHERE parent_identifier_id IN (...) statement and chunks/streams the inserts.

Motivation and Context

JIRA: PP-4786

Production is throwing:

IntegrityError: duplicate key value violates unique constraint
"recursiveequivalentscache_parent_identifier_id_identifier_i_key"
DETAIL: Key (parent_identifier_id, identifier_id)=(188174, 2678719) already exists.

The refresh task rebuilds the cache with a delete-then-insert per parent identifier, implicitly assuming it is the only writer of the table. It is not:

1. The Identifier creation listener inserts (id, id) self-reference rows from ordinary webapp/import transactions, independent of this task.
2. The task runs under task_acks_late=True / task_reject_on_worker_lost=True and re-queues itself via task.replace() across thousands of batches (replaced_task_nesting: 5348 in the failing log). A lost worker or a lapsed visibility timeout lets the broker redeliver, and a second refresh chain can run concurrently on overlapping identifier chains. The Redis TaskLock only guards against the beat schedule starting a second run — a redelivery carries the same root_id, so it re-acquires the same lock and slips through.

When either writer commits a conflicting row between our delete and our flush, the batched INSERT ... RETURNING hits the unique constraint and the whole task aborts. Since every parent's rows are recomputed from the same source equivalencies, a row slipped in under us holds the same value we were about to write, so ON CONFLICT DO NOTHING converges on the correct chain instead of failing.

How Has This Been Tested?

• Added TestInsertCacheRows and TestProcessIdentifierIds::test_tolerates_rows_the_delete_did_not_remove. Confirmed both reproduce the production IntegrityError when the ON CONFLICT clause is removed, and pass with it in place.
• Full run of test_refresh_equivalents.py, test_equivalents.py, and test_listeners.py passes.
• mypy and pre-commit checks are clean.

Checklist

• I have updated the documentation accordingly.
• All new and existing tests passed.

🤖 Generated with Claude Code

[greptile-apps]

Greptile Summary

This PR makes recursive equivalency cache refreshes tolerate concurrent cache writes.

• Adds a shared chunked insert helper with PostgreSQL conflict handling.
• Replaces per-parent cache deletes with one bulk delete for affected parents.
• Updates identity-cache repair to stream inserts in partitions.
• Adds tests for repeated inserts and refreshes with pre-existing rows.

Confidence Score: 5/5

This looks safe to merge.

• No blocking issues found in the changed code.

Important Files Changed

Filename	Overview
src/palace/manager/sqlalchemy/refresh_equivalents.py	Routes cache rebuild inserts through chunked PostgreSQL conflict handling and bulk-deletes affected parent rows.
tests/manager/sqlalchemy/test_refresh_equivalents.py	Adds tests for conflict-tolerant cache inserts and refresh behavior when rows already exist.

_{Reviews (2): Last reviewed commit: "Fix UniqueViolation in equivalent_identi..." | Re-trigger Greptile}

[codecov]

Codecov Report

❌ Patch coverage is 76.47059% with 4 lines in your changes missing coverage. Please review.
✅ Project coverage is 93.46%. Comparing base (7061612) to head (6c6b376).
⚠️ Report is 12 commits behind head on main.

Files with missing lines	Patch %	Lines
...c/palace/manager/sqlalchemy/refresh_equivalents.py	76.47%	2 Missing and 2 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3553      +/-   ##
==========================================
- Coverage   93.46%   93.46%   -0.01%     
==========================================
  Files         512      512              
  Lines       46614    46618       +4     
  Branches     6352     6353       +1     
==========================================
+ Hits        43570    43571       +1     
- Misses       1968     1969       +1     
- Partials     1076     1078       +2     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[dbernstein]

Note: why we did not pursue "align the broker/lock timeouts"

While reviewing this fix we considered whether raising broker_transport_options_visibility_timeout (currently 1h) above the equivalents lock_timeout (2h) — or otherwise "aligning" the two — would be a better/additional defense than making the inserts idempotent. Conclusion: no, and it should not be resurrected later without this context.

1. visibility_timeout is a single global broker knob, applied to every queue and task. Putting it above 2h raises it for everything, and the observable side effect is a system-wide regression on the failure path: on a total connection loss (the broker-side redelivery path) a dropped message stays invisible for the full new window before any worker re-picks it. Broad cost for a narrow gain.

2. It doesn't target the redelivery path that actually causes the race. task_time_limit = 1800 (30 min, hard) means no single task execution can exceed 30 min, which is already under the 1h visibility window — so visibility-timeout-triggered redelivery of an in-flight hop is already structurally prevented. The concurrent chain comes from task_reject_on_worker_lost=True (worker OOM/time-limit-kill → immediate requeue), which is independent of visibility_timeout. Raising it does essentially nothing for this bug.

3. "Aligning" by lowering the locks instead would break a deliberate invariant. The 2h locks are sweep-guards, not per-hop guards; the invariant the codebase actually maintains is lock_timeout >= task_time_limit (see the rationale in custom_lists.py, _SWEEP_LOCK_TTL). Lowering those under the visibility window would reopen duplicate-sweep launches in importer/bibliotheca/custom_lists — a regression in unrelated tasks.

Net: reject_on_worker_lost redelivery — the real culprit — can't be locked or timed out of existence, which is exactly why idempotent (ON CONFLICT DO NOTHING) inserts are the correct primary defense. The per-false-redelivery wasted work is already bounded to a single 200-row batch, which these inserts now make harmless, so there is no timeout/config change here worth the blast radius.