Scope custom queue ID deduplication to each named queue

[fubhy]

Summary

Offering the same custom ID to two independently named memory or SQL queues silently drops the second queue's item, so that queue's consumer remains blocked.

Important

This PR starts with focused failing reproduction tests. Add the implementation fix to this same branch; CI is expected to fail until that fix is included.

Custom ID deduplication crosses named queues

Module: PersistedQueue
Audit ID: unstable-state-pq-1
Severity / confidence: high / high

What happens

Offering the same custom ID to two independently named memory or SQL queues silently drops the second queue's item, so that queue's consumer remains blocked.

Why it happens

The memory layer uses one factory-wide Set for every queue name, while the SQL table uniquely indexes id without queue_name. Redis correctly scopes its ID set by queue name.

Expected behavior

Deduplication applies when an ID already exists in the same queue; queue names define independent queues.

Relevant implementation

These links and excerpts are pinned to audit base c9b56ab507f224426ee8388dc450da447ec4715f.

• packages/effect/src/unstable/persistence/PersistedQueue.ts:70-72
• packages/effect/src/unstable/persistence/PersistedQueue.ts:297-320
• packages/effect/src/unstable/persistence/PersistedQueue.ts:800-801
• packages/effect/src/unstable/persistence/PersistedQueue.ts:814-815
• packages/effect/src/unstable/persistence/PersistedQueue.ts:857-863

View problematic code at packages/effect/src/unstable/persistence/PersistedQueue.ts:70-72

   * If an element with the same id already exists in the queue, it will not be
   * added again.
   */

View exact lines on GitHub

