@happyvertical/smrt-jobs

Durable forge projections

ForgeDeliveryCollection and ForgeProjectionRuntime provide a provider-neutral inbox for durable forge webhook projection. The inbox identity is (tenantId, provider, deliveryId), so provider retries are atomically deduplicated. A worker lease owns each attempt; failures use bounded exponential retry, then remain in dead_letter until an operator calls replay() from the owning tenant context.

leaseMs must be a finite positive duration. retryBaseMs and retryMaxMs must be finite, non-negative durations no greater than 2,147,483,647 ms; invalid timing configuration is rejected before a lease can be claimed. A lease that would exceed the JavaScript Date range is rejected before any lease write; a retry that would exceed that range is durably dead-lettered instead of being left leased.

const inbox = await ForgeDeliveryCollection.create({ db });

await withTenant({ tenantId }, () =>
  inbox.accept({
    provider: 'github',
    deliveryId: request.headers.get('x-github-delivery')!,
    eventName: 'pull_request',
    repositoryKey: 'example/forge-repository',
    payload,
  }),
);

const runtime = new ForgeProjectionRuntime({ db, workerId: workerKey });
await runtime.processNext({
  async observe(delivery) {
    return {
      projection: 'pull-request-revision',
      subjectKey: `${delivery.repositoryKey}:pr:${delivery.payload.number}`,
      version: Number(delivery.payload.revision),
      value: delivery.payload,
    };
  },
  async project(observation, context) {
    // Always use context.db: it is the transaction shared by the application
    // projection, monotonic checkpoint, and inbox completion.
    await writeProjection(context.db, observation);
  },
});

projection and subjectKey are application-defined. A pull request is not treated as a tracker issue. version must be a non-negative monotonic signed 32-bit integer (0 through 2,147,483,647); observations at or below the durable checkpoint are acknowledged without reapplying side effects. Provider normalization runs with the delivery's tenant context restored. Generated REST/MCP/CLI surfaces are disabled for both system tables; operator replay requires an explicit in-process tenant context.

The manifest-driven migration adds _smrt_forge_deliveries and _smrt_forge_projection_checkpoints. Run smrt db:migrate before starting a forge projection worker.

Background job execution for s-m-r-t objects. Provides persistent queue storage, retry strategies, cron-based scheduling, and a fluent JobBuilder API via the withBackgroundJobs() mixin.

Installation

pnpm add @happyvertical/smrt-jobs

Usage

Add background capabilities to a SmrtObject

import { withBackgroundJobs, TaskRunner } from '@happyvertical/smrt-jobs';
import { Document } from './document.js';

// Mixin adds .bg() and .background() to any SmrtObject class
const BackgroundDocument = withBackgroundJobs(Document);
const doc = new BackgroundDocument({ db });
await doc.initialize();

// Quick enqueue — runs immediately when a TaskRunner picks it up
const handle = await doc.bg('generateSummary', { format: 'md' });

// Fluent builder for advanced options
const handle2 = await doc.background('generateSummary', { format: 'md' })
  .delay('5m')
  .priority('high')
  .retries(5)
  .queue('analysis')
  .timeout(600000)
  .enqueue();

// Wait for result (polling-based)
const result = await handle2.wait({ timeout: 60000, pollInterval: 100 });

Run a TaskRunner to process jobs

import { TaskRunner } from '@happyvertical/smrt-jobs';

const runner = new TaskRunner({
  concurrency: 5,
  pollInterval: 1000,
  // Empty checks back off exponentially to this cap (20× by default).
  idlePollInterval: 20000,
  queues: ['default', 'analysis'],
});
await runner.initialize(db);
await runner.start();

// Listen for events
runner.on('job:completed', (job, result) => { /* ... */ });
runner.on('job:failed', (job, error) => { /* ... */ });

// Graceful shutdown
process.on('SIGTERM', () => runner.stop());

Polling delays are capped at the effective worker lease TTL, including a larger pollInterval, to bound orphan-recovery sweep spacing. Both polling intervals must be finite positive milliseconds within the Node timer range. The default 20× idle cap reduces combined claim and recovery traffic; latency-sensitive workers can set a smaller cap.

Back MCP task operations with durable jobs

McpTaskStore persists the MCP io.modelcontextprotocol/tasks lifecycle on the same _smrt_jobs row that executes the operation. createTask() creates a correlated job, getTask() maps its queue state, and cancelTask() cancels that exact job without leaving a second record behind. Long-running task actions can request client input through JobExecutionContext.task:

