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

> Explore Openship job-runner asynchronous task processing with dynamic BullMQ or in-process backends for robust background job execution in SaaS and self-hosted setups.

- Repository: [oblien/openship](https://github.com/oblien/openship)
- Tags: deep-dive
- Published: 2026-08-19

---

**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)](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)](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:

```typescript
// 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)](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)](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)](https://github.com/oblien/openship/blob/main/apps/api/src/modules/backups/backup.orchestrator.ts), the system enqueues a run for each backup request:

```typescript
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)](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)](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)](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

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

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

```

### Scheduling a Recurring Maintenance Job

```typescript
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

```typescript
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`](https://github.com/oblien/openship/blob/main/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`](https://github.com/oblien/openship/blob/main/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`](https://github.com/oblien/openship/blob/main/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`](https://github.com/oblien/openship/blob/main/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.