View problematic code at packages/effect/src/unstable/persistence/PersistedQueue.ts:297-320

  const ids = new Set<string>()
  const queues = new Map<string, {
    latch: Latch.Latch
    items: Set<Entry>
  }>()
  const getOrCreateQueue = (name: string) => {
    let queue = queues.get(name)
    if (!queue) {
      queue = {
        latch: Latch.makeUnsafe(false),
        items: new Set()
      }
      queues.set(name, queue)
    }
    return queue
  }

  return PersistedQueueStore.of({
    offer: (options) =>
      Effect.sync(() => {
        if (ids.has(options.id)) return
        ids.add(options.id)
        const queue = getOrCreateQueue(options.name)
        queue.items.add({ id: options.id, attempts: 0, element: options.element })

View exact lines on GitHub

View problematic code at packages/effect/src/unstable/persistence/PersistedQueue.ts:800-801

        id VARCHAR(36) NOT NULL,
        queue_name VARCHAR(100) NOT NULL,

View exact lines on GitHub

View problematic code at packages/effect/src/unstable/persistence/PersistedQueue.ts:814-815

        id VARCHAR(36) NOT NULL,
        queue_name VARCHAR(100) NOT NULL,

View exact lines on GitHub

View problematic code at packages/effect/src/unstable/persistence/PersistedQueue.ts:857-863

  yield* sql.onDialectOrElse({
    mssql: () =>
      sql`IF NOT EXISTS (SELECT * FROM sys.indexes WHERE name = N'idx_${tableName}_id')
        CREATE UNIQUE INDEX idx_${tableNameSql}_id ON ${tableNameSql} (id)`,
    mysql: () => sql`CREATE UNIQUE INDEX ${sql(`idx_${tableName}_id`)} ON ${tableNameSql} (id)`.pipe(Effect.ignore),
    orElse: () => sql`CREATE UNIQUE INDEX IF NOT EXISTS ${sql(`idx_${tableName}_id`)} ON ${tableNameSql} (id)`
  })

View exact lines on GitHub

Reproduction

pnpm test --run packages/effect/test/unstable/persistence/PersistedQueue.test.ts

Observed failure: The second queue's take fiber remained pending after the same custom ID was offered to both queues.

Implementation handoff

The initial reproduction tests on this branch are the regression specification for the implementation fix that should follow in this PR.

1. Start with the pinned implementation excerpts and the Why it happens analysis above.
2. Change the implementation so it satisfies the stated Expected behavior; do not weaken or remove the reproduction assertions.
3. Run the focused reproduction command(s) and confirm the observed failures become passing tests:

pnpm test --run packages/effect/test/unstable/persistence/PersistedQueue.test.ts

4. Run the affected package's existing tests, then the repository lint and type checks before requesting review.

Audit provenance

• Audit base: c9b56ab507f224426ee8388dc450da447ec4715f
• Reproduction base: c9b56ab507f224426ee8388dc450da447ec4715f
• Findings: unstable-state-pq-1
• Initial patch: focused reproduction tests; implementation fix pending

Closes EFF-443

[changeset-bot]

🦋 Changeset detected

Latest commit: b452221

The changes in this PR will be included in the next version bump.

This PR includes changesets to release 30 packages

Name	Type
effect	Patch
@effect/opentelemetry	Patch
@effect/platform-browser	Patch
@effect/platform-bun	Patch
@effect/platform-deno	Patch
@effect/platform-node-shared	Patch
@effect/platform-node	Patch
@effect/vitest	Patch
@effect/ai-anthropic	Patch
@effect/ai-openai-compat	Patch
@effect/ai-openai	Patch
@effect/ai-openrouter	Patch
@effect/atom-react	Patch
@effect/atom-solid	Patch
@effect/atom-vue	Patch
@effect/sql-clickhouse	Patch
@effect/sql-d1	Patch
@effect/sql-libsql	Patch
@effect/sql-mssql	Patch
@effect/sql-mysql2	Patch
@effect/sql-pg	Patch
@effect/sql-pglite	Patch
@effect/sql-sqlite-bun	Patch
@effect/sql-sqlite-do	Patch
@effect/sql-sqlite-node	Patch
@effect/sql-sqlite-react-native	Patch
@effect/sql-sqlite-wasm	Patch
@effect/docgen	Patch
@effect/doctest	Patch
@effect/openapi-generator	Patch

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

[pullfrog]

Important

The reproduction test is correct and currently fails as expected. The PR is mergeable once the implementation fix is added.

Reviewed changes

This PR currently contains only the regression specification for the cross-queue custom-id deduplication bug in PersistedQueue. The test adds a focused case that offers the same custom id to two named queues and asserts that the second queue can still take its item.

• Reproduction: packages/effect/test/unstable/persistence/PersistedQueue.test.ts adds a test that fails against the current implementation.
• Root cause: The memory store uses a factory-wide ids Set (PersistedQueue.ts:297) to deduplicate, while the SQL store creates a unique index on id without including queue_name. Redis already scopes the id set per queue name.
• Required follow-up: Update the memory store to deduplicate per queue name, and change the SQL unique index/conflict target from (id) to (id, queue_name) so duplicate ids are only rejected within the same queue.

I verified the test reproduces the issue: second.take never completes because the second offer with the shared id is dropped by the factory-wide memory deduplication.

", "comments": [{"path": "packages/effect/test/unstable/persistence/PersistedQueue.test.ts", "line": 11, "body": "The test accurately reproduces the bug: the memory store's factory-wide `ids` Set causes `second.offer` to be silently dropped when the same custom id has already been offered to `first`.

Note that the test currently only exercises layerStoreMemory; once the SQL implementation is fixed, consider adding an equivalent regression case for the SQL store so the unique-index/conflict-target change is also covered."}]}

  ^{｜ ⚠️ this action is pinned to a commit SHA, which freezes the cleanup step — switch to @v0 or keep the SHA fresh with Dependabot ｜ Fix it ➔ ｜ View workflow run ｜ Using Kimi K2 (free via Pullfrog for OSS) ｜ 𝕏}

[github-actions]

Bundle Size Analysis

Generated from PR build output; treat the content below as untrusted.

File Name	Current Size	Previous Size	Difference
basic.ts	7.06 KB	7.06 KB	0.00 KB (0.00%)
batching.ts	9.86 KB	9.86 KB	0.00 KB (0.00%)
brand.ts	6.34 KB	6.34 KB	0.00 KB (0.00%)
cache.ts	10.71 KB	10.71 KB	-0.00 KB (-0.02%)
config.ts	20.60 KB	20.60 KB	0.00 KB (0.00%)
differ.ts	20.20 KB	20.20 KB	0.00 KB (0.00%)
http-client.ts	21.58 KB	21.54 KB	+0.04 KB (+0.19%)
logger.ts	10.84 KB	10.84 KB	0.00 KB (0.00%)
metric.ts	8.98 KB	8.98 KB	0.00 KB (0.00%)
optic.ts	7.18 KB	7.18 KB	0.00 KB (0.00%)
pubsub.ts	14.99 KB	14.99 KB	0.00 KB (0.00%)
queue.ts	11.66 KB	11.66 KB	0.00 KB (0.00%)
schedule.ts	10.83 KB	10.83 KB	0.00 KB (0.00%)
schema-class.ts	19.14 KB	19.14 KB	0.00 KB (0.00%)
schema-fromJsonSchemaDocument.ts	28.96 KB	28.96 KB	0.00 KB (0.00%)
schema-representation-roundtrip.ts	25.29 KB	25.29 KB	0.00 KB (0.00%)
schema-string-transformation.ts	13.38 KB	13.38 KB	0.00 KB (0.00%)
schema-string.ts	10.94 KB	10.94 KB	0.00 KB (0.00%)
schema-template-literal.ts	15.17 KB	15.17 KB	0.00 KB (0.00%)
schema-toArbitraryLazy.ts	21.94 KB	21.94 KB	0.00 KB (0.00%)
schema-toCodeDocument.ts	24.34 KB	24.34 KB	0.00 KB (0.00%)
schema-toCodecJson.ts	19.18 KB	19.18 KB	0.00 KB (0.00%)
schema-toEquivalence.ts	19.01 KB	19.01 KB	0.00 KB (0.00%)
schema-toFormatter.ts	18.87 KB	18.87 KB	0.00 KB (0.00%)
schema-toJsonSchemaDocument.ts	22.60 KB	22.60 KB	0.00 KB (0.00%)
schema-toRepresentation.ts	19.52 KB	19.52 KB	0.00 KB (0.00%)
schema.ts	18.41 KB	18.41 KB	0.00 KB (0.00%)
stm.ts	12.63 KB	12.63 KB	0.00 KB (0.00%)
stream.ts	9.80 KB	9.80 KB	0.00 KB (0.00%)