npm package: @pico-brief/task-manager
Description: A TypeScript library that makes sure a piece of expensive work only runs once at a time, even when many servers or processes try to start it simultaneously. When the work finishes it caches the outcome, so any later caller gets the result instantly withou
Latest version: 0.1.0

# task-manager

A TypeScript library that makes sure a piece of expensive work only runs **once at a time**, even when many servers or processes try to start it simultaneously. When the work finishes it caches the outcome, so any later caller gets the result instantly without re-running the job.

---

## The problem this solves

Imagine your application needs to generate a large PDF report. Ten users click "Generate Report" at the same moment. Without coordination, all ten servers start generating the same PDF simultaneously — wasting resources and potentially producing ten identical files. With `task-manager`, only the first request actually does the work. The other nine wait and, when the first one finishes, they all receive the same result.

This pattern is called **task deduplication**. It is useful any time a job is:

- **Expensive** — slow, CPU-heavy, or costly to call (e.g. an AI API)
- **Idempotent** — running it once or ten times should produce the same result
- **Shared** — many callers might trigger it at the same time

---

## How it works (plain English)

When you call `manager.run(taskId, work)`:

1. **One caller wins the lock.** The first caller to arrive claims an exclusive lock on `taskId` and starts running `work`. Think of this like grabbing a "Currently working on it" sign and putting it on the door.

2. **Other callers become waiters.** Any other process that calls `run()` with the same `taskId` while the first is still working sees the lock is taken and starts polling every 500 ms to check whether the job has finished.

3. **The owner keeps the lock alive with a heartbeat.** Every 3 seconds the owner refreshes the lock's expiry. This is how waiters know the owner is still alive and making progress.

4. **If the owner dies, a waiter takes over.** If the lock stops being refreshed for 30 seconds (the `takeoverAfterMs` threshold), a waiting process assumes the owner crashed. It acquires the lock itself and re-runs the work from scratch.

5. **The result is cached.** Success or failure is written to the store so that any caller arriving *after* the job has already finished gets the result immediately — the work function is never called again for that `taskId`.

---

## Installation

```bash
npm install @pico-brief/task-manager
```

```bash
# or with yarn
yarn add @pico-brief/task-manager
```

**Requirements:** Node.js 18+, TypeScript 5+ (if using TypeScript).

---

## Quick start

```typescript
import { TaskManager, InMemoryTaskStore } from '@pico-brief/task-manager';

// Create a store (use InMemoryTaskStore for a single process,
// RedisTaskStore for multiple servers).
const store = new InMemoryTaskStore();
const manager = new TaskManager(store);

// Run a task. The second argument is the actual work to do.
const succeeded = await manager.run('my-first-task', async (reportProgress) => {
  console.log('Doing the work...');

  // Optionally report how far along you are (a number from 0 to 1).
  await reportProgress(0.5); // 50% done

  console.log('Work complete!');
});

if (succeeded) {
  console.log('Task finished successfully.');
} else {
  const error = await manager.getError('my-first-task');
  console.log('Task failed:', error);
}
```

`run()` always returns a `boolean`:
- `true` — the work completed without throwing
- `false` — the work threw an error (or exhausted all retries)

---

## Choosing a store

The store is where the library keeps track of which tasks are running, their progress, and their outcomes.

### `InMemoryTaskStore`

Keeps everything in the current process's memory. Fast and zero-config.

```typescript
import { InMemoryTaskStore } from 'task-manager';
const store = new InMemoryTaskStore();
```

**Use this when:** running a single Node.js process, writing tests, or prototyping.

**Do not use this when:** you have multiple servers, because each server has its own memory — they cannot coordinate with each other.

### `RedisTaskStore`

Keeps everything in a shared Redis instance that all your servers can see.

```typescript
import { RedisTaskStore } from 'task-manager';
import { createClient } from 'redis';

const redis = createClient({ socket: { host: 'localhost', port: 6379 } });
await redis.connect();

const store = new RedisTaskStore(redis);
```

**Use this when:** you have multiple servers that need to coordinate, which is the typical production setup.

---

## Common use cases

### 1. Preventing duplicate API calls

You have an endpoint that fetches and caches a slow third-party API response. Many users might hit it at the same time.

