Skip to content

Fix redis worker debounce #1966

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Apr 22, 2025
Merged

Fix redis worker debounce #1966

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3c753ec
Added some Redis worker debounce tests (one failing that reproduces a…
matt-aitken Apr 22, 2025
48f19f0
Added some tests for acking
matt-aitken Apr 22, 2025
470cb2f
Added a deduplicationKey to prevent acking when items are queued
matt-aitken Apr 22, 2025
72cbabe
The worker passes the deduplicationKey back in for acking
matt-aitken Apr 22, 2025
be02fe6
Improved logs and removed events from test
matt-aitken Apr 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 39 additions & 8 deletions packages/redis-worker/src/queue.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ export type QueueItem<TMessageCatalog extends MessageCatalogSchema> = {
visibilityTimeoutMs: number;
attempt: number;
timestamp: Date;
deduplicationKey?: string;
};

export type AnyQueueItem = {
Expand All @@ -36,6 +37,7 @@ export type AnyQueueItem = {
visibilityTimeoutMs: number;
attempt: number;
timestamp: Date;
deduplicationKey?: string;
};

export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
Expand Down Expand Up @@ -98,11 +100,13 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
}): Promise<void> {
try {
const score = availableAt ? availableAt.getTime() : Date.now();
const deduplicationKey = nanoid();
const serializedItem = JSON.stringify({
job,
item,
visibilityTimeoutMs,
attempt,
deduplicationKey,
});

const result = await this.redis.enqueueItem(
Expand Down Expand Up @@ -136,7 +140,7 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
return [];
}

const dequeuedItems = [];
const dequeuedItems: Array<QueueItem<TMessageCatalog>> = [];

for (const [id, serializedItem, score] of results) {
const parsedItem = JSON.parse(serializedItem) as any;
Expand Down Expand Up @@ -186,6 +190,7 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
visibilityTimeoutMs,
attempt: parsedItem.attempt ?? 0,
timestamp,
deduplicationKey: parsedItem.deduplicationKey,
});
}

Expand All @@ -200,14 +205,22 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
}
}

async ack(id: string): Promise<void> {
async ack(id: string, deduplicationKey?: string): Promise<void> {
try {
await this.redis.ackItem(`queue`, `items`, id);
const result = await this.redis.ackItem(`queue`, `items`, id, deduplicationKey ?? "");
if (result === 0) {
this.logger.error(`SimpleQueue ${this.name}.ack(): ack operation returned 0`, {
queue: this.name,
id,
deduplicationKey,
});
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Escalate ACK mismatches instead of silently logging them

ack() only logs when the Lua script returns 0 (deduplication‑key mismatch) or -1 (item missing) but still resolves void.
Call‑sites (e.g. Worker.processItem) will therefore treat the job as successfully acknowledged even though it remains in Redis, leading to:

  • the job becoming “stuck” until TTL‑cleanup or a manual redrive
  • misleading success metrics / traces

Consider propagating the return value (or throwing) so that the caller can retry, reschedule or surface an error.

- const result = await this.redis.ackItem(...);
- if (result === 0) {
-   this.logger.error(/* … */);
- }
+ const result = await this.redis.ackItem(...);
+ if (result <= 0) {
+   throw new Error(
+     `ACK failed for id=${id} (result=${result}, dedupKey=${deduplicationKey ?? "∅"})`
+   );
+ }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
async ack(id: string, deduplicationKey?: string): Promise<void> {
try {
await this.redis.ackItem(`queue`, `items`, id);
const result = await this.redis.ackItem(`queue`, `items`, id, deduplicationKey ?? "");
if (result === 0) {
this.logger.error(`SimpleQueue ${this.name}.ack(): ack operation returned 0`, {
queue: this.name,
id,
deduplicationKey,
});
}
async ack(id: string, deduplicationKey?: string): Promise<void> {
try {
const result = await this.redis.ackItem(`queue`, `items`, id, deduplicationKey ?? "");
if (result <= 0) {
throw new Error(
`ACK failed for id=${id} (result=${result}, dedupKey=${deduplicationKey ?? "∅"})`
);
}
// …rest of the logic…

} catch (e) {
this.logger.error(`SimpleQueue ${this.name}.ack(): error acknowledging item`, {
queue: this.name,
error: e,
id,
deduplicationKey,
});
throw e;
}
Expand Down Expand Up @@ -367,15 +380,32 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
this.redis.defineCommand("ackItem", {
numberOfKeys: 2,
lua: `
local queue = KEYS[1]
local items = KEYS[2]
local queueKey = KEYS[1]
local itemsKey = KEYS[2]
local id = ARGV[1]
local deduplicationKey = ARGV[2]

redis.call('ZREM', queue, id)
redis.call('HDEL', items, id)
-- Get the item from the hash
local item = redis.call('HGET', itemsKey, id)
if not item then
return -1
end

-- Only check deduplicationKey if a non-empty one was passed in
if deduplicationKey and deduplicationKey ~= "" then
local success, parsed = pcall(cjson.decode, item)
if success then
if parsed.deduplicationKey and parsed.deduplicationKey ~= deduplicationKey then
return 0
end
end
end

-- Remove from sorted set and hash
redis.call('ZREM', queueKey, id)
redis.call('HDEL', itemsKey, id)
return 1
`,
`,
Comment on lines 384 to +412
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Lua script: avoid false‑positives when no deduplicationKey is stored

If the stored item predates this feature it will have no deduplicationKey.
When an old item is replaced by a new one with a key while another worker still holds the old key (""), the current logic bypasses the check (deduplicationKey argument is empty) and deletes the new item – re‑introducing the race.

Guard by returning 0 whenever the stored item has a key but none is supplied:

- if deduplicationKey and deduplicationKey ~= "" then
+ if deduplicationKey == nil or deduplicationKey == "" then
+   local parsed = cjson.decode(item)
+   if parsed.deduplicationKey then
+     return 0
+   end
+ else
   -- existing path

This keeps backward‑compatibility (old items without a key are still ACKed) while protecting new ones.

Committable suggestion skipped: line range outside the PR's diff.

});

this.redis.defineCommand("moveToDeadLetterQueue", {
Expand Down Expand Up @@ -468,6 +498,7 @@ declare module "@internal/redis" {
queue: string,
items: string,
id: string,
deduplicationKey: string,
callback?: Callback<number>
): Result<number, Context>;

Expand Down
Loading
Loading