Bulk datastore

.github/workflows/ci.yml

@@ -15,8 +15,7 @@ jobs:
     timeout-minutes: 10
     strategy:
       matrix:
-        # 22.22.2+ toolcache ships a broken global npm (missing promise-retry); pin until fixed upstream.
-        node-version: [20.x, 22.22.1, 24.x]
+        node-version: [22.x, 24.x, 26.x]
 
     steps:
       - name: Perform source code checkout
@@ -42,7 +41,7 @@ jobs:
         run: |
           npm install -g npm@11.2.0
           npm install -g pnpm@10.30.0
-          pnpm install
+          pnpm install --ignore-scripts
 
       - name: Build
         run: pnpm run build

.nvmrc

@@ -1 +1 @@
-20
+22

biome.json

@@ -1,5 +1,5 @@
 {
-  "$schema": "https://biomejs.dev/schemas/2.4.11/schema.json",
+  "$schema": "https://biomejs.dev/schemas/2.5.6/schema.json",
   "vcs": {
     "enabled": false,
     "clientKind": "git",
@@ -20,7 +20,7 @@
   "linter": {
     "enabled": true,
     "rules": {
-      "recommended": true
+      "preset": "recommended"
     }
   },
   "json": {

examples/sample-memory-datastore-app/src/main.ts

@@ -18,7 +18,7 @@ type TaskMapping = {
  * Consumers only need `ChronoHandlerRegistrar` -- they never see `use()` or
  * `scheduleTask()`, which keeps the type covariant in TaskMapping.
  */
-function registerHandlers(registrar: ChronoHandlerRegistrar<TaskMapping>) {
+function registerHandlers(registrar: ChronoHandlerRegistrar<TaskMapping, DatastoreOptions>) {
   const processor1 = registrar.registerTaskHandler({
     kind: 'async-messaging',
     handler: async (task) => {

examples/sample-memory-datastore-app/tsconfig.json

@@ -1,11 +1,8 @@
 {
   "extends": "../../tsconfig.base.json",
   "compilerOptions": {
-    "baseUrl": "${configDir}/src",
-    "outDir": "${configDir}/build",
-    "paths": {
-      "*": ["*", "node_modules/*", "src/*"]
-    }
+    "rootDir": "./src",
+    "outDir": "./build"
   },
   "include": ["src/**/*"]
 }

package.json

@@ -2,7 +2,7 @@
   "name": "chrono-monorepo",
   "description": "Monorepo for chrono packages",
   "engines": {
-    "node": ">=20.18.3",
+    "node": ">=22.18.0",
     "pnpm": ">=10.6.2"
   },
   "keywords": [],
@@ -23,16 +23,16 @@
     "publish:alpha": "pnpm i && pnpm build && pnpm publish -r --tag alpha"
   },
   "devDependencies": {
-    "@biomejs/biome": "^2.4.11",
+    "@biomejs/biome": "^2.5.6",
     "@faker-js/faker": "^9.9.0",
-    "@types/node": "^20.19.31",
+    "@types/node": "^22.20.1",
     "fishery": "^2.4.0",
     "husky": "^9.1.7",
-    "lint-staged": "^16.4.0",
+    "lint-staged": "^17.3.0",
     "rimraf": "^6.1.3",
-    "tsdown": "^0.21.7",
-    "typescript": "^5.9.3",
-    "vitest": "^4.1.4",
-    "vitest-mock-extended": "^4.0.0"
+    "tsdown": "^0.22.14",
+    "typescript": "^6.0.3",
+    "vitest": "^4.1.10",
+    "vitest-mock-extended": "^5.1.1"
   }
 }

packages/chrono-core/README.md

@@ -172,13 +172,18 @@ Delay doubles each retry with an optional cap and jitter.
 
 ## Processor Configuration
 
-Each task handler runs on a processor that polls the datastore for tasks. Configure processor behavior via `processorConfiguration`:
+Each task handler runs on a processor that polls the datastore for tasks. Configure processor behavior via `processorConfiguration`.
+
+### Simple processor (default)
+
+Claims and processes one task at a time per worker loop:
 
 ```typescript
 chrono.registerTaskHandler({
   kind: "send-email",
   handler: async (task) => { /* ... */ },
   processorConfiguration: {
+    type: "simple", // optional — this is the default
     maxConcurrency: 5,
     claimIntervalMs: 100,
     taskHandlerTimeoutMs: 30_000,
@@ -187,7 +192,7 @@ chrono.registerTaskHandler({
 });
 ```
 
-### Options
+#### Simple processor options
 
 | Option | Type | Default | Description |
 |--------|------|---------|-------------|
@@ -199,6 +204,37 @@ chrono.registerTaskHandler({
 | `taskHandlerMaxRetries` | `number` | `5` | Maximum number of retries before a task is marked as failed |
 | `processLoopRetryIntervalMs` | `number` | `20000` | Interval in ms before retrying after an unexpected error in the processing loop |
 
+### Bulk processor
+
+Claims and processes tasks in batches. Requires a datastore that implements `BulkDatastore` (for example `ChronoMongoDatastore`):
+
+```typescript
+chrono.registerTaskHandler({
+  kind: "send-email",
+  handler: async (task) => { /* ... */ },
+  processorConfiguration: {
+    type: "bulk",
+    batchSize: 50,
+    batchIntervalMs: 1_000,
+    taskHandlerTimeoutMs: 30_000,
+    taskHandlerMaxRetries: 10,
+  },
+});
+```
+
+`type: "bulk"` is only accepted when the `Chrono` instance was constructed with a bulk-capable datastore. TypeScript will report a compile-time error otherwise; a runtime guard also throws if bulk configuration is used with a non-bulk datastore.
+
+#### Bulk processor options
+
+| Option | Type | Default | Description |
+|--------|------|---------|-------------|
+| `batchSize` | `number` | `25` | Maximum number of tasks to claim per batch |
+| `claimStaleTimeoutMs` | `number` | `10000` | Time in ms before a claimed task is considered stale and can be re-claimed |
+| `taskHandlerTimeoutMs` | `number` | `5000` | Maximum time in ms a task handler can run before timing out |
+| `taskHandlerMaxRetries` | `number` | `5` | Maximum number of retries before a task is marked as failed |
+| `batchIntervalMs` | `number` | `5000` | Interval in ms between batch processing loop iterations |
+| `processLoopRetryIntervalMs` | `number` | `20000` | Interval in ms before retrying after an unexpected error in the processing loop |
+
 ## Events
 
 ### Chrono Events
@@ -370,8 +406,20 @@ See the existing implementations for reference:
 | `Task` | Type | Task document type |
 | `TaskMappingBase` | Type | Base type constraint for task mappings |
 | `ScheduleTaskInput` | Type | Input type for `scheduleTask()` |
-| `RegisterTaskHandlerInput` | Type | Input type for `registerTaskHandler()` |
+| `RegisterTaskHandlerInput` | Type | Discriminated union input type for `registerTaskHandler()` (simple or bulk) |
+| `RegisterTaskHandlerSimpleInput` | Type | Simple processor registration input |
+| `RegisterTaskHandlerBulkInput` | Type | Bulk processor registration input (`type: 'bulk'` required) |
 | `RegisterTaskHandlerResponse` | Type | Return type of `registerTaskHandler()` |
+| `ProcessorConfiguration` | Type | Processor configuration union (`simple` \| `bulk`) |
+| `SimpleProcessor` | Class | Simple (single-task) processor implementation |
+| `SimpleProcessorConfiguration` | Type | Configuration for the simple processor |
+| `BulkProcessor` | Class | Bulk (batch) processor implementation |
+| `BulkProcessorConfiguration` | Type | Configuration for the bulk processor |
+| `BulkDatastore` | Interface | Bulk datastore interface (`claimMany`, `completeMany`, etc.) |
+| `isBulkDatastore` | Function | Runtime guard for `BulkDatastore` support |
+| `BulkWriteResult` | Type | Result type for bulk datastore write operations |
+| `ClaimManyInput` | Type | Input type for `claimMany()` |
+| `RetryManyItem` | Type | Input item type for `retryMany()` |
 | `ScheduleInput` | Type | Datastore-level schedule input |
 | `ClaimTaskInput` | Type | Datastore-level claim input |
 | `DeleteInput` | Type | Datastore-level delete input |

packages/chrono-core/src/bulk-datastore.ts

@@ -0,0 +1,54 @@
+import type { TaskMappingBase } from './chrono';
+import type { Datastore, Task } from './datastore';
+
+export type ClaimManyInput<TaskKind> = {
+  kind: TaskKind;
+  batchSize: number;
+  claimStaleTimeoutMs: number;
+};
+
+export type RetryManyItem = {
+  taskId: string;
+  retryAt: Date;
+};
+
+export type BulkWriteResult<TaskKind, TaskData> = {
+  succeeded: Task<TaskKind, TaskData>[];
+  failed: { taskId: string; error: unknown }[];
+};
+
+export interface BulkDatastore<TaskMapping extends TaskMappingBase, _DatastoreOptions> {
+  claimMany<TaskKind extends Extract<keyof TaskMapping, string>>(
+    input: ClaimManyInput<TaskKind>,
+  ): Promise<Task<TaskKind, TaskMapping[TaskKind]>[]>;
+
+  completeMany<TaskKind extends keyof TaskMapping>(
+    taskIds: string[],
+  ): Promise<BulkWriteResult<TaskKind, TaskMapping[TaskKind]>>;
+
+  retryMany<TaskKind extends keyof TaskMapping>(
+    items: RetryManyItem[],
+  ): Promise<BulkWriteResult<TaskKind, TaskMapping[TaskKind]>>;
+
+  failMany<TaskKind extends keyof TaskMapping>(
+    taskIds: string[],
+  ): Promise<BulkWriteResult<TaskKind, TaskMapping[TaskKind]>>;
+}
+
+/**
+ * Runtime guard for {@link BulkDatastore} support on a {@link Datastore} instance.
+ */
+export function isBulkDatastore<TaskMapping extends TaskMappingBase, DatastoreOptions>(
+  datastore: Datastore<TaskMapping, DatastoreOptions>,
+): datastore is Datastore<TaskMapping, DatastoreOptions> & BulkDatastore<TaskMapping, DatastoreOptions> {
+  return (
+    'claimMany' in datastore &&
+    typeof datastore.claimMany === 'function' &&
+    'completeMany' in datastore &&
+    typeof datastore.completeMany === 'function' &&
+    'retryMany' in datastore &&
+    typeof datastore.retryMany === 'function' &&
+    'failMany' in datastore &&
+    typeof datastore.failMany === 'function'
+  );
+}

packages/chrono-core/src/chrono.ts

@@ -1,13 +1,16 @@
 import { EventEmitter } from 'node:events';
 
 import type { BackoffStrategyOptions } from './backoff-strategy';
+import { type BulkDatastore, isBulkDatastore } from './bulk-datastore';
 import type { Datastore, ScheduleInput, Task } from './datastore';
 import { ChronoEvents, type ChronoEventsMap } from './events';
 import type { ChronoPlugin } from './plugins';
 import { ChronoPluginContext } from './plugins/chrono-plugin-context';
 import { createProcessor, type Processor } from './processors';
+import type { BulkProcessorConfiguration } from './processors/bulk-processor';
 import type { ProcessorConfiguration } from './processors/create-processor';
 import type { ProcessorEventsMap } from './processors/events';
+import type { SimpleProcessorConfiguration } from './processors/simple-processor';
 import { promiseWithTimeout } from './utils/promise-utils';
 
 export type TaskMappingBase = Record<string, unknown>;
@@ -18,17 +21,35 @@ export type ScheduleTaskInput<TaskKind, TaskData, DatastoreOptions> = ScheduleIn
   DatastoreOptions
 >;
 
-export type RegisterTaskHandlerInput<TaskKind, TaskData> = {
+type RegisterTaskHandlerBase<TaskKind, TaskData> = {
   /** The type of task */
   kind: TaskKind;
   /** The handler function to process the task */
   handler: (task: Task<TaskKind, TaskData>) => Promise<void>;
   /** The options for the backoff strategy to use when the task handler fails */
   backoffStrategyOptions?: BackoffStrategyOptions;
-  /** The configuration for the processor to use when processing the task */
-  processorConfiguration?: ProcessorConfiguration;
 };
 
+export type RegisterTaskHandlerSimpleInput<TaskKind, TaskData> = RegisterTaskHandlerBase<TaskKind, TaskData> & {
+  /** The configuration for the simple processor to use when processing the task */
+  processorConfiguration?: Partial<SimpleProcessorConfiguration> & { type?: 'simple' };
+};
+
+export type RegisterTaskHandlerBulkInput<TaskKind, TaskData> = RegisterTaskHandlerBase<TaskKind, TaskData> & {
+  /** The configuration for the bulk processor to use when processing the task */
+  processorConfiguration: Partial<BulkProcessorConfiguration> & { type: 'bulk' };
+};
+
+export type RegisterTaskHandlerInput<TaskKind, TaskData> =
+  | RegisterTaskHandlerSimpleInput<TaskKind, TaskData>
+  | RegisterTaskHandlerBulkInput<TaskKind, TaskData>;
+
+type BulkDatastoreRegistrationCheck<
+  TaskMapping extends TaskMappingBase,
+  DatastoreOptions,
+  DatastoreImpl extends Datastore<TaskMapping, DatastoreOptions>,
+> = DatastoreImpl extends BulkDatastore<TaskMapping, DatastoreOptions> ? unknown : never;
+
 /**
  * Response from registering a task handler.
  * @returns The processor instance that can be used to start and stop the processor.
@@ -63,9 +84,18 @@ export interface ChronoTaskScheduler<TaskMapping extends TaskMappingBase, Datast
  * which is the key property needed to pass a single wide Chrono to multiple
  * narrowly-typed outbox handlers without any casts.
  */
-export interface ChronoHandlerRegistrar<out TaskMapping extends TaskMappingBase> {
+export interface ChronoHandlerRegistrar<
+  TaskMapping extends TaskMappingBase,
+  DatastoreOptions,
+  DatastoreImpl extends Datastore<TaskMapping, DatastoreOptions> = Datastore<TaskMapping, DatastoreOptions>,
+> {
   registerTaskHandler<TaskKind extends Extract<keyof TaskMapping, string>>(
-    input: RegisterTaskHandlerInput<TaskKind, TaskMapping[TaskKind]>,
+    input: RegisterTaskHandlerSimpleInput<TaskKind, TaskMapping[TaskKind]>,
+  ): RegisterTaskHandlerResponse<TaskKind, TaskMapping>;
+
+  registerTaskHandler<TaskKind extends Extract<keyof TaskMapping, string>>(
+    input: RegisterTaskHandlerBulkInput<TaskKind, TaskMapping[TaskKind]> &
+      BulkDatastoreRegistrationCheck<TaskMapping, DatastoreOptions, DatastoreImpl>,
   ): RegisterTaskHandlerResponse<TaskKind, TaskMapping>;
 }
 
@@ -74,18 +104,24 @@ export interface ChronoHandlerRegistrar<out TaskMapping extends TaskMappingBase>
  * @param datastore - The datastore instance to use for storing and retrieving tasks.
  * @returns The Chrono instance that can be used to start and stop the processors as well as receive chrono instance events.
  */
-export class Chrono<TaskMapping extends TaskMappingBase, DatastoreOptions>
+export class Chrono<
+    TaskMapping extends TaskMappingBase,
+    DatastoreOptions,
+    DatastoreImpl extends Datastore<TaskMapping, DatastoreOptions> = Datastore<TaskMapping, DatastoreOptions>,
+  >
   extends EventEmitter<ChronoEventsMap>
-  implements ChronoHandlerRegistrar<TaskMapping>, ChronoTaskScheduler<TaskMapping, DatastoreOptions>
+  implements
+    ChronoHandlerRegistrar<TaskMapping, DatastoreOptions, DatastoreImpl>,
+    ChronoTaskScheduler<TaskMapping, DatastoreOptions>
 {
-  private readonly datastore: Datastore<TaskMapping, DatastoreOptions>;
+  private readonly datastore: DatastoreImpl;
   private readonly processors: Map<keyof TaskMapping, Processor<keyof TaskMapping, TaskMapping>> = new Map();
-  private readonly pluginContexts: ChronoPluginContext<TaskMapping, DatastoreOptions>[] = [];
+  private readonly pluginContexts: ChronoPluginContext<TaskMapping, DatastoreOptions, DatastoreImpl>[] = [];
   private started = false;
 
   readonly exitTimeoutMs = 60_000;
 
-  constructor(datastore: Datastore<TaskMapping, DatastoreOptions>) {
+  constructor(datastore: DatastoreImpl) {
     super();
 
     this.datastore = datastore;
@@ -97,12 +133,16 @@ export class Chrono<TaskMapping extends TaskMappingBase, DatastoreOptions>
    * @param plugin - The plugin to register
    * @returns The plugin's API (if any) for type-safe access to plugin functionality
    */
-  use<PluginAPI>(plugin: ChronoPlugin<TaskMapping, DatastoreOptions, PluginAPI>): PluginAPI {
+  use<PluginAPI>(plugin: ChronoPlugin<TaskMapping, DatastoreOptions, PluginAPI, DatastoreImpl>): PluginAPI {
     if (this.started) {
       throw new Error(`Cannot register plugin "${plugin.name}" after Chrono has started`);
     }
 
-    const context = new ChronoPluginContext<TaskMapping, DatastoreOptions>(this, this.processors, this.datastore);
+    const context = new ChronoPluginContext<TaskMapping, DatastoreOptions, DatastoreImpl>(
+      this,
+      this.processors,
+      this.datastore,
+    );
 
     const api = plugin.register(context);
 
@@ -165,19 +205,32 @@ export class Chrono<TaskMapping extends TaskMappingBase, DatastoreOptions>
     return task;
   }
 
+  public registerTaskHandler<TaskKind extends Extract<keyof TaskMapping, string>>(
+    input: RegisterTaskHandlerSimpleInput<TaskKind, TaskMapping[TaskKind]>,
+  ): RegisterTaskHandlerResponse<TaskKind, TaskMapping>;
+
+  public registerTaskHandler<TaskKind extends Extract<keyof TaskMapping, string>>(
+    input: RegisterTaskHandlerBulkInput<TaskKind, TaskMapping[TaskKind]> &
+      BulkDatastoreRegistrationCheck<TaskMapping, DatastoreOptions, DatastoreImpl>,
+  ): RegisterTaskHandlerResponse<TaskKind, TaskMapping>;
+
   public registerTaskHandler<TaskKind extends Extract<keyof TaskMapping, string>>(
     input: RegisterTaskHandlerInput<TaskKind, TaskMapping[TaskKind]>,
   ): RegisterTaskHandlerResponse<TaskKind, TaskMapping> {
     if (this.processors.has(input.kind)) {
       throw new Error('Handler for task kind already exists');
     }
 
+    if (input.processorConfiguration?.type === 'bulk' && !isBulkDatastore(this.datastore)) {
+      throw new Error('Bulk processor requires a datastore that implements BulkDatastore');
+    }
+
     const processor = createProcessor({
       kind: input.kind,
       datastore: this.datastore,
       handler: input.handler,
       backoffStrategyOptions: input.backoffStrategyOptions,
-      configuration: input.processorConfiguration,
+      configuration: input.processorConfiguration satisfies ProcessorConfiguration | undefined,
     });
 
     this.processors.set(input.kind, processor);