```typescript
import { TaskManager, RedisTaskStore } from 'task-manager';
import Redis from 'ioredis';
import { fetchWeatherData, saveToCache, getFromCache } from './weather';

const store = new RedisTaskStore(new Redis());
const manager = new TaskManager(store);

async function getWeatherReport(city: string): Promise<WeatherData> {
  const taskId = `weather:${city}`;

  // If the report is already cached, return it immediately.
  const cached = await getFromCache(taskId);
  if (cached) return cached;

  // Otherwise, run the fetch. If ten requests come in at once for the same
  // city, only one HTTP call is made — the others wait.
  const succeeded = await manager.run(taskId, async () => {
    const data = await fetchWeatherData(city);
    await saveToCache(taskId, data);
  });

  if (!succeeded) {
    throw new Error(`Failed to fetch weather for ${city}`);
  }

  return getFromCache(taskId)!;
}
```

### 2. Generating a file on demand

A user requests a PDF export. You want to avoid generating the same PDF multiple times if they click "Export" impatiently.

```typescript
async function exportUserReport(userId: string): Promise<string> {
  const taskId = `report:pdf:${userId}`;

  const succeeded = await manager.run(taskId, async (reportProgress) => {
    await reportProgress(0.1);

    const data = await db.fetchUserData(userId);
    await reportProgress(0.4);

    const pdf = await generatePdf(data);
    await reportProgress(0.8);

    await s3.upload(`reports/${userId}.pdf`, pdf);
    await reportProgress(1.0);
  });

  if (!succeeded) {
    const error = await manager.getError(taskId);
    throw new Error(`Report generation failed: ${error}`);
  }

  return `reports/${userId}.pdf`;
}
```

While the PDF is generating you can poll its progress from a separate request:

```typescript
async function getExportProgress(userId: string): Promise<number> {
  // Returns a number between 0 and 1, or null if not started.
  return manager.getProgress(`report:pdf:${userId}`) ?? 0;
}
```

### 3. Expensive AI or ML jobs

Generating an embedding, running inference, or calling a paid AI API that you never want to call twice for the same input.

```typescript
async function getEmbedding(text: string): Promise<number[]> {
  // Use a stable hash of the input as the task ID.
  const taskId = `embedding:${hashText(text)}`;

  const succeeded = await manager.run(taskId, async () => {
    const embedding = await openai.createEmbedding(text);
    await vectorDb.store(taskId, embedding);
  });

  if (!succeeded) throw new Error('Embedding failed');

  return vectorDb.fetch(taskId);
}
```

### 4. Scheduled or background jobs

A job runs on a cron schedule but can also be triggered manually. You want to make sure it never runs more than once at a time, even if two cron workers fire simultaneously.

```typescript
// This runs on every server in your cluster. Only one will actually
// execute the work; the others will wait and receive the same result.
async function runNightlySync() {
  const taskId = `nightly-sync:${todayDateString()}`;

  const succeeded = await manager.run(taskId, async (progress) => {
    await syncUsers();
    await progress(0.33);

    await syncOrders();
    await progress(0.66);

    await syncInventory();
    await progress(1.0);
  });

  return succeeded;
}
```

### 5. Re-running a failed task

