Faktory Worker for Node.js

Common errors

• faktory worker failed to start: Error: connect ECONNREFUSED 127.0.0.1:7419

  cause The Faktory job server is not running or is not accessible at the default host and port (localhost:7419), or `FAKTORY_URL` environment variable is misconfigured.
  fix Ensure the Faktory server is running and accessible from the worker process. Verify `FAKTORY_URL` environment variable, or explicitly pass connection options to `faktory.connect()` or `faktory.work()`.

• Error: Job not found <jid>

  cause This error can occur intermittently, especially with large job backlogs or during worker deployments, indicating the Faktory server cannot locate a job that a worker attempted to process.
  fix Review Faktory server logs for related errors. Ensure consistent deployment practices to avoid sudden worker restarts on jobs. Consider Faktory server configuration and network stability. This might be a server-side issue or transient network problem.

• TypeError: faktory.connect is not a function (when using require)

  cause Attempting to use ES module named imports syntax with CommonJS `require()` or `faktory-worker` is exporting functions directly and not as properties of a default object when using CommonJS.
  fix For CommonJS, use `const faktory = require('faktory-worker');` and then access `faktory.connect()`, `faktory.register()`, `faktory.work()`. For ES modules, use `import { connect, register, work } from 'faktory-worker';`.

Warnings

• gotcha It is crucial to properly manage Faktory client connections. Clients should be reused across multiple job pushes rather than creating a new client for each job. Remember to call `client.close()` when the client is no longer needed to release resources, especially in short-lived scripts or server shutdowns.
  Fix: Initialize a single client instance for an application lifetime (e.g., on application startup) and reuse it. Always ensure `await client.close()` is called in `finally` blocks or during application shutdown for graceful disconnection.
• gotcha Job functions registered with `faktory.register()` must correctly handle asynchronous operations. If an `async` job function returns before all `await` calls are resolved, the job will be `ACK`ed prematurely by the Faktory server, potentially leading to incomplete work.
  Fix: Always `await` all asynchronous operations within your job functions. Ensure the function only returns once all work is truly complete or an error has been properly thrown.
• breaking The library requires Node.js version 16 or higher. Older Node.js versions are not supported and will result in runtime errors or unexpected behavior.
  Fix: Upgrade your Node.js environment to version 16 or newer. Use a Node.js version manager like `nvm` to easily switch and manage Node.js versions.
• breaking This `faktory-worker` library is compatible with Faktory server versions `>v1.6.1`. Using it with older Faktory server versions may lead to protocol incompatibilities, unexpected job processing failures, or connection issues.
  Fix: Ensure your Faktory server instance is updated to version 1.6.1 or newer. Refer to the Faktory server's official documentation for upgrade instructions.
• gotcha If a Faktory worker process crashes while processing a job, the job will sit in the 'busy' state until its reservation timeout expires on the Faktory server. Only after the timeout will Faktory consider it failed and potentially re-enqueue it for retry, depending on job configuration. This can cause delays.
  Fix: Design job functions to be idempotent where possible. Monitor worker health and Faktory busy queues. Configure appropriate job reservation timeouts on the Faktory server to balance responsiveness and retry safety. Implement robust error handling and logging within job functions to identify issues before a full worker crash.

Install

• npm install faktory-worker npm
• yarn add faktory-worker yarn
• pnpm add faktory-worker pnpm

Imports

• connect
  wrong:

  const faktory = require('faktory-worker'); const client = await faktory.connect();

  correct:

  import { connect } from 'faktory-worker';

  For ESM, prefer named imports for individual functions. `connect` returns a `Client` instance for pushing jobs. In CommonJS, `require('faktory-worker')` returns an object from which `connect` is accessed.
• register
  wrong:

  const faktory = require('faktory-worker'); faktory.register('JobType', ...);

  correct:

  import { register } from 'faktory-worker';

  `register` is used to associate a job type string with a JavaScript function that processes the job payload. This is typically done in the worker process.
