Roll back dedup marker when a task enqueue fails

The key-value deduplication path reserved a marker before dispatching
to the queue but never undid it when the dispatch failed.  A transient
backend failure therefore left the marker behind, so the retry was
silently deduplicated against a task that had never reached the queue.

The cas claim now stores a unique token instead of a bare `true`, and a
failed dispatch conditionally clears it (cas succeeds only while the
stored value is still our token).  The conditional clear keeps a stale
rollback from deleting a marker that another concurrent enqueue has
already re-claimed.  A rollback that itself fails is logged and
swallowed so the original enqueue error still reaches the caller.

The enqueueMany requirement for deduplicated multi-item batches now
keys on whether deduplication is actually applied—a native queue or the
cas fallback—rather than on nativeDeduplication alone.  Under the
"open" fallback (no native dedup, no cas) no marker is taken, so the
batch fans out without deduplication instead of throwing.
ParallelMessageQueue likewise rejects a deduplicated batch when the
wrapped queue lacks enqueueMany, since fanning out cannot carry one key
atomically.

#798

Assisted-by: Claude Code:claude-opus-4-8

Author: 2chanhaeng

docs/manual/tasks.md

@@ -308,21 +308,33 @@ follow-ups add it.
 For `~Context.enqueueTaskMany()`, a single `deduplicationKey` applies to the
 whole batch: the batch enqueues as a unit or is skipped as a unit, never
 partially.  Per-item deduplication means calling `~Context.enqueueTask()` in
-a loop, each with its own key.  A queue that declares
-`~MessageQueue.nativeDeduplication` must also implement
-`~MessageQueue.enqueueMany()` to carry a multi-item batch's key as one unit;
-fanning the key out across separate `~MessageQueue.enqueue()` calls cannot drop
-a whole batch, so Fedify rejects that combination instead of silently leaking
-duplicates.
+a loop, each with its own key.  Deduplicating a multi-item batch requires the
+queue to implement `~MessageQueue.enqueueMany()` so the batch enqueues
+atomically—whether the check is native or the key–value fallback.  Fanning the
+key out across separate `~MessageQueue.enqueue()` calls cannot enqueue a whole
+batch as one unit: a native per-message key cannot cover it, and a key–value
+marker could not be rolled back cleanly if only some of the fanned-out enqueues
+failed.  So when deduplication is actually applied—a native queue, or a
+key–value store with `~KvStore.cas`—Fedify rejects a multi-item batch with a
+`deduplicationKey` on a queue without `~MessageQueue.enqueueMany()` instead of
+risking duplicates.  Under the `"open"` fallback (no native deduplication and no
+`cas`), no marker is taken, so the batch simply fans out without deduplication.
+
+This applies through `~ParallelMessageQueue` as well: wrapping a queue that
+lacks `~MessageQueue.enqueueMany()` does not make batch enqueue atomic, so a
+deduplicated multi-item batch on such a wrapper is likewise rejected rather than
+collapsed onto one message.
 
 > [!WARNING]
 > The key–value fallback is **best-effort, not transactional**.  The marker
-> write and the enqueue are separate operations, so a crash between them, the
-> `"open"` fallback under concurrency, a non-atomic third-party `~KvStore.cas`,
-> or reuse of a key within its TTL window can still admit a duplicate or
-> suppress a task.  Cleanup is by TTL expiry, not active deletion on handler
-> success.  Deployments needing strict guarantees use a queue with
-> `nativeDeduplication: true`, where the backend owns an atomic check.
+> write and the enqueue are separate operations.  Fedify rolls the marker back
+> when an enqueue fails, so a transient failure does not suppress the retry, but
+> a crash before that rollback, the `"open"` fallback under concurrency, a
+> non-atomic third-party `~KvStore.cas`, or reuse of a key within its TTL window
+> can still admit a duplicate or suppress a task.  Cleanup is otherwise by TTL
+> expiry, not active deletion on handler success.  Deployments needing strict
+> guarantees use a queue with `nativeDeduplication: true`, where the backend
+> owns an atomic check.
 
 
 Limitations

packages/fedify/src/federation/middleware.ts

@@ -3124,8 +3124,8 @@ export class ContextImpl<TContextData> implements Context<TContextData> {
     return this.#codec ??= new TaskCodec(this);
   }
 
