Fix UniqueViolation in equivalent_identifiers_refresh cache rebuild (PP-4786)

[jonathangreen]

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

[claude]

Claude finished @jonathangreen's task in 6m 13s —— View job

---

Code Review

• Read the PR diff and changed files
• Analyze refresh_equivalents.py changes
• Analyze test coverage
• Post review feedback

Summary

This is a clean, well-targeted fix. Routing every recursiveequivalentscache insert through _insert_cache_rows with INSERT ... ON CONFLICT (parent_identifier_id, identifier_id) DO NOTHING correctly neutralizes the production IntegrityError: because each parent's chain is recomputed from the same source equivalencies, any row a concurrent writer (the Identifier creation listener or a redelivered refresh chain) commits under us holds the exact value we were about to write, so skipping the conflict converges rather than aborting. I verified the surrounding assumptions that make this safe: is_parent is a Computed column (identifier.py:1357), so the Core pg_insert bypassing the ORM doesn't drop any Python-side default; the recursive fn_recursive_equivalents query dedupes via UNION, so a single VALUES batch never contains duplicate keys; and add_identity_equivalents interleaving Core inserts with the open yield_per server-side cursor is exercised end-to-end by test_sweeps_across_multiple_partitions. The _delete_cache_rows/_insert_cache_rows sorted-order deadlock rationale is sound, and the switch from accumulated ORM add_all to per-partition Core inserts is a memory improvement for the full-table sweep. The already-resolved discussion about not chunking the DELETE ... IN (...) is correct — psycopg2 interpolates parameters client-side, so the 65 535-parameter cap doesn't apply, and parent_ids is bounded by the batch anyway.

The new tests (TestInsertCacheRows, TestDeleteCacheRows, test_tolerates_rows_the_delete_did_not_remove, and the empty/no-op/multi-partition/multi-chunk cases) give good regression coverage of the race and the batching paths. I found no correctness, performance, or convention issues that warrant a change — nice work.

[greptile-apps]

Greptile Summary

This PR fixes a production IntegrityError (UniqueViolation) in the equivalent_identifiers_refresh Celery task by routing all RecursiveEquivalencyCache inserts through a new _insert_cache_rows helper that uses PostgreSQL INSERT ... ON CONFLICT (parent_identifier_id, identifier_id) DO NOTHING. It also refactors process_identifier_ids to issue a single bulk DELETE ... WHERE parent_identifier_id IN (...) instead of per-parent deletes, and switches add_identity_equivalents from per-row session.add() calls to the same chunked, conflict-tolerant insert path.

• _insert_cache_rows: Sorts rows by (parent_identifier_id, identifier_id) before batched insertion to prevent deadlocks between concurrent writers, and uses ON CONFLICT DO NOTHING to skip rows already committed by the Identifier creation listener or a re-delivered Celery task.
• _delete_cache_rows: Collapses per-parent DELETE calls into a single WHERE … IN (...) statement and short-circuits on an empty parent set.
• Tests: New TestInsertCacheRows and TestDeleteCacheRows classes cover empty-input guards, conflict-skip semantics, multi-chunk batching, and multi-partition streaming; test_tolerates_rows_the_delete_did_not_remove directly reproduces the production failure scenario.

Confidence Score: 5/5

Safe to merge — the change is narrowly scoped to the cache rebuild path, the fix is mechanically correct, and the new tests directly reproduce and verify the production failure scenario.

The ON CONFLICT DO NOTHING target matches the UniqueConstraint(parent_identifier_id, identifier_id) defined in the model, the sorted insertion order is the correct deadlock-prevention technique, itertools.batched is available in the project's minimum Python 3.12 baseline, and the computed is_parent column requires no explicit value in the INSERT dicts. Tests cover all new code paths.

No files require special attention.

Important Files Changed

Filename	Overview
src/palace/manager/sqlalchemy/refresh_equivalents.py	Introduces _insert_cache_rows (ON CONFLICT DO NOTHING, sorted for deadlock safety, chunked at 10k) and _delete_cache_rows helpers; refactors process_identifier_ids to bulk delete then bulk insert, and add_identity_equivalents to use partitioned streaming with conflict-tolerant inserts. Logic is correct and well-targeted at the production race condition.
tests/manager/sqlalchemy/test_refresh_equivalents.py	Adds TestInsertCacheRows, TestDeleteCacheRows, and several new cases to existing classes; covers conflict skip, empty inputs, multi-chunk batching, multi-partition cursor interleaving, and the monkeypatched delete-disabled conflict scenario that directly reproduces the production IntegrityError.

_{Reviews (3): Last reviewed commit: "Update comments [skip claude]" | Re-trigger Greptile}

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 93.47%. Comparing base (bf23006) to head (cf62739).

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #3564   +/-   ##
=======================================
  Coverage   93.46%   93.47%           
=======================================
  Files         512      512           
  Lines       46611    46619    +8     
  Branches     6352     6353    +1     
=======================================
+ Hits        43567    43575    +8     
  Misses       1968     1968           
  Partials     1076     1076           

☔ 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.

[jonathangreen]

I did a bit of cleanup here and re-opened the PR not from a fork, so it looks like this PR is from me, but its really @dbernstein in #3553. So I'm going to merge this one since he is out on vacation.