Skip to content

Commit

Permalink
fix(events): do not publish removed event on non-existent jobs (#2227)
Browse files Browse the repository at this point in the history
  • Loading branch information
roggervalf committed Oct 24, 2023
1 parent 177799e commit 67de4fb
Show file tree
Hide file tree
Showing 2 changed files with 26 additions and 4 deletions.
6 changes: 3 additions & 3 deletions src/commands/removeJob-1.lua
Original file line number Diff line number Diff line change
Expand Up @@ -53,9 +53,9 @@ local function removeJob( prefix, jobId, parentKey, removeChildren)

local prev = removeJobFromAnyState(prefix, jobId)

rcall("DEL", jobKey, jobKey .. ":logs", jobKey .. ":dependencies", jobKey .. ":processed")

rcall("XADD", prefix .. "events", "*", "event", "removed", "jobId", jobId, "prev", prev);
if rcall("DEL", jobKey, jobKey .. ":logs", jobKey .. ":dependencies", jobKey .. ":processed") > 0 then
rcall("XADD", prefix .. "events", "*", "event", "removed", "jobId", jobId, "prev", prev)
end
end

local prefix = KEYS[1]
Expand Down
24 changes: 23 additions & 1 deletion tests/test_events.ts
Original file line number Diff line number Diff line change
Expand Up @@ -505,6 +505,28 @@ describe('events', function () {
await worker.close();
});

describe('when jobs removal is attempted on non-existed records', async () => {
it('should not publish removed events', async () => {
const numRemovals = 100;
const trimmedQueue = new Queue(queueName, {
connection,
});

const client = await trimmedQueue.client;

for (let i = 0; i < numRemovals; i++) {
await trimmedQueue.remove(i.toString());
}

const eventsLength = await client.xlen(trimmedQueue.keys.events);

expect(eventsLength).to.be.eql(0);

await trimmedQueue.close();
await removeAllQueueData(new IORedis(), queueName);
});
});

describe('when maxLen is 0', function () {
it('should trim events automatically', async () => {
const trimmedQueue = new Queue(queueName, {
Expand Down Expand Up @@ -679,7 +701,7 @@ describe('events', function () {
});
});

describe('when jobs are retried inmediately', function () {
describe('when jobs are retried immediately', function () {
it('should trim events so its length is at least the threshold', async () => {
const numJobs = 80;
const trimmedQueue = new Queue(queueName, {
Expand Down

0 comments on commit 67de4fb

Please sign in to comment.