Openship Job-Runner Asynchronous Task Processing: BullMQ and In-Process Architecture

Openship implements a unified JobRunner abstraction that dynamically selects between BullMQ (Redis-backed) or in-process (Postgres-polling) backends to execute background tasks like backups and maintenance jobs across both SaaS and self-hosted environments.

The oblien/openship repository uses a sophisticated job-runner asynchronous task processing system to handle background work—including backup runs, retention pruning, and permission clean-ups—without coupling execution logic to infrastructure constraints. This architecture inspects the runtime environment at startup and instantiates the appropriate backend, ensuring reliable task processing whether deployed as a distributed SaaS service or a single-process desktop application.

JobRunner Abstraction and Backend Selection

The platform defines a strict contract in [apps/api/src/lib/job-runner/types.ts](https://github.com/oblien/openship/blob/main/apps/api/src/lib/job-runner/types.ts) that decouples what work needs to be done from how it is executed. At startup, [apps/api/src/lib/job-runner/index.ts](https://github.com/oblien/openship/blob/main/apps/api/src/lib/job-runner/index.ts) creates a singleton instance by checking for the presence of REDIS_URL in the environment:

• BullMQ backend: Activated when REDIS_URL is configured. This provides distributed, persistent job processing suitable for SaaS deployments.
• In-process backend: Falls back to Postgres polling when Redis is unavailable, enabling zero-dependency operation for self-hosted desktop mode.

Both implementations expose identical methods, allowing consumers to enqueue jobs without knowing which concrete runner is active. The runner initializes when the API service boots:

// apps/api/src/app.ts
import { getJobRunner } from "./lib/job-runner";

await getJobRunner().start({
  processRun: async (runId) => { /* invoke backup runner */ },
});

BullMQ Backend for Distributed Processing

The Redis-backed implementation lives in [apps/api/src/lib/job-runner/bullmq.ts](https://github.com/oblien/openship/blob/main/apps/api/src/lib/job-runner/bullmq.ts). It creates a BullMQ Queue and Worker pair, wiring the processRun callback supplied during startup.

Key characteristics:

• Jobs persist in Redis and survive process restarts
• Multiple worker instances can consume from the same queue across different hosts
• Built-in retry logic with exponential back-off
• Rich observability via BullMQ's event system

Errors are logged with the prefix [job-runner:bullmq] for easy filtering in centralized logging systems.

In-Process Backend for Self-Hosted Mode

For environments without Redis, [apps/api/src/lib/job-runner/in-process.ts](https://github.com/oblien/openship/blob/main/apps/api/src/lib/job-runner/in-process.ts) implements a polling-based runner that queries Postgres directly.

Implementation details:

• Polls the database for rows with status='queued' (e.g., backup_run table entries)
• Manages recurring schedules using an in-memory cron parser (cron-parser package)
• Reloads persisted queue rows on restart, ensuring no work is lost during crashes
• Logs diagnostic messages with the prefix [job-runner:in-process]

This backend requires no external dependencies beyond the Postgres database already used by the application, making it ideal for single-node desktop deployments.

Core JobRunner API Methods

All backends implement the following contract defined in the types file:

Method	Signature	Purpose
enqueueRun	enqueueRun(runId: string)	Queues a one-off job. BullMQ stores it in Redis; the in-process runner writes a Postgres row for the poller to pick up.
scheduleRecurring	scheduleRecurring({ jobId, cronExpression, onTick })	Registers or replaces a cron schedule. BullMQ uses its internal scheduler; in-process keeps an in-memory timer.
removeRecurring	removeRecurring(jobId: string)	Unschedules a recurring job by ID.
shutdown	shutdown(deadlineMs?: number)	Graceful stop that waits for in-flight jobs to complete. Used by SIGTERM handlers.
describe	describe(): string	Returns backend identifier (e.g., "bullmq" or "in-process") for logging.

Production Usage Patterns

The job runner powers several critical subsystems across the codebase. Consumers import the singleton using import { getJobRunner } from "../../lib/job-runner" (or relative variants) and interact with the uniform API.

Backup Orchestration

In [apps/api/src/modules/backups/backup.orchestrator.ts](https://github.com/oblien/openship/blob/main/apps/api/src/modules/backups/backup.orchestrator.ts), the system enqueues a run for each backup request:

import { getJobRunner } from "./lib/job-runner";

async function triggerBackup(runId: string) {
  await getJobRunner().enqueueRun(runId);
}

Scheduled Maintenance Tasks

Recurring cleanup jobs leverage scheduleRecurring to register cron-driven maintenance:

• Audit log pruning: [apps/api/src/modules/audit/audit-prune-schedule.ts](https://github.com/oblien/openship/blob/main/apps/api/src/modules/audit/audit-prune-schedule.ts) schedules daily retention cleanup.
• Pending grant cleanup: [apps/api/src/modules/permissions/pending-grant-prune-schedule.ts](https://github.com/oblien/openship/blob/main/apps/api/src/modules/permissions/pending-grant-prune-schedule.ts) clears stale permission grants.
• Webhook event pruning: [apps/api/src/modules/github/webhook-event-prune-schedule.ts](https://github.com/oblien/openship/blob/main/apps/api/src/modules/github/webhook-event-prune-schedule.ts) periodically deletes old webhook events.

Each schedules its logic during module initialization, trusting the runner abstraction to handle the underlying execution model.

Implementation Examples

Enqueueing a One-Off Backup Run

import { getJobRunner } from "./lib/job-runner";

async function triggerBackup(runId: string) {
  await getJobRunner().enqueueRun(runId);
}

Scheduling a Recurring Maintenance Job

import { getJobRunner } from "./lib/job-runner";

await getJobRunner().scheduleRecurring({
  jobId: "audit-prune",
  cronExpression: "0 3 * * *", // every day at 03:00 UTC
  onTick: async () => {
    // Dynamic import keeps the initial module tree light
    const { pruneAuditLogs } = await import("./modules/audit/audit-prune");
    await pruneAuditLogs();
  },
});

Graceful Shutdown Handling

process.on("SIGTERM", async () => {
  const runner = getJobRunner();
  await runner.shutdown(10_000); // wait up to 10 seconds for in-flight jobs
  process.exit(0);
});

Summary

• Unified Abstraction: The JobRunner interface in apps/api/src/lib/job-runner/types.ts decouples business logic from execution details, allowing the same code to run in SaaS or desktop environments.
• Dual Backends: BullMQ provides enterprise-grade distributed processing via Redis, while the in-process backend enables zero-dependency operation using Postgres polling.
• Automatic Selection: apps/api/src/lib/job-runner/index.ts selects the appropriate backend at runtime based on the REDIS_URL environment variable.
• Reliable Execution: Both backends support graceful shutdowns, persistent job storage, and recurring cron schedules, ensuring background work completes reliably across deployments.

Frequently Asked Questions

How does Openship handle asynchronous task processing without Redis?

When REDIS_URL is not configured, Openship instantiates the in-process backend from apps/api/src/lib/job-runner/in-process.ts. This implementation polls the Postgres database for rows with status='queued' and executes them within the same process, eliminating the need for external message brokers while maintaining job persistence through the database.

What happens to queued jobs when the Openship process restarts?

Both backends ensure job durability across restarts. The BullMQ backend persists jobs in Redis, which survives process termination. The in-process backend reloads persisted rows from Postgres (such as backup_run records) when the poller restarts, re-queueing any work that was interrupted.

How do I schedule a recurring maintenance task in Openship?

Use the scheduleRecurring method on the JobRunner singleton, passing a unique jobId, a valid cronExpression, and an onTick callback. This works identically across both backends—BullMQ registers a recurring BullMQ job, while the in-process backend manages an in-memory cron timer. Example implementations can be found in apps/api/src/modules/audit/audit-prune-schedule.ts.

What is the graceful shutdown behavior for the job runner?

The shutdown(deadlineMs?) method allows the runner to wait for in-flight jobs to complete before terminating. During a SIGTERM event, the API service calls this method with a timeout (typically 10 seconds), giving active backup runs or maintenance tasks time to finish cleanly before the process exits. This prevents data corruption or orphaned job states.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}