-  get #enqueueTasks(): ReturnType<typeof enqueueTasks> {
-    return enqueueTasks(this) as ReturnType<typeof enqueueTasks>;
+  get #enqueueTasks() {
+    return enqueueTasks(this);
   }
 
   clone(data: TContextData): Context<TContextData> {

packages/fedify/src/federation/mq.test.ts

@@ -5,11 +5,13 @@ import {
   assertFalse,
   assertGreater,
   assertGreaterOrEqual,
+  assertRejects,
 } from "@std/assert";
 import { delay } from "es-toolkit";
 import {
   InProcessMessageQueue,
   type MessageQueue,
+  type MessageQueueEnqueueOptions,
   ParallelMessageQueue,
 } from "./mq.ts";
 
@@ -438,6 +440,92 @@ test("ParallelMessageQueue inherits nativeDeduplication", () => {
   assert(workers.nativeDeduplication);
 });
 
+test(
+  "ParallelMessageQueue forwards deduplicationKey to the wrapped queue",
+  async () => {
+    class RecordingQueue implements MessageQueue {
+      readonly nativeDeduplication = true;
+      readonly singles: (MessageQueueEnqueueOptions | undefined)[] = [];
+      readonly batches: (MessageQueueEnqueueOptions | undefined)[] = [];
+      enqueue(
+        _message: unknown,
+        options?: MessageQueueEnqueueOptions,
+      ): Promise<void> {
+        this.singles.push(options);
+        return Promise.resolve();
+      }
+      enqueueMany(
+        _messages: readonly unknown[],
+        options?: MessageQueueEnqueueOptions,
+      ): Promise<void> {
+        this.batches.push(options);
+        return Promise.resolve();
+      }
+      listen(): Promise<void> {
+        return Promise.resolve();
+      }
+    }
+
+    const inner = new RecordingQueue();
+    const workers = new ParallelMessageQueue(inner, 5);
+    await workers.enqueue({ x: 1 }, { deduplicationKey: "k1" });
+    await workers.enqueueMany([{ x: 1 }, { x: 2 }], { deduplicationKey: "k2" });
+    assertEquals(inner.singles[0]?.deduplicationKey, "k1");
+    assertEquals(inner.batches[0]?.deduplicationKey, "k2");
+  },
+);
+
+test(
+  "ParallelMessageQueue rejects a deduplicated batch when the wrapped queue " +
+    "lacks enqueueMany",
+  async () => {
+    class NoBulkQueue implements MessageQueue {
+      readonly nativeDeduplication = true;
+      readonly enqueued: unknown[] = [];
+      enqueue(message: unknown): Promise<void> {
+        this.enqueued.push(message);
+        return Promise.resolve();
+      }
+      listen(): Promise<void> {
+        return Promise.resolve();
+      }
+    }
+
+    const inner = new NoBulkQueue();
+    const workers = new ParallelMessageQueue(inner, 5);
+    await assertRejects(
+      () =>
+        workers.enqueueMany([{ x: 1 }, { x: 2 }], { deduplicationKey: "k" }),
+      TypeError,
+      "enqueueMany",
+    );
+    // It threw before enqueuing anything.
+    assertEquals(inner.enqueued.length, 0);
+  },
+);
+
+test(
+  "ParallelMessageQueue still fans out a non-deduplicated batch when the " +
+    "wrapped queue lacks enqueueMany",
+  async () => {
+    class NoBulkQueue implements MessageQueue {
+      readonly enqueued: unknown[] = [];
+      enqueue(message: unknown): Promise<void> {
+        this.enqueued.push(message);
+        return Promise.resolve();
+      }
+      listen(): Promise<void> {
+        return Promise.resolve();
+      }
+    }
+
+    const inner = new NoBulkQueue();
+    const workers = new ParallelMessageQueue(inner, 5);
+    await workers.enqueueMany([{ x: 1 }, { x: 2 }, { x: 3 }]);
+    assertEquals(inner.enqueued.length, 3);
+  },
+);
+
 const queues: Record<string, () => Promise<MessageQueue>> = {
   InProcessMessageQueue: () => Promise.resolve(new InProcessMessageQueue()),
 };

packages/fedify/src/federation/mq.ts

@@ -451,6 +451,15 @@ export class ParallelMessageQueue implements MessageQueue {
     options?: MessageQueueEnqueueOptions,
   ): Promise<void> {
     if (this.queue.enqueueMany == null) {
+      if (options?.deduplicationKey != null) {
+        throw new TypeError(
+          "Cannot enqueue a batch with a deduplicationKey: the wrapped queue " +
+            "does not implement enqueueMany, so ParallelMessageQueue would " +
+            "have to fan out to individual enqueue() calls that cannot share " +
+            "one deduplicationKey atomically.  Wrap a queue that implements " +
+            "enqueueMany instead.",
+        );
+      }
       const results = await Promise.allSettled(
         messages.map((message) => this.queue.enqueue(message, options)),
       );