PersistedQueue: offer option that de-duplicates an id only while it is pending or processing

Chiusa
  • Qualcosa non va?
#10,313 0 commenti 0 reazioni 1 assegnatario Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

@effect-bot ci sta già lavorando.

Dal 2/11/2026.

  • #10316 di @effect-bot — integrata
  • #10320 di @EmilEinarsen — chiusa senza merge

Valutazione

Questa issue non è ancora stata valutata.

Descrizione

Use case

A common queue pattern is "at most one job per key at a time, but run again when the next trigger arrives after the previous one finished". For example, a recompute job keyed by an entity id. Ten triggers in a burst should coalesce into one pending job. A trigger that arrives after that job has completed should enqueue a new one.

Today offer with an id de-duplicates until layerCleanup deletes the settled element (default timeToLive is 30 days). So a later trigger with the same id is silently dropped, and the work doesn't run again.

Current workaround

Before offering, rename the settled row's id so the ON CONFLICT (id, queue_name) DO NOTHING in the SQL store no longer matches:

UPDATE effect_queue SET id = id || '#' || sequence
WHERE queue_name = $1 AND id = $2 AND state IN ('completed', 'failed')

This works, but it writes to the store's own table from outside the module, so it depends on the table's schema and breaks if the store's migrations change it.

Proposal

An option on offer, for example:

queue.offer(value, { id, deduplicate: "whilePending" }) // default stays "untilCleanup"

With "whilePending", an offer is ignored while an element with that id is pending or processing, and enqueues anew once it is completed (and possibly failed, see below).

For Postgres, the offer could become:

INSERT INTO effect_queue (id, queue_name, element, state, attempts, visible_at, created_at, updated_at)
VALUES ($id, $name, $element, 'pending', 0, $visible_at, now(), now())
ON CONFLICT (id, queue_name) DO UPDATE SET
  element = EXCLUDED.element, state = 'pending', attempts = 0, last_failure = NULL,
  acquired_at = NULL, acquired_by = NULL,
  visible_at = EXCLUDED.visible_at, updated_at = EXCLUDED.updated_at
WHERE effect_queue.state IN ('completed', 'failed')

The Redis and memory stores would need the same check on the stored state.

Open questions

  • Failed elements: they are the dead-letter record, so reusing that row on a new offer erases the previous failure. An alternative is to keep the failed row and insert a new one, which needs the unique key to allow several settled rows per id. Or "whilePending" could apply to completed elements only.
  • Ordering: reusing the row keeps the old sequence, so among elements with the same visible_at the re-offered one sorts first. A delete followed by an insert would avoid that.

I'm happy to open a PR once you're fine with the shape.

Lingua principale
TypeScript
Stelle
16.7k
Fork
808
Merge medio
11h 31m
PR unite (30g)
834

Preparare l'ambiente

Questo progetto non fornisce container di sviluppo, Dockerfile né guida per i contributori, quindi l'ambiente è a tuo carico: parti dal suo README e consulta la nostra guida al primo contributo per i passaggi generali.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di Effect-TS/effect

Tutte le issue di Effect-TS/effect

Issue simili

Altre issue su TypeScript

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.