feat(queries): add on_conflict option to enqueue for dedupe conflicts#681
Open
janbjorge wants to merge 5 commits into
Open
feat(queries): add on_conflict option to enqueue for dedupe conflicts#681janbjorge wants to merge 5 commits into
janbjorge wants to merge 5 commits into
Conversation
Batch enqueues with dedupe_key currently fail atomically: one duplicate
raises DuplicateJobError and nothing is inserted. Add an on_conflict
parameter ('raise' default, 'skip') to Queries, SyncQueries and
InMemoryQueries enqueue. Skip mode uses ON CONFLICT DO NOTHING against
the partial dedupe index and returns one entry per input, None marking
skipped positions, so callers keep the 1:1 input mapping.
The enqueue query now feeds the insert from UNNEST ... WITH ORDINALITY
ordered by input position and returns (id, dedupe_key) in id order;
align_ids_with_dedupe_keys maps rows back to input positions in O(n).
Benchmarked at parity with the previous query shape up to 100k-row
batches.
The in-memory adapter previously double-inserted within-batch duplicate
keys where PostgreSQL raises; it now raises, matching the database
behavior.
Also expose --dedupe-key and --on-conflict on the pgq queue CLI
command.
Closes #678
Collapse the queue command's two enqueue branches into one skip-mode call: for a single job a None result is exactly the duplicate case, so on_conflict only decides the message and exit code. Drops the try/except and the duplicated six-argument call. Move env_from_dsn into test/helpers.py and reuse it from the CLI tests instead of keeping per-file copies. Flatten the in-memory duplicate pre-check and deduplicate the row-verification SQL in the skip-mode tests.
Every branch on on_conflict now ends in assert_never, so adding a value to the OnConflict literal or the CLI enum fails mypy at each decision point instead of silently falling into the raise path. Verified: a temporary third value produces five arg-type errors, one per site. Rewrite align_ids_with_dedupe_keys from a hand-rolled merge walk to a dict lookup plus iterator: non-null keys are unique among inserted rows (pop hands the id to the first within-batch occurrence) and null-key rows pair positionally. Drops the row-order dependence for keyed rows.
Add edge cases (empty batch, triplicate within-batch key, repeated key conflicting with an existing job, null-key ordering) and a seeded reference-model test: simulate_enqueue mirrors the skip-mode INSERT's ON CONFLICT DO NOTHING semantics and ascending id assignment, and one thousand generated batches assert the alignment reproduces it exactly.
asyncio_mode is auto; async def is enough. Aligns the new dedupe tests with the documented convention instead of the file's legacy decorators.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Batch enqueues with
dedupe_keyfail atomically today: one duplicate raisesDuplicateJobErrorand nothing is inserted. This addson_conflict: Literal["raise", "skip"] = "raise"toenqueue()acrossQueries,SyncQueries,InMemoryQueries, and the repository port — the default is unchanged, whileskipinserts the non-duplicates and reports the rest.Noneat skipped positions, so callers keep the 1:1 mapping and know exactly what was inserted.ON CONFLICT ... DO NOTHINGagainst the partial dedupe index keeps concurrent enqueues correct. The insert reads fromUNNEST ... WITH ORDINALITYordered by input position;align_ids_with_dedupe_keysmaps the returned(id, dedupe_key)rows back to input positions in O(n). Benchmarked at parity with the previous query shape up to 100k-row batches.on_conflictends inassert_never, so a future third mode fails mypy at each decision point instead of silently falling into the raise path.pgq queuegains--dedupe-keyand--on-conflict(skip exits 0 with a "Skipped" message, raise exits 1).queuedlog entry. Docs updated: reliability (idempotency), quickstart, CLI reference, completion tracking, ports reference.Closes #678
Testing
make checkpassed (804 passed, 11 skipped; ruff, mypy strict, import-linter)gatherON CONFLICT DO NOTHINGsemantics independentlyChecklist