Skip to content

feat(queries): add on_conflict option to enqueue for dedupe conflicts#681

Open
janbjorge wants to merge 5 commits into
mainfrom
feat/enqueue-on-conflict
Open

feat(queries): add on_conflict option to enqueue for dedupe conflicts#681
janbjorge wants to merge 5 commits into
mainfrom
feat/enqueue-on-conflict

Conversation

@janbjorge

@janbjorge janbjorge commented Jul 11, 2026

Copy link
Copy Markdown
Owner

Batch enqueues with dedupe_key fail atomically today: one duplicate raises DuplicateJobError and nothing is inserted. This adds on_conflict: Literal["raise", "skip"] = "raise" to enqueue() across Queries, SyncQueries, InMemoryQueries, and the repository port — the default is unchanged, while skip inserts the non-duplicates and reports the rest.

job_ids = await queries.enqueue(
    entrypoints,
    payloads,
    priorities,
    dedupe_key=dedupe_keys,
    on_conflict="skip",
)
# One entry per input, e.g. [JobId(11), None, JobId(12)]
  • Shape-preserving return: skip mode returns one entry per input with None at skipped positions, so callers keep the 1:1 mapping and know exactly what was inserted.
  • Enforced in SQL: ON CONFLICT ... DO NOTHING against the partial dedupe index keeps concurrent enqueues correct. The insert reads from UNNEST ... WITH ORDINALITY ordered by input position; align_ids_with_dedupe_keys maps 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.
  • Exhaustive branching: every branch on on_conflict ends in assert_never, so a future third mode fails mypy at each decision point instead of silently falling into the raise path.
  • In-memory parity fix: the in-memory adapter silently double-inserted within-batch duplicate keys where PostgreSQL raises; it now raises, matching the database.
  • CLI: pgq queue gains --dedupe-key and --on-conflict (skip exits 0 with a "Skipped" message, raise exits 1).
  • Skipped rows produce no job and no queued log entry. Docs updated: reliability (idempotency), quickstart, CLI reference, completion tracking, ports reference.

Closes #678

Testing

  • make check passed (804 passed, 11 skipped; ruff, mypy strict, import-linter)
  • New Postgres integration tests: single/batch shape preservation, interleaved null keys (guards the id-to-position mapping against real row contents), within-batch duplicates, no log rows for skipped, concurrent-skip race via pool + gather
  • New in-memory and CLI tests; the alignment helper has 12 unit tests including a seeded 1000-case reference-model test that mirrors the ON CONFLICT DO NOTHING semantics independently
  • Benchmarked old vs new enqueue query at 1–100k batch sizes (parity; skip adds ~10% for the arbiter index probes)

Checklist

  • I have read the Contributing Guide
  • I have added or updated tests
  • I have updated documentation if necessary

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.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Configurable behavior for duplicates in enqueue

1 participant