🐛 Bug Report: Pulse Worker Cleanup Issue in Fastify 5.x Integration

[channainfo]

Description

I am integrating Pulse into my Fastify 5.x application and writing a spec to test the integration. The test itself runs as expected, but I am encountering the following warning message when Jest completes execution:

Jest did not exit one second after the test run has completed.
This usually means that there are asynchronous operations that weren't stopped in your tests. Consider running Jest with `--detectOpenHandles` to troubleshoot this issue.

It appears that worker.stop() does not properly clean up the worker after the test completes, leading to Jest detecting an open handle.

---

Steps to Reproduce

1. Use Pulse with the following configuration:

   const workerConfig: PulseConfig = {
     db: { address: mongoURI, collection: "jobs_test" },
     processEvery: "2 seconds",
     maxConcurrency: 4,
     disableAutoIndex: false,
     defaultConcurrency: 5,
   };
   

2. Define and start the worker in beforeAll:

   worker = new Pulse(workerConfig);
   await worker.start();
   

3. Stop the worker and in-memory MongoDB in afterAll:

   await worker.stop();
   await mongoServer.stop();
   

4. Run the test suite with Jest.

---

Expected Behavior

• Jest should exit cleanly after tests complete.
• worker.stop() should correctly clean up all resources.

---

Actual Behavior

• Jest does not exit after tests complete.
• The warning suggests open asynchronous operations are still running, possibly from Pulse's internal processes.

---

Possible Solutions Explored

• Running Jest with --detectOpenHandles confirms the issue is related to Pulse.
• Ensured worker.stop() is called in afterAll, but it does not resolve the issue.

---

Environment

Dependency	Version
Fastify	5.x
Pulse	@pulsecron/pulse latest
Jest	x.x.x
Node.js	x.x.x
MongoMemoryServer	latest

---

Additional Context

Would appreciate any guidance on whether Pulse requires additional cleanup beyond worker.stop() or if there is an open issue regarding resource cleanup.

Thank you! 🙏

Here is my full spec

//tests/pulse.spec.ts
/* eslint-disable @typescript-eslint/no-explicit-any */
import Pulse, { Job, PulseConfig } from "@pulsecron/pulse";
import { MongoMemoryServer } from "mongodb-memory-server";

describe("worker", () => {
  let worker: Pulse;
  let mongoServer: MongoMemoryServer;

  beforeAll(async () => {
    // Start an in-memory MongoDB server
    mongoServer = await MongoMemoryServer.create();
    const mongoURI = mongoServer.getUri();
    console.log("Mongo URI: ", mongoURI);

    const workerConfig: PulseConfig = {
      db: { address: mongoURI, collection: "jobs_test" },
      processEvery: "2 seconds",
      maxConcurrency: 4,
      disableAutoIndex: false,
      defaultConcurrency: 5, // Max 5 jobs running at the same time
    };

    worker = new Pulse(workerConfig);
    await worker.start();
  });

  afterAll(async () => {
    console.log("Stopping worker");
    await worker.stop();
    console.log("Stopping mongoServer");
    await mongoServer.stop();
  });

  it("should have a default concurrency of 5", async () => {
    worker.define("with_arg_job", async (job: Job) => {
      console.info("Executing job: with_arg_job", job.attrs.data);
    });

    worker.define("without_arg_job", async () => {
      console.info("Executing job: without_arg_job");
    });

    await worker.create("with_arg_job", { email: "test@pulse.org" }).save();
    await worker.create("without_arg_job", {}).save();
  });
});

Test output

[channainfo]

If I use the mongo option, everything works as expected. Therefore, for now, I am adopting this approach. Even if a mongoURI is provided, I will manually convert it into a mongo option to ensure consistency, as demonstrated in the code below.

// config/worker.ts

import Pulse, { PulseConfig } from "@pulsecron/pulse";
import { Env } from "libs/env";
import { Db, MongoClient } from "mongodb";

export const collectionName = `jobs_${Env.envName()}`;

