Merge pull request #125 from quirrel-dev/stale-checker-test

Bugfix: Re-Enqueue repeated jobs if orphaned

Author: Skn0tt

src/activity/activity.ts

@@ -100,7 +100,7 @@ export class Activity<ScheduleType extends string> implements Closable {
     options: SubscriptionOptions = {}
   ) {
     this.redis = redisFactory();
-    this.producer = new Producer<ScheduleType>(redisFactory);
+    this.producer = new Producer<ScheduleType>(redisFactory, null as any);
 
     this.redis.on("pmessage", (_pattern, channel, message) =>
       this.handleMessage(channel, message)

src/index.ts

@@ -54,6 +54,7 @@ export default class Owl<ScheduleType extends string> {
   public createProducer() {
     return new Producer<ScheduleType>(
       this.redisFactory,
+      this.scheduleMap,
       this.onError,
       this.staleCheckerConfig,
       this.logger

src/producer/producer.ts

@@ -10,6 +10,7 @@ import {
 } from "../encodeRedisKey";
 import { defineLocalCommands } from "../redis-commands";
 import type { Logger } from "pino";
+import { ScheduleMap } from "..";
 
 declare module "ioredis" {
   interface Commands {
@@ -46,6 +47,7 @@ export class Producer<ScheduleType extends string> implements Closable {
 
   constructor(
     redisFactory: () => Redis,
+    scheduleMap: ScheduleMap<ScheduleType>,
     onError?: OnError<ScheduleType>,
     staleCheckerConfig?: StaleCheckerConfig,
     private readonly logger?: Logger
@@ -64,6 +66,7 @@ export class Producer<ScheduleType extends string> implements Closable {
       this.redis,
       this.acknowledger,
       this,
+      scheduleMap,
       staleCheckerConfig,
       this.logger
     );

src/shared/stale-checker.ts

@@ -1,12 +1,27 @@
 import type { Redis } from "ioredis";
 import type { Logger } from "pino";
+import { ScheduleMap } from "..";
 import { Closable } from "../Closable";
 import { tenantToRedisPrefix } from "../encodeRedisKey";
 import type { Producer } from "../producer/producer";
 import { computeTimestampForNextRetry } from "../worker/retry";
+import { getNextExecutionDate } from "../worker/worker";
 import type { Acknowledger } from "./acknowledger";
 import { scanTenantsForProcessing } from "./scan-tenants";
 
+function valueAndScoreToObj(arr: (string | number)[]) {
+  const result: { value: string; score: number }[] = [];
+
+  for (let i = 0; i < arr.length; i += 2) {
+    result.push({
+      value: String(arr[i]),
+      score: Number(arr[i + 1]),
+    });
+  }
+
+  return result;
+}
+
 const oneMinute = 60 * 1000;
 
 export interface StaleCheckerConfig {
@@ -23,6 +38,7 @@ export class StaleChecker<ScheduleType extends string> implements Closable {
     private readonly redis: Redis,
     private readonly acknowledger: Acknowledger<ScheduleType>,
     private readonly producer: Producer<ScheduleType>,
+    private readonly scheduleMap: ScheduleMap<ScheduleType>,
     config: StaleCheckerConfig = {},
     private readonly logger?: Logger
   ) {
@@ -53,7 +69,7 @@ export class StaleChecker<ScheduleType extends string> implements Closable {
   ) {
     const result = await this.redis
       .pipeline()
-      .zrangebyscore(key, min, max)
+      .zrangebyscore(key, min, max, "WITHSCORES")
       .zremrangebyscore(key, min, max)
       .exec();
 
@@ -67,7 +83,7 @@ export class StaleChecker<ScheduleType extends string> implements Closable {
       throw remRangeByScoreResult[0];
     }
 
-    return rangeByScoreResult[1] as string[];
+    return valueAndScoreToObj(rangeByScoreResult[1]);
   }
 
   private parseJobDescriptor(descriptor: string) {
@@ -103,14 +119,18 @@ export class StaleChecker<ScheduleType extends string> implements Closable {
 
           const staleJobs = await this.producer.findJobs(
             tenant,
-            staleJobDescriptors.map(this.parseJobDescriptor)
+            staleJobDescriptors.map(({ value }) =>
+              this.parseJobDescriptor(value)
+            )
           );
 
           const pipeline = this.redis.pipeline();
 
           const error = "Job Timed Out";
 
-          for (const job of staleJobs) {
+          for (let i = 0; i < staleJobs.length; i++) {
+            const job = staleJobs[0];
+            const score = staleJobDescriptors[0].score;
             if (!job) {
               this.logger?.error(
                 { tenant },
@@ -125,6 +145,18 @@ export class StaleChecker<ScheduleType extends string> implements Closable {
               job.count
             );
 
+            // TODO: duplicated logic in producer, please extract
+            let nextExecutionDate: number | undefined = undefined;
+
+            if (!job.schedule?.times || job.count < job.schedule?.times) {
+              nextExecutionDate = getNextExecutionDate(
+                this.scheduleMap,
+                job.schedule?.type,
+                job.schedule?.meta ?? "",
+                new Date(score)
+              );
+            }
+
             this.logger?.trace(
               { tenant, job },
               "Stale-Checker: Adding Failure report to pipeline"
@@ -136,6 +168,7 @@ export class StaleChecker<ScheduleType extends string> implements Closable {
                 queueId: job.queue,
                 jobId: job.id,
                 timestampForNextRetry,
+                nextExecutionDate,
               },
               job,
               error,

src/worker/worker.ts

@@ -55,6 +55,29 @@ function parseTenantFromChannel(topic: string) {
   return "";
 }
 
+export function getNextExecutionDate<ScheduleType extends string>(
+  scheduleMap: ScheduleMap<ScheduleType>,
+  schedule_type: ScheduleType | undefined,
+  schedule_meta: string,
+  lastExecution: Date
+): number | undefined {
+  if (!schedule_type) {
+    return undefined;
+  }
+
+  const scheduleFunc = scheduleMap[schedule_type];
+  if (!scheduleFunc) {
+    throw new Error(`Schedule ${schedule_type} not found.`);
+  }
+
+  const result = scheduleFunc(lastExecution, schedule_meta);
+  if (!result) {
+    return undefined;
+  }
+
+  return +result;
+}
+
 export class Worker<ScheduleType extends string> implements Closable {
   private readonly redis;
   private readonly redisSub;
@@ -112,21 +135,12 @@ export class Worker<ScheduleType extends string> implements Closable {
     schedule_meta: string,
     lastExecution: Date
   ): number | undefined {
-    if (!schedule_type) {
-      return undefined;
-    }
-
-    const scheduleFunc = this.scheduleMap[schedule_type];
-    if (!scheduleFunc) {
-      throw new Error(`Schedule ${schedule_type} not found.`);
-    }
-
-    const result = scheduleFunc(lastExecution, schedule_meta);
-    if (!result) {
-      return undefined;
-    }
-
-    return +result;
+    return getNextExecutionDate(
+      this.scheduleMap,
+      schedule_type,
+      schedule_meta,
+      lastExecution
+    );
   }
 
   private readonly distributor = new JobDistributor(

test/functional/stale-check.test.ts

@@ -47,6 +47,7 @@ describeAcrossBackends("stale-check", (backend) => {
           jobId: "stalling-job",
           queueId: "stally-stall",
           timestampForNextRetry: undefined,
+          nextExecutionDate: undefined,
         },
         "Job Timed Out",
       ],
@@ -105,4 +106,31 @@ describeAcrossBackends("stale-check", (backend) => {
     await env.producer.staleChecker.check();
     expect(env.errors).to.deep.equal([]);
   });
+
+  describe("when a schedule jobs goes stale", () => {
+    it("is re-scheduled", async () => {
+      worker = await env.owl.createWorker(async (job, ack) => {
+        // let job be stale
+      });
+
+      await env.producer.enqueue({
+        tenant: "",
+        id: "@cron",
+        payload: "null",
+        queue: "cron-job",
+        schedule: {
+          type: "every",
+          meta: "1000",
+        },
+      });
+
+      await delay(200);
+
+      await env.producer.staleChecker.check();
+
+      const job = await env.producer.findById("", "cron-job", "@cron");
+      expect(job).not.to.be.null;
+      expect(+job.runAt).to.be.greaterThan(Date.now());
+    });
+  });
 });

test/util.ts

@@ -7,13 +7,17 @@ export function describeAcrossBackends(
   runTests: (backend: Backend) => void
 ) {
   describe(topic, () => {
-    describe("Redis", () => {
-      runTests("Redis");
-    });
+    if (process.env.TEST_BACKEND !== "In-Memory") {
+      describe("Redis", () => {
+        runTests("Redis");
+      });
+    }
 
-    describe("In-Memory", () => {
-      runTests("In-Memory");
-    });
+    if (process.env.TEST_BACKEND !== "Redis") {
+      describe("In-Memory", () => {
+        runTests("In-Memory");
+      });
+    }
   });
 }