@inf-minds/scheduler

Generic job scheduler with lease management and relay-based dispatch for @inf-minds stack.

Features

• Lease Management: Claim jobs with time-based leases, automatic expiry recovery
• Job Polling: Query pending jobs with filtering by account, job type, priority
• Relay Dispatch: Route jobs to workers via centralized relay
• Platform Agnostic: Works in Node.js and Cloudflare Workers environments

Installation

pnpm add @inf-minds/scheduler

Components

LeaseManager

Manages job leases for exclusive execution:

import { createLeaseManager } from '@inf-minds/scheduler';

const leaseManager = createLeaseManager(db, {
  leaseTimeoutMs: 90000, // 90 seconds
  accountId: 'optional-account-filter',
});

// Claim a job
const claimed = await leaseManager.claim('job-123', 'worker-456');

// Release a job
await leaseManager.release('job-123');

// Recover expired leases
const recovered = await leaseManager.recoverExpired();

JobPoller

Finds pending jobs ready for dispatch:

import { createJobPoller } from '@inf-minds/scheduler';

const jobPoller = createJobPoller(db, {
  maxJobsPerPoll: 10,
  jobTypes: ['mind-execution', 'rag-processing'], // optional
  accountId: 'account-123', // optional
});

const pendingJobs = await jobPoller.getPendingJobs();

Dispatcher

Dispatches jobs to workers via relay:

import { createDispatcher } from '@inf-minds/scheduler';

const dispatcher = createDispatcher({
  relayUrl: 'https://relay.example.com',
  authSecret: process.env.INTERNAL_AUTH_SECRET,
});

const success = await dispatcher.dispatch(job, worker);

Usage Example

Cloudflare Workers Durable Object

import { DurableObject } from 'cloudflare:workers';
import { createLeaseManager, createJobPoller, createDispatcher } from '@inf-minds/scheduler';
import { createWorkerDiscoveryClient } from '@inf-minds/router';

export class CoordinatorDO extends DurableObject {
  private accountId: string | null = null;

  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);

    if (url.pathname === '/start') {
      const { accountId } = await request.json();
      this.accountId = accountId;
      await this.ctx.storage.setAlarm(Date.now() + 2000);
      return new Response('Started');
    }

    return new Response('Not found', { status: 404 });
  }

  async alarm(): Promise<void> {
    if (!this.accountId) return;

    // Get database and dependencies
    const db = getDb(this.env);

    // Create components
    const leaseManager = createLeaseManager(db, {
      leaseTimeoutMs: 90000,
      accountId: this.accountId,
    });

    const jobPoller = createJobPoller(db, {
      maxJobsPerPoll: 10,
      accountId: this.accountId,
    });

    const dispatcher = createDispatcher({
      relayUrl: this.env.RELAY_URL,
      authSecret: this.env.INTERNAL_AUTH_SECRET,
    });

    // Worker discovery (subscribe to relay for worker updates)
    const workerDiscovery = createWorkerDiscoveryClient({
      relayUrl: this.env.RELAY_URL,
      authSecret: this.env.INTERNAL_AUTH_SECRET,
    });

    await workerDiscovery.connect();

    // Poll for pending jobs
    const pendingJobs = await jobPoller.getPendingJobs();

    for (const job of pendingJobs) {
      // Find available worker
      const worker = workerDiscovery.getAvailableWorker(job.type);
      if (!worker) continue;

      // Claim the job
      const claimed = await leaseManager.claim(job.id, worker.workerId);
      if (!claimed) continue;

      // Dispatch via relay
      const success = await dispatcher.dispatch(job, worker);

      if (!success) {
        // Release lease on dispatch failure
        await leaseManager.release(job.id);
      }
    }

    // Recover expired leases
    await leaseManager.recoverExpired();

    // Schedule next alarm
    await this.ctx.storage.setAlarm(Date.now() + 2000);
  }
}

Architecture

The scheduler package provides building blocks for job scheduling:

┌─────────────┐
│ JobPoller   │──→ Find pending jobs
└─────────────┘

┌─────────────┐
│LeaseManager │──→ Claim jobs with leases
└─────────────┘

┌─────────────┐
│ Dispatcher  │──→ Send jobs to workers via relay
└─────────────┘

The consuming application wires these together based on their platform:

• Cloudflare Workers: Use Durable Object alarms
• Node.js: Use setInterval or event loops

Dependencies

• @inf-minds/jobs - Job schema and types
• @inf-minds/router - Worker registry and discovery
• drizzle-orm - Database operations

License

Apache-2.0