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

Abierto
  • ¿Algo no está bien?
#10,313 0 comentarios 0 reacciones 1 asignado Ver en GitHub

Los mantenedores suelen responder en 1 día

@effect-bot ya está trabajando en esto.

Desde el 2/11/2026.

  • #10316 de @effect-bot — abierto
  • #10320 de @EmilEinarsen — abierto

Evaluación

Este issue todavía no se ha evaluado.

Descripción

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.

Lenguaje dominante
TypeScript
Estrellas
16.7k
Forks
808
Merge medio
11 h 3 min
PR fusionados (30 d)
839

Preparar el entorno

Este proyecto no incluye contenedor de desarrollo, Dockerfile ni guía de contribución, así que la configuración corre por tu cuenta: empieza por su README y consulta nuestra guía para la primera contribución para los pasos generales.

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de Effect-TS/effect

Todos los issues de Effect-TS/effect

Issues similares

Más issues de TypeScript

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.