• work
  wrong:

  const faktory = require('faktory-worker'); faktory.work();

  correct:

  import { work } from 'faktory-worker';

  `work` starts the Faktory worker process, which fetches and executes registered jobs from the Faktory server. It handles graceful shutdown on `SIGINT`/`SIGTERM`.
• Client

  import { Client } from 'faktory-worker';

  While `connect()` is the primary way to get a client instance, the `Client` class is exported for type-checking or advanced usage. A `Client` represents a connection handle and is safe for concurrent use.

Quickstart

This quickstart demonstrates both the worker setup, registering a job ('ResizeImage'), and how a client pushes that job. It includes a basic `process.argv` check to allow running either the worker or client part independently, mirroring real-world deployment where these are separate processes.

import { connect, register, work, ClientOptions } from 'faktory-worker';

// 1. Define your job processing logic
interface ResizeImagePayload {
  id: number;
  size: string;
}

register('ResizeImage', async (payload: ResizeImagePayload) => {
  console.log(`[Worker] Processing job ResizeImage for image ${payload.id} with size ${payload.size}`);
  // Simulate an asynchronous operation, e.g., image resizing or database update
  await new Promise(resolve => setTimeout(resolve, Math.random() * 1000 + 500));
  console.log(`[Worker] Finished ResizeImage for image ${payload.id}`);
});

// 2. Worker startup (typically run in a long-running background process)
async function startFaktoryWorker() {
  console.log('Starting Faktory worker...');
  try {
    const workerOptions: ClientOptions = {
      queues: ['default', 'images'], // Listen to specified queues
      concurrency: 5,                  // Process up to 5 jobs concurrently
      // Faktory server URL can be set via FAKTORY_URL environment variable
      // or explicitly passed: { host: 'localhost', port: 7419 }
      // factory: { host: 'localhost', port: 7419 },
    };
    await work(workerOptions);
    console.log('Faktory worker started and waiting for jobs...');
  } catch (error) {
    console.error(`Faktory worker failed to start: ${error}`);
    // In Node.js, ensure the process exits on critical errors
    if (typeof process !== 'undefined' && process.exit) {
      process.exit(1);
    }
  }
}

// 3. Client job pushing (typically run from an application server or another process)
async function pushFaktoryJob(payload: ResizeImagePayload) {
  let client;
  try {
    console.log(`[Client] Connecting to Faktory server to push job for image ${payload.id}...`);
    client = await connect(); // Connects to Faktory server

    console.log(`[Client] Pushing job 'ResizeImage' with payload:`, payload);
    await client.job('ResizeImage', payload).push();
    console.log(`[Client] Job 'ResizeImage' for image ${payload.id} pushed successfully.`);

    // Example of pushing a bulk of jobs
    // const job1 = client.job('ResizeImage', { id: 102, size: 'medium' });
    // const job2 = client.job('ResizeImage', { id: 103, size: 'small' });
    // const rejected = await client.pushBulk([job1, job2]);
    // if (Object.keys(rejected).length > 0) {
    //   console.error('[Client] Some bulk jobs were rejected:', rejected);
    // }

  } catch (error) {
    console.error(`[Client] Failed to push job: ${error}`);
  } finally {
    if (client) {
      await client.close(); // Important: reuse client or close after use
      console.log('[Client] Disconnected from Faktory server.');
    }
  }
}

// To run this quickstart:
// 1. Ensure a Faktory server is running (e.g., docker run --rm -p 7419:7419 -p 7420:7420 contribsys/faktory)
// 2. Save this file (e.g., `app.ts`).
// 3. Run the worker in one terminal: `ts-node app.ts worker`
// 4. Run the client to push a job in another terminal: `ts-node app.ts client`

const mode = process.argv[2];

if (mode === 'worker') {
  startFaktoryWorker();
} else if (mode === 'client') {
  pushFaktoryJob({ id: 101, size: 'large' });
} else {
  console.log('Usage: ts-node app.ts [worker|client]');
  console.log('Example: ts-node app.ts worker (to start the job processor)');
  console.log('Example: ts-node app.ts client (to push a job)');
  // For a truly single-file runnable demo, you'd push then immediately start a worker
  // but in practice, these are separate long-running processes.
}