PersistedQueue SQL store: a stale handler's retry releases the newer claim its own worker made, so the element is delivered again while in flight

What version of Effect is running?

effect 4.0.5, @effect/sql-sqlite-node 4.0.5, Node 26.11.1. The SQL store's claim and settlement code is unchanged on main (bbedfc7) and on v4/next-minor (14fba8b).

What steps can reproduce the bug?

The SQL store fences every settlement on acquired_by = workerId, and a store instance has one workerId. If a handler's lock lapses (the database is unreachable for longer than lockExpiration, so refreshes fail) and the same store's poll re-claims the row for another of its takers, the first handler's settlement still matches the row. When that handler then fails, its retry clears the newer claim, and the element is delivered a third time while the newer handler is still running. If the newer claim belongs to another store, the same retry matches nothing and the claim survives.

This test, added inside suiteWith in packages/effect/test/persistence/PersistedQueueSqlTest.ts (with suiteWith's body made a block and Latch and Schedule imported), fails on main for the same store and passes for another store (pnpm test --run packages/sql/sqlite-node/test/Persistence.test.ts -t "PersistedQueue SQL claims"). Like the Redis ownership tests in #9229, it keeps refreshes paused on the TestClock while the database clock runs:

// Lock refreshes run on the virtual clock, so they stay paused until the
// test ticks it, like a worker whose refreshes keep failing while the
// database is unavailable for longer than lockExpiration. The database clock
// keeps running, so the lock lapses while the first handler is still running.
testApi.layer(Layer.empty, { concurrent: false, timeout: "90 seconds" })("PersistedQueue SQL claims", (it) => {
  const makeOwnershipQueue = Effect.fnUntraced(function*(tableName: string) {
    const store = yield* PersistedQueue.makeStoreSql({
      tableName,
      pollInterval: "100 millis",
      lockRefreshInterval: "100 millis",
      lockExpiration: "1 second"
    })
    const factory = yield* PersistedQueue.makeFactory.pipe(
      Effect.provideService(PersistedQueue.PersistedQueueStore, store)
    )
    return yield* factory.make({ name: tableName, schema: Schema.String, retrySchedule: Schedule.spaced(0) })
  })

  for (const sameStore of [true, false]) {
    it.effect(
      `a stale failure does not release a newer claim from ${sameStore ? "the same" : "another"} store`,
      () =>
        Effect.gen(function*() {
          const tableName = sameStore ? "effect_queue_stale_same_store" : "effect_queue_stale_other_store"
          const stale = yield* makeOwnershipQueue(tableName)
          const queue = sameStore ? stale : yield* makeOwnershipQueue(tableName)
          yield* stale.offer("job", { id: "job-1" })

          const aTaken = Latch.makeUnsafe()
          const aRelease = Latch.makeUnsafe()
          const aFiber = yield* stale.take(() =>
            aTaken.open.pipe(
              Effect.andThen(aRelease.await),
              Effect.andThen(Effect.fail("boom"))
            )
          ).pipe(Effect.forkScoped)
          yield* aTaken.await

          // the lock lapses in the database (whole seconds on some dialects)
          yield* Effect.sleep("2500 millis").pipe(TestClock.withLive)

          const bTaken = Latch.makeUnsafe()
          const replacement = yield* queue.take((_, { attempts }) =>
            Effect.gen(function*() {
              assert.strictEqual(attempts, 2)
              yield* bTaken.open
              return yield* Effect.never
            })
          ).pipe(Effect.forkScoped)
          yield* bTaken.await.pipe(Effect.timeout("5 seconds"), TestClock.withLive)

          // only the replacement is live now: keep its lock refreshed
          yield* Effect.sleep("100 millis").pipe(
            TestClock.withLive,
            Effect.andThen(TestClock.adjust("100 millis")),
            Effect.forever,
            Effect.forkScoped
          )

          // the stale handler fails while the replacement is still running
          yield* aRelease.open
          yield* Fiber.await(aFiber)

          const duplicate = yield* queue.take((_, metadata) => Effect.succeed(metadata)).pipe(Effect.forkScoped)
          yield* Effect.sleep("1500 millis").pipe(TestClock.withLive)
          assert.isUndefined(duplicate.pollUnsafe())

          // once the replacement is interrupted the element is delivered again
          yield* Fiber.interrupt(replacement)
          const redelivered = yield* Fiber.join(duplicate).pipe(Effect.timeout("5 seconds"), TestClock.withLive)
          assert.strictEqual(redelivered.attempts, 2)
        }),
      { timeout: 30000 }
    )
  }
})

// SQL PersistedQueue: the database is unavailable for longer than
// lockExpiration while a handler (h1) keeps running. When it comes back, the
// worker's poll can re-claim the element before its lock refresh renews it, so
// another taker of the same worker starts h2. When h1 then fails, its retry
// passes the `acquired_by = workerId` check, releases h2's claim, and the
// element is delivered a third time (h3) while h2 is still running.
//
//   node stale-settlement.ts same-worker   # every taker on one store (one worker)
//   node stale-settlement.ts two-workers   # control: the other takers are on a second store
import { SqliteClient } from "@effect/sql-sqlite-node"
import { Effect, References, Schema } from "effect"
import { PersistedQueue } from "effect/persistence"
import { SqlClient } from "effect/sql"
import { mkdtempSync } from "node:fs"
import { tmpdir } from "node:os"
import { join } from "node:path"
import { DatabaseSync } from "node:sqlite"

const mode = process.argv[2] === "two-workers" ? "two-workers" : "same-worker"
const filename = join(mkdtempSync(join(tmpdir(), "pq-")), "queue.db")
const t0 = Date.now()
const log = (...args: Array<unknown>) => console.log(`${((Date.now() - t0) / 1000).toFixed(1)}s`, ...args)

const running = new Set<string>()
const deliveries: Array<string> = []

const makeWorker = Effect.fnUntraced(function*(worker: string, concurrency: number) {
  const store = yield* PersistedQueue.makeStoreSql({
    lockExpiration: "2 seconds",
    lockRefreshInterval: "500 millis",
    pollInterval: "100 millis"
  })
  const factory = yield* PersistedQueue.makeFactory.pipe(
    Effect.provideService(PersistedQueue.PersistedQueueStore, store)
  )
  const queue = yield* factory.make({ name: "jobs", schema: Schema.Struct({ n: Schema.Number }) })
  const handler = (_: { n: number }, { attempts }: { attempts: number }) => {
    const name = `${worker}/attempt-${attempts}`
    return Effect.gen(function*() {
      deliveries.push(name)
      log(`${name} starts; also running: [${[...running].join(", ")}]`)
      running.add(name)
      yield* Effect.sleep(attempts === 1 ? "7 seconds" : "8 seconds")
      if (attempts === 1) {
        log(`${name} fails`)
        return yield* Effect.fail("boom")
      }
      log(`${name} succeeds`)
    }).pipe(Effect.ensuring(Effect.sync(() => running.delete(name))))
  }
  for (let i = 0; i < concurrency; i++) {
    yield* queue.take(handler).pipe(Effect.ignore, Effect.forever, Effect.forkScoped)
  }
  return queue
})

const program = Effect.gen(function*() {
  const sql = yield* SqlClient.SqlClient
  const row = sql<{ state: string; attempts: number; held: number }>`
    SELECT state, attempts, acquired_by IS NOT NULL AS held FROM effect_queue`.pipe(
    Effect.map((rows) => JSON.stringify(rows[0]))
  )

  const w1 = yield* makeWorker("w1", mode === "same-worker" ? 3 : 1)
  if (mode === "two-workers") yield* makeWorker("w2", 2)

  yield* w1.offer({ n: 1 }, { id: "job-1" })
  yield* Effect.sleep(process.env.LOCK_AT ?? "750 millis")

  // Another connection holds the write lock for 4s (longer than
  // lockExpiration): lock refreshes and polls fail until it is released.
  log("database unavailable for 4s")
  const other = new DatabaseSync(filename)
  other.exec("BEGIN IMMEDIATE")
  yield* Effect.sleep("4 seconds")
  other.exec("COMMIT")
  other.close()
  log("database back")

  yield* Effect.sleep("2 seconds")
  log("row before h1 settles:", yield* row)
  yield* Effect.sleep("3 seconds")
  log("row after h1 settled:", yield* row)
  yield* Effect.sleep("7 seconds")
  log("row at the end:", yield* row)
  console.log(`RESULT ${mode}: deliveries=${deliveries.join(",")}`)
}).pipe(Effect.scoped)

Effect.runPromise(program.pipe(
  Effect.provide(SqliteClient.layer({ filename, busyTimeout: 0 })),
  // the failed refreshes and polls log warnings while the database is locked
  Effect.provideService(References.MinimumLogLevel, "Error")
)).then(() => process.exit(0), (e) => {
  console.error(e)
  process.exit(1)
})

0.0s w1/attempt-1 starts; also running: []
0.8s database unavailable for 4s
4.8s database back
4.9s w1/attempt-2 starts; also running: [w1/attempt-1]
6.8s row before h1 settles: {"state":"pending","attempts":2,"held":1}
7.0s w1/attempt-1 fails
7.9s w1/attempt-3 starts; also running: [w1/attempt-2]
9.8s row after h1 settled: {"state":"pending","attempts":3,"held":1}
12.9s w1/attempt-2 succeeds
15.9s w1/attempt-3 succeeds
16.8s row at the end: {"state":"completed","attempts":3,"held":0}
RESULT same-worker: deliveries=w1/attempt-1,w1/attempt-2,w1/attempt-3

0.0s w1/attempt-1 starts; also running: []
0.8s database unavailable for 4s
4.8s database back
4.9s w2/attempt-2 starts; also running: [w1/attempt-1]
6.8s row before h1 settles: {"state":"pending","attempts":2,"held":1}
7.0s w1/attempt-1 fails
9.8s row after h1 settled: {"state":"pending","attempts":2,"held":1}
12.9s w2/attempt-2 succeeds
16.8s row at the end: {"state":"completed","attempts":2,"held":0}
RESULT two-workers: deliveries=w1/attempt-1,w2/attempt-2

What is the expected behavior?

A settlement affects only the claim it was taken under. Once the row has been re-claimed, the stale handler's retry leaves the newer claim in place, as it already does when that claim is held by another store. This is the contract #9229 (for #9220) gives the Redis store ("a stale handler can no longer release another live claim"), including its "same-store stale success preserves the replacement and its refresh registration" test.

What do you see instead?

duplicate receives { id: 'job-1', attempts: 3 } while replacement still holds the element (6 of 6 runs). In the program, w1/attempt-3 starts while w1/attempt-2 is still running. With the other takers on a second store, nothing is delivered after attempt-1 fails.

Additional information

At 4.0.5:

• One workerId per store: L1401. complete, fail, retry and interrupt match only sequence and acquired_by = workerId: L1509-L1552.
• Every dialect's poll re-claims a lapsed row with the same workerId (only SQLite was run; L1608-L1691), so a re-claim by the same store is indistinguishable from the original claim.
• Each settlement also deletes the sequence from the refresh set that the newer claim shares: L1502-L1506.

#9220 found the SQL store unaffected; that holds only when the newer claim is on another store.

While the database is down, the poll and the refresh both retry every 500 ms, and whichever runs first after it recovers decides the outcome. In the program the poll won in 13 of 13 runs where the outage began between two refreshes (555-902 ms after the offer), and the refresh won in 11 of 11 runs where it began at 500 ms (a refresh tick) or at 930-990 ms, just before the next. An event-loop stall alone did not reproduce it: after the stall the refresh renewed the lock before the poll ran, in every run.

Fix idea: fence settlements on the acquisition rather than the worker (a per-claim token written by the poll), and key refresh registrations by acquisition, as #9229 did for the Redis store.