async generate(options: Record<string, never>, context: JobExecutionContext) {
  const { tone } = await context.task!.requestInput({ tone: { type: 'string' } });
  return this.render(tone);
}

Run a TaskRunner for the mcp-tasks queue in application deployments. Task cancellation is cooperative: the job row becomes cancelled immediately and a running handler must observe its context before doing further side effects.

Liveness-safe job execution

TaskRunner records heartbeat telemetry, but recovery keys on a worker incarnation's live lease rather than a per-job heartbeat threshold. A blocked event loop must not make a still-running handler appear dead and cause a concurrent duplicate execution. See the live-set and off-loop lease-renewal details in Worker liveness & recovery.

Job handlers remain at-least-once. Avoid synchronous, CPU-bound, or otherwise long-running work when possible; make external effects idempotent because a process crash after an effect but before its terminal write still permits a later retry.

Schedule recurring jobs with ScheduleRunner

The ScheduleRunner polls the _smrt_agent_schedules table for due cron entries and creates SmrtJob records for the TaskRunner to execute. Wire them together via events:

import { ScheduleRunner } from '@happyvertical/smrt-jobs';

const scheduleRunner = new ScheduleRunner({ pollInterval: 30000 });
await scheduleRunner.initialize(db);
await scheduleRunner.start();

// Connect TaskRunner events to update schedule state
taskRunner.on('job:completed', (job) => {
  const scheduleId = job.args?._scheduleId;
  if (scheduleId) scheduleRunner.handleJobCompletion(scheduleId, true);
});
taskRunner.on('job:failed', (job, error) => {
  const scheduleId = job.args?._scheduleId;
  if (scheduleId) scheduleRunner.handleJobCompletion(scheduleId, false, error.message);
});

System Tables

| Table | Purpose | |-------|---------| | _smrt_jobs | Persistent job queue (SmrtJob records) | | _smrt_agent_schedules | Cron schedule entries polled by ScheduleRunner |

API

Classes

| Export | Description | |--------|------------| | SmrtJob | Persistent job record stored in _smrt_jobs | | SmrtJobCollection | Collection with claimReady(), listReady(), listByStatus(), stats(), cleanup() | | JobBuilder | Fluent API: .delay(), .priority(), .retries(), .queue(), .timeout(), .enqueue() | | JobHandle | Track, wait, cancel, or retry an enqueued job | | JobContextLogger | Logger that auto-injects job context (jobId, attempt, queue) | | TaskRunner | Polling-based execution engine with concurrency control and liveness leases | | ScheduleRunner | Polls for due cron schedules and creates SmrtJob entries |

TaskRunner uses SmrtJobCollection.claimReady() so multiple workers can poll the same queue without duplicate-claiming a pending row.

Functions

| Export | Description | |--------|------------| | createTaskRunner(config?) | Factory for creating a configured TaskRunner | | createScheduleRunner(config?) | Factory for creating a configured ScheduleRunner | | withBackgroundJobs(Class) | Mixin that adds .bg() and .background() to any SmrtObject class | | parseDelay(delay) | Parse human-readable delay strings ('5m', '1h', '30s') to milliseconds | | priorityToNumber(priority) | Convert priority label ('critical'/'high'/'normal'/'low') to number | | createHmacDurableJobPayloadSigner({ keyId, key }) | Sign canonical JSON job payloads with a server-only key and verify them after persistence |

Key Types

Priority, JobStatus, JobResult, WaitOptions, BgOptions, BackgroundCapable, TaskRunnerConfig, TaskRunnerEvents, ScheduleRunnerConfig, ScheduleRunnerEvents, ScheduleInfo, JobContext, TimeoutBehavior, SmrtJobData, ListReadyOptions, DurableJobPayloadSigner, DurableJobPayloadIntegrity

Dependencies

• @happyvertical/smrt-core -- ORM and code generation
• @happyvertical/smrt-config -- configuration loading
• @happyvertical/smrt-types -- shared type definitions
• @happyvertical/jobs -- retry strategies
• @happyvertical/sql -- database interface
• @happyvertical/logger -- structured logging
• @happyvertical/utils -- ID generation utilities
• Peer (optional): @happyvertical/smrt-svelte, svelte