By default, once a task is `failed`, all subsequent callers see the cached failure immediately (so they don't pile up retrying at the same time). To explicitly reset a task and let it run again:

```typescript
// Clear the cached state so the next run() starts fresh.
await manager.clearTask('report:pdf:user-123');

// Now re-run normally.
const succeeded = await manager.run('report:pdf:user-123', generateReport);
```

---

## Pipelines

A **pipeline** is a sequence of steps that run one after another. Each step is itself a deduplicated task. This is useful for multi-stage jobs where you want:

- Steps to run in a fixed order
- Each step to be individually deduplicated (two processes running the same pipeline coordinate step-by-step)
- Partial resume — if the pipeline crashes halfway through, it picks up from where it left off rather than starting over

### Basic pipeline

```typescript
import { TaskManager, InMemoryTaskStore, PipelineStep } from 'task-manager';

const manager = new TaskManager(new InMemoryTaskStore());

const steps: PipelineStep[] = [
  {
    taskId: 'etl:extract',
    run: async (ctx, reportProgress) => {
      console.log(`Running step ${ctx.stepIndex + 1} of ${ctx.totalSteps}`);
      await extractDataFromSource();
      await reportProgress(1.0);
    },
  },
  {
    taskId: 'etl:transform',
    run: async (ctx, reportProgress) => {
      await reportProgress(0.0);
      await transformData();
      await reportProgress(1.0);
    },
  },
  {
    taskId: 'etl:load',
    run: async (ctx, reportProgress) => {
      await loadIntoWarehouse();
      await reportProgress(1.0);
    },
  },
];

const pipeline = manager.pipeline('etl-job', steps);
const succeeded = await pipeline.run();

if (succeeded) {
  console.log('ETL complete!');
} else {
  const status = await pipeline.getStatus();
  console.log(`Failed at step "${status.failedTaskId}": ${status.error}`);
}
```

### Checking pipeline progress

`getStatus()` gives you a snapshot of where the pipeline is right now. You can call it from a separate process or a polling endpoint while the pipeline is running.

```typescript
const status = await pipeline.getStatus();

console.log(status.pipelineId);      // 'etl-job'
console.log(status.currentStep);     // 1  (zero-based index of active step)
console.log(status.totalSteps);      // 3
console.log(status.overallProgress); // 0.38  (a number from 0 to 1)
console.log(status.status);          // 'running' | 'completed' | 'failed'
```

`overallProgress` accounts for each step equally. If there are 4 steps and step 1 (the second step) is 50% done, `overallProgress` is `(1 completed + 0.5 in progress) / 4 = 0.375`.

### Partial resume after a crash

By default (`resumeFromLastCompleted: true`), if the pipeline is interrupted and you run it again, it skips any steps that already reached `completed` and continues from the first incomplete step.

```typescript
// First run — crashes halfway through step 2.
await pipeline.run(); // returns false

// Second run — step 1 is already completed in the store, so it is skipped.
// Only steps 2 and 3 run.
await pipeline.run(); // picks up from step 2
```

To disable this and always run every step from scratch:

```typescript
const pipeline = manager.pipeline('etl-job', steps, {
  resumeFromLastCompleted: false,
});
```

---

## Advanced usage

### Retrying on failure

By default, a task that throws is marked as failed with no retries. You can configure automatic retries with a backoff delay.

```typescript
const succeeded = await manager.run('flaky-api-call', async () => {
  await callUnreliableExternalApi();
}, {
  maxRetries: 4,                              // 1 initial attempt + 4 retries = 5 total tries
  backoff: {
    type: 'exponential',
    initialDelayMs: 500,                      // wait 500 ms before the first retry
    multiplier: 2,                            // double the wait each time
    maxDelayMs: 15_000,                       // never wait more than 15 seconds
    jitter: true,                             // add randomness to avoid thundering herd
  },
});
```

#### Fixed backoff

Always waits the same amount of time between retries. Simpler, but can cause all servers to retry at the same moment.

```typescript
backoff: { type: 'fixed', delayMs: 2000 }   // always wait 2 seconds
```

#### Exponential backoff

Waits longer and longer between retries. With `jitter: true` (the default) the wait is randomised within `[0, cappedDelay]`, which prevents a thundering herd — the situation where every server retries at exactly the same instant and overloads the downstream service.

```typescript
backoff: {
  type: 'exponential',
  initialDelayMs: 200,   // retry 1: up to  200 ms
  multiplier: 3,         // retry 2: up to  600 ms
  maxDelayMs: 10000,    // retry 3: up to 1800 ms, retry 4+: capped at 10 s
}
```

### Conditional retries

Use `shouldRetry` to inspect the error and decide whether a retry makes sense. This is useful when some errors are permanent (e.g. "resource not found") and others are transient (e.g. "connection timeout").

```typescript
const succeeded = await manager.run('delete-resource', async () => {
  await callApi('/resource/123', 'DELETE');
}, {
  maxRetries: 5,
  backoff: { type: 'exponential', initialDelayMs: 1_000 },
  shouldRetry: (errorMessage) => {
    // Do not retry if the resource was not found — it will never succeed.
    if (errorMessage.includes('404')) return false;
    // Do not retry authentication failures.
    if (errorMessage.includes('401')) return false;
    // Retry everything else (timeouts, 5xx errors, etc.).
    return true;
  },
});
```

### Default options for all tasks

Pass options to the `TaskManager` constructor to apply them as defaults to every `run()` and `pipeline()` call. Individual calls can still override them.

```typescript
const manager = new TaskManager(store, {
  // Every task gets 3 retries by default...
  maxRetries: 3,
  backoff: { type: 'exponential', initialDelayMs: 500 },
  // ...and a 2-minute overall timeout for waiters.
  giveUpAfterMs: 120_000,
});

// This call inherits maxRetries: 3 from the defaults above,
// but overrides the timeout to 5 minutes.
await manager.run('long-job', work, { giveUpAfterMs: 300_000 });

// This call gets all defaults as-is.
await manager.run('quick-job', work);
```

### Per-step options in pipelines

Each step in a pipeline can have its own options that override the pipeline-level defaults. This lets you tune each step individually.

```typescript
const pipeline = manager.pipeline('data-import', [
  {
    taskId: 'import:download',
    run: downloadFromS3,
    options: {
      // Download can take a long time — give it more headroom.
      giveUpAfterMs: 600_000,   // 10 minutes
    },
  },
  {
    taskId: 'import:validate',
    run: validateSchema,
    // Uses the pipeline-level defaults.
  },
  {
    taskId: 'import:insert',
    run: insertIntoDatabase,
    options: {
      // Database writes are fragile — retry aggressively.
      maxRetries: 10,
      backoff: { type: 'exponential', initialDelayMs: 200, maxDelayMs: 5_000 },
      shouldRetry: (err) => err.includes('deadlock'),
    },
  },
], {
  // Pipeline-level defaults applied to every step.
  maxRetries: 2,
  backoff: { type: 'fixed', delayMs: 1_000 },
});
```

### Tuning lock and heartbeat timing

The defaults work well for most jobs, but you can tune them for very short or very long tasks.

```typescript
await manager.run('quick-check', work, {
  // Lock only needs to live for a short job.
  lockTtlMs: 3_000,           // lock expires after 3 s if not renewed
  heartbeatIntervalMs: 1_000, // renew every 1 s (must be less than lockTtlMs)

  // Waiters check for completion frequently.
  pollIntervalMs: 100,        // check every 100 ms

  // Waiters attempt a takeover quickly if the owner goes silent.
  takeoverAfterMs: 5_000,     // take over after 5 s of silence
  giveUpAfterMs: 15_000,      // give up entirely after 15 s
});
```

**Rule of thumb:** `lockTtlMs` must be comfortably larger than `heartbeatIntervalMs`. If the heartbeat fires every 3 seconds, the lock TTL should be at least 6–10 seconds, so a single missed heartbeat does not cause the lock to expire prematurely.

### Reading and checking results

```typescript
// Was the task completed or failed?
const status = await store.getStatus('my-task');
// 'running' | 'completed' | 'failed' | null (not started)

// What went wrong?
const error = await manager.getError('my-task');
// The error message string, or null if the task succeeded or hasn't failed.

// How far along is it?
const progress = await manager.getProgress('my-task');
// A number from 0 to 1, or null if no progress has been reported yet.

// Reset the task entirely so it can be run again.
await manager.clearTask('my-task');
```

### Building a progress polling endpoint

A common pattern is to kick off a long-running task and let the client poll a separate endpoint for progress.

```typescript
// POST /reports/:id — kicks off the job
app.post('/reports/:id', async (req, res) => {
  const taskId = `report:${req.params.id}`;

  // Fire and forget — do not await so the HTTP response returns immediately.
  manager.run(taskId, async (reportProgress) => {
    await buildReport(req.params.id, reportProgress);
  });

  res.json({ taskId, message: 'Report generation started.' });
});

// GET /reports/:id/progress — client polls this
app.get('/reports/:id/progress', async (req, res) => {
  const taskId = `report:${req.params.id}`;

  const [progress, error] = await Promise.all([
    manager.getProgress(taskId),
    manager.getError(taskId),
  ]);

  const status = await store.getStatus(taskId);

  res.json({
    status:   status ?? 'pending',
    progress: progress ?? 0,
    error:    error ?? null,
  });
});
```

---

## API reference

### `TaskManager`

```typescript
new TaskManager(store: TaskStore, defaultOptions?: TaskOptions)
```

The main entry point. Creates a task manager backed by the given store. `defaultOptions` are merged into every `run()` and `pipeline()` call and can be overridden per-call.

---

#### `manager.run(taskId, work, options?)`

```typescript
run(
  taskId:  string,
  work:    (reportProgress: ProgressReporter) => Promise<void>,
  options?: TaskOptions,
): Promise<boolean>
```

Runs `work` under a distributed lock identified by `taskId`.

| Parameter | Description |
|---|---|
| `taskId` | A unique string that identifies this unit of work. Two callers with the same `taskId` will be deduplicated. |
| `work` | The async function to run. Receives a `reportProgress(n)` callback you can call with a number from `0` to `1`. |
| `options` | Optional overrides for this specific call (see [Options](#options) below). |

Returns `true` if the work completed successfully, `false` if it failed.

---

#### `manager.pipeline(pipelineId, steps, options?)`

```typescript
pipeline(
  pipelineId: string,
  steps:      PipelineStep[],
  options?:   PipelineOptions,
): Pipeline
```

Creates a `Pipeline` object. Call `.run()` on the returned object to execute it.

---

#### `manager.getProgress(taskId)`

```typescript
getProgress(taskId: string): Promise<number | null>
```

Returns the most recently reported progress value (0–1), or `null` if no progress has been reported yet.

---

#### `manager.getError(taskId)`

```typescript
getError(taskId: string): Promise<string | null>
```

Returns the cached error message for a failed task, or `null` if the task has not failed (or the error TTL has expired).

---

#### `manager.clearTask(taskId)`

```typescript
clearTask(taskId: string): Promise<void>
```

Deletes all stored state for the task (lock, status, progress, error). The next `run()` call will execute the work function again from scratch.

---

### `Pipeline`

Created by `manager.pipeline(...)`. Never instantiated directly.

---

#### `pipeline.run()`

```typescript
run(): Promise<boolean>
```

Executes the pipeline step by step. Returns `true` if every step completed, `false` if any step failed. On failure, remaining steps are not run.

---

#### `pipeline.getStatus()`

```typescript
getStatus(): Promise<PipelineStatus>
```

Returns a snapshot of the pipeline's current state. Safe to call from any process at any time, including while the pipeline is running.

```typescript
interface PipelineStatus {
  pipelineId:      string;
  currentStep:     number;          // zero-based index of the active step
  totalSteps:      number;
  overallProgress: number;          // 0–1, weighted across all steps
  status:          'running' | 'completed' | 'failed';
  failedTaskId?:   string;          // taskId of the step that failed
  error?:          string;          // error message from the failed step
}
```

---

### Options

All options are optional. Values shown are the defaults.

```typescript
interface TaskOptions {
  // --- Locking -----------------------------------------------------------------

  /** How long the lock lives without a heartbeat renewal, in ms. */
  lockTtlMs?: number;              // default: 10_000 (10 seconds)

  /** How often the lock owner sends a heartbeat to keep the lock alive, in ms. */
  heartbeatIntervalMs?: number;    // default: 3_000  (3 seconds)

  // --- Waiting -----------------------------------------------------------------

  /** How often a waiting process checks whether the task has finished, in ms. */
  pollIntervalMs?: number;         // default: 500    (0.5 seconds)

  /**
   * How long a waiter will wait before assuming the owner has died
   * and attempting to take over the task, in ms.
   */
  takeoverAfterMs?: number;        // default: 30_000 (30 seconds)

  /**
   * Hard timeout for waiters. If the task is still not done after this
   * many ms, the waiter throws an error, in ms.
   */
  giveUpAfterMs?: number;          // default: 120_000 (2 minutes)

  // --- Retries -----------------------------------------------------------------

  /** Maximum number of retries after the initial attempt fails. */
  maxRetries?: number;             // default: 0 (no retries)

  /** Strategy that controls how long to wait between retries. */
  backoff?: BackoffStrategy;       // default: { type: 'fixed', delayMs: 1_000 }

  /**
   * Called before each retry with the error message from the last attempt.
   * Return false to stop retrying early.
   */
  shouldRetry?: (errorMessage: string) => boolean;  // default: () => true

  // --- Error caching -----------------------------------------------------------

  /** How long the error message is kept in the store after a failure, in ms. */
  errorTtlMs?: number;             // default: 60_000 (1 minute)
}
```

`PipelineOptions` extends `TaskOptions` with one additional field:

```typescript
interface PipelineOptions extends TaskOptions {
  /**
   * When true (the default), steps already marked 'completed' in the store
   * are skipped on re-run, enabling partial resume after a crash.
   */
  resumeFromLastCompleted?: boolean;  // default: true
}
```

---

### Backoff strategies

#### Fixed

```typescript
{ type: 'fixed'; delayMs: number }
```

Waits exactly `delayMs` milliseconds before every retry.

#### Exponential

```typescript
{
  type:           'exponential';
  initialDelayMs: number;        // base delay for the first retry
  multiplier?:    number;        // how much to multiply each time  (default: 2)
  maxDelayMs?:    number;        // ceiling — never wait longer than this (default: 30_000)
  jitter?:        boolean;       // randomise within [0, delay] to spread retries (default: true)
}
```

With `multiplier: 2` and `initialDelayMs: 500`, the delays before retries are:

| Retry | No jitter | With jitter |
|---|---|---|
| 1st | 500 ms | 0–500 ms |
| 2nd | 1 000 ms | 0–1 000 ms |
| 3rd | 2 000 ms | 0–2 000 ms |
| 4th | 4 000 ms | 0–4 000 ms |
| 5th+ | capped at `maxDelayMs` | 0–`maxDelayMs` |

---

## Implementing a custom store

`InMemoryTaskStore` and `RedisTaskStore` are the two built-in implementations, but you can write your own for any backing store (PostgreSQL, DynamoDB, etc.) by implementing the `TaskStore` interface.

```typescript
import { TaskStore, TaskStatus } from 'task-manager';

class PostgresTaskStore implements TaskStore {
  async acquireLock(taskId, token, ttlMs) { /* ... */ }
  async renewLock(taskId, token, ttlMs)   { /* ... */ }
  async releaseLock(taskId, token)        { /* ... */ }

  async getStatus(taskId)                 { /* ... */ }
  async setStatus(taskId, status)         { /* ... */ }

  async getProgress(taskId)               { /* ... */ }
  async setProgress(taskId, progress)     { /* ... */ }

  async getError(taskId)                  { /* ... */ }
  async setError(taskId, error, ttlMs)    { /* ... */ }

  async clearTask(taskId)                 { /* ... */ }
}
```

The only method with strict requirements is `acquireLock`: it **must be atomic**. It should succeed (return `true`) only if no lock currently exists for `taskId`, and it must set the new lock in the same atomic operation — there must be no gap between checking and writing. In Redis this is done with a Lua script or `SET NX`. In PostgreSQL it can be done with `INSERT ... ON CONFLICT DO NOTHING`.

All other methods have no atomicity requirements.

---

## TypeScript types

All types are exported from the package root:

```typescript
import type {
  TaskStore,          // the interface you implement for a custom store
  TaskStatus,         // 'running' | 'completed' | 'failed'
  TaskOptions,        // options accepted by run() and pipeline()
  ResolvedTaskOptions,// options with all defaults filled in
  BackoffStrategy,    // fixed | exponential
  WorkFn,             // (reportProgress: ProgressReporter) => Promise<void>
  ProgressReporter,   // (progress: number) => Promise<void>
  PipelineStep,       // { taskId, run, options? }
  PipelineOptions,    // TaskOptions + resumeFromLastCompleted
  PipelineContext,    // { pipelineId, stepIndex, totalSteps }
  PipelineStatus,     // what getStatus() returns
} from 'task-manager';
```

---

## Development

```bash
# Install dependencies
npm install

# Build (emits to dist/)
npm run build

# Watch mode
npm run dev

# Run tests
npm test

# Run tests in watch mode
npm run test:watch
```