export interface IWorkerConfig {
  mongoURI?: string;
  mongoClient?: MongoClient;
}

export function createWorker(workerConfig?: IWorkerConfig): { worker: Pulse; mongoClient: MongoClient } {
  let { mongoURI, mongoClient } = workerConfig || {};
  const queueCount = parseInt(Env.fetch("QUEUE_CONCURRENT_COUNT", "4"));

  let config: PulseConfig = {
    maxConcurrency: queueCount,
    disableAutoIndex: false,
  };

  if (mongoClient !== undefined) {
    const mongo = mongoClient.db(collectionName);

    config = {
      ...config,
      mongo,
    };
  } else {
    mongoURI = mongoURI || Env.fetch("MONGO_URI");
    mongoClient = new MongoClient(mongoURI);
    mongoClient.connect();
    const mongo = mongoClient.db(collectionName) as Db;
    config = {
      ...config,
      mongo,
      // db: { address: mongoURI, collection: collectionName },
    };
  }

  const worker = new Pulse(config);
  return { worker, mongoClient };
}

and here is the jest spec

// tests/config/worker.spec.ts
import Pulse, { Job } from "@pulsecron/pulse";
import { createWorker } from "config/worker";
import AppLog from "libs/app_log";
import { MongoClient } from "mongodb";
import { MongoMemoryServer } from "mongodb-memory-server";

describe("worker", () => {
  describe("with mongoURI", () => {
    let worker: Pulse;
    let mongoClient: MongoClient;
    let mongoServer: MongoMemoryServer;

    beforeAll(async () => {
      mongoServer = await MongoMemoryServer.create();
      const mongoURI = mongoServer.getUri();

      ({ worker, mongoClient } = await createWorker({ mongoURI }));
      await worker.start();
    });

    afterEach(async () => {
      await worker.cancel({});
    });

    afterAll(async () => {
      await worker.stop();
      await mongoClient.close();
      await mongoServer.stop();
    });

    it("define and create job successfully", async () => {
      expect(worker).toBeDefined();
      expect(mongoClient).toBeDefined();

      worker.define("with_arg_job", async (job: Job) => {
        AppLog.info("Executing job: with_arg_job", job.attrs.data);
      });

      worker.define("without_arg_job", async () => {
        AppLog.info("Executing job: without_arg_job");
      });

      await worker.create("with_arg_job", { email: "test@pulse.org" }).save();
      await worker.create("without_arg_job", {}).save();

      const jobs = await worker.jobs();
      expect(jobs).toHaveLength(2);
    });
  });

  describe("with mongoClient", () => {
    let worker: Pulse;
    let mongoClient: MongoClient;
    let mongoServer: MongoMemoryServer;

    beforeAll(async () => {
      mongoServer = await MongoMemoryServer.create();
      const mongoURI = mongoServer.getUri();
      mongoClient = new MongoClient(mongoURI);
      await mongoClient.connect();

      ({ worker, mongoClient } = await createWorker({ mongoClient }));
      await worker.start();
    });

    afterEach(async () => {
      await worker.cancel({});
    });

    afterAll(async () => {
      await worker.stop();
      await mongoClient.close();
      await mongoServer.stop();
    });

    it("define and create job successfully", async () => {
      expect(worker).toBeDefined();
      expect(mongoClient).toBeDefined();

      worker.define("with_arg_job", async (job: Job) => {
        AppLog.info("Executing job: with_arg_job", job.attrs.data);
      });

      worker.define("without_arg_job", async () => {
        AppLog.info("Executing job: without_arg_job");
      });

      await worker.create("with_arg_job", { email: "test@pulse.org" }).save();
      await worker.create("without_arg_job", {}).save();

      const jobs = await worker.jobs();
      expect(jobs).toHaveLength(2);
    });
  });
});

[code-xhyun]

Due to personal circumstances, I had difficulty managing the project. Starting this month, I plan to gradually resume overseeing it. I have also applied to renew the documentation site domain, and it is expected to be restored soon