@polyweave/durability (1.0.9)

Published 2026-07-11 15:23:55 +00:00 by Dvorak

Set up this registry in your project .npmrc file:

@polyweave:registry=https://hub.kl1.tenere.ai/api/packages/Polyweave/npm/

To install the package using npm, run the following command:

npm install @polyweave/durability@1.0.9

or add it to the package.json file:

"@polyweave/durability": "1.0.9"

For more information on the npm registry, see the documentation.

@polyweave/durability

WAL-based crash recovery, segmented SSTable-style stores, checkpoint/recovery managers, durable pipelines, durable routers, durable systems, and snapshot persistence helpers for Polyweave.

This package consolidates what was previously split across @polyweave/core/durability and @polyweave/persistence.

Synopsis

import {
  FileWALStore,
  MemoryWALStore,
  MapletCompactionManager,
  MapletCompactionConfig,
  CheckpointManager,
  RecoveryManager,
  createDurablePipeline,
  createDurableSystem,
  createDurableConversation,
  createDurableRouter,
  createTreeSnapshotClient,
  createFactSnapshotClient,
  normalizeTreeCheckpoint,
  restoreStoreFromCheckpoint,
  checkpointTree,
  checkpointFacts,
  rebuildTextIndexFromTree,
  WALEntryType
} from '@polyweave/durability';

// --- File-based WAL store ---
const walStore = new FileWALStore({
  dataDir: '/var/lib/polyweave/wal',
  encoding: 'jsonl',
  syncIntervalMs: 100
});

// --- Durable pipeline wrapping an adapter ---
const pipeline = createDurablePipeline({
  store: walStore,
  provider: myAdapter,
  wrappers: [
    { type: 'queue', id: 'main-queue', options: { concurrency: 4 } },
    { type: 'rateLimit', id: 'main-rl', limits: { maxTokensPerMinute: 100000 } },
    { type: 'retry', id: 'main-retry', options: { maxRetries: 3 } }
  ],
  checkpointEvery: 100,
  onReady: (err, recoveryResult) => {
    if (err) console.error(err);
    else console.log(`Recovered ${recoveryResult.framesToReplay?.length} frames`);
  }
});

// Send frames through the durable pipeline
pipeline.duplex.sink.write({ type: 'TEXT', id: 'msg-1', streamId: 's1', content: { text: 'hello' } });

// Force a checkpoint
pipeline.checkpoint((err, lsn) => {
  console.log('Checkpoint LSN:', lsn);
});

// Graceful shutdown
pipeline.close((err) => {
  if (!err) console.log('Shut down');
});

// --- Durable router ---
const router = createDurableRouter({
  dataDir: '/var/lib/polyweave/routes',
  factory: (streamId) => createDialogWrapper(adapter, { maxMessages: 50 }),
  idleTimeout: 300
});

router.sink.write({ type: 'TEXT', streamId: 'user-42', content: { text: 'hi' } });
router.source.pipe(outputSink);

// --- Snapshot + checkpoint of a tree store ---
const snapshotClient = createTreeSnapshotClient(treeStore, 'full');
checkpointTree(walStore, snapshotClient, {
  wrapperId: 'tree-1',
  wrapperType: 'treeStore'
}, (err, lsn) => {
  if (err) throw err;
});

// --- Restore a store from checkpoint ---
restoreStoreFromCheckpoint({
  walStore,
  store: restoredStore,
  wrapperId: 'tree-1',
  wrapperType: 'treeStore',
  normalizeState: normalizeTreeCheckpoint,
  onDone: (err, restored) => {
    console.log('Restored:', restored);
  }
});

Install

npm install @polyweave/durability

Why

Polyweave pipelines — AI dialog, agent loops, routing — need crash recovery. If a process dies mid-conversation, queued messages, in-flight frames, rate-limit counters, and retry state must survive. This package provides WAL-based persistence (with synchronous append for minimal overhead), SSTable-style segmented storage with Maplet quotient filters for O(1) lookups, checkpoint managers that snapshot wrapper state every N operations, recovery managers that replay uncommitted frames and adjust timers after crash, and three convenience entry points: createDurablePipeline (single pipeline), createDurableSystem (auto-intercepts durability META frames from wrappers), and createDurableRouter (routes frames to isolated WAL-backed instances by streamId). The snapshot helpers bridge the gap between WAL durability and Polyweave state stores (trees, facts), enabling full-system persistence.

API

WAL Types and Utilities

`WALEntryType`

enum WALEntryType

Frozen object enumerating WAL entry types.

Value	Description
`FRAME_ENQUEUED`	Frame entered the pipeline
`FRAME_PROCESSED`	Frame completed successfully
`FRAME_FAILED`	Frame processing failed
`FRAME_RETRYING`	Frame scheduled for retry
`TIMER_SCHEDULED`	Timer was set
`TIMER_FIRED`	Timer fired
`TIMER_CANCELLED`	Timer was cancelled
`STATE_CHANGE`	Wrapper state changed
`CHECKPOINT`	Full state checkpoint

`CommitStatus`

enum CommitStatus

Value	Description
`PENDING`	Frame in-flight
`COMMITTED`	Frame completed successfully
`FAILED`	Frame failed (no more retries)
`CANCELLED`	Frame was cancelled

`createWALEntry(type, payload, options?)`

createWALEntry(type: WALEntryType, payload: Object, options?: Object): WALEntry

Creates a generic WAL entry shell.

Name	Type	Required	Default	Description
`type`	`string`	yes	—	Entry type from `WALEntryType`
`payload`	`Object`	yes	—	Entry-specific payload (spread into entry)
`options`	`Object`	no	`{}`	Additional fields spread into entry

Return value: { type, timestamp, lsn: null, ...payload, ...options }. The lsn field is set to null and will be assigned by the store on append.

`createFrameEnqueuedEntry(frame, wrapperId)`

createFrameEnqueuedEntry(frame: Frame, wrapperId: string): WALEntry

Creates a FRAME_ENQUEUED entry with serialized frame data.

Name	Type	Required	Description
`frame`	`Frame`	yes	The frame being enqueued
`wrapperId`	`string`	yes	Wrapper that owns the frame

Return value: WAL entry with frameId, streamId, wrapperId, frame (serialized via serializeFrame), enqueuedAt, type: FRAME_ENQUEUED.

`createFrameProcessedEntry(frameId, wrapperId, responseFrame?)`

createFrameProcessedEntry(frameId: string, wrapperId: string, responseFrame?: Frame | null): WALEntry

Creates a FRAME_PROCESSED entry.

Name	Type	Required	Description
`frameId`	`string`	yes	ID of completed frame
`wrapperId`	`string`	yes	Wrapper that processed it
`responseFrame`	`Frame\|null`	no	Optional response frame (serialized)

`createFrameFailedEntry(frameId, wrapperId, error)`

createFrameFailedEntry(frameId: string, wrapperId: string, error: Error | Object): WALEntry

Creates a FRAME_FAILED entry.

Name	Type	Required	Description
`frameId`	`string`	yes	ID of failed frame
`wrapperId`	`string`	yes	Wrapper that failed
`error`	`Error\|Object`	yes	Error object. `message`/`code`/`stack` extracted into entry

`createFrameRetryingEntry(frameId, wrapperId, retryCount, nextRetryAt)`

createFrameRetryingEntry(frameId: string, wrapperId: string, retryCount: number, nextRetryAt: number): WALEntry

Creates a FRAME_RETRYING entry.

Name	Type	Required	Description
`frameId`	`string`	yes	ID of retrying frame
`wrapperId`	`string`	yes	Wrapper that owns retry
`retryCount`	`number`	yes	Current retry attempt number
`nextRetryAt`	`number`	yes	Absolute timestamp for next retry

`createTimerScheduledEntry(timerId, fireAt, wrapperId, context)`

createTimerScheduledEntry(timerId: string, fireAt: number, wrapperId: string, context: Object): WALEntry

Creates a TIMER_SCHEDULED entry.

`createTimerFiredEntry(timerId)`

createTimerFiredEntry(timerId: string): WALEntry

Creates a TIMER_FIRED entry.

`createTimerCancelledEntry(timerId)`

createTimerCancelledEntry(timerId: string): WALEntry

Creates a TIMER_CANCELLED entry.

`createStateChangeEntry(wrapperId, wrapperType, stateDelta, options?)`

createStateChangeEntry(wrapperId: string, wrapperType: string, stateDelta: Object, options?: {
  pointKey?: string | null
}): WALEntry

Creates a STATE_CHANGE entry.

Name	Type	Required	Default	Description
`wrapperId`	`string`	yes	—	Wrapper whose state changed
`wrapperType`	`string`	yes	—	Wrapper type (`queue`, `rateLimit`, etc.)
`stateDelta`	`Object`	yes	—	The state change delta
`options.pointKey`	`string\|null`	no	`null`	Key for point-lookup indexing

`createCheckpointEntry(pipelineState)`

createCheckpointEntry(pipelineState: PipelineState): WALEntry

Creates a CHECKPOINT entry with full pipeline state.

`serializeFrame(frame)`

serializeFrame(frame: Frame): Frame

Serializes a frame for WAL storage. If frame.content.data is a Buffer, replaces it with a blob reference (blobRef) and marks _hasBinaryData: true. Otherwise returns the frame unchanged.

`deserializeFrame(serialized, blobStore?)`

deserializeFrame(serialized: Frame, blobStore?: { get: (ref: string) => Buffer }): Frame

Deserializes a frame from WAL storage. If _hasBinaryData is truthy and blobStore is provided, restores the binary data from the blob store.

`parseWALEntry(line)`

parseWALEntry(line: string): Object | null

Parses a JSON line into a WAL entry. Returns null for empty/whitespace-only lines.

Error conditions: Throws SyntaxError if the JSON line is malformed.

`stringifyWALEntry(entry)`

stringifyWALEntry(entry: WALEntry): string

Stringifies a WAL entry to a JSON line.

WAL Store Interface

`validateWALStore(store)`

validateWALStore(store: Object): void

Validates that an object implements the complete WALStoreInterface. Checks for the presence of all required methods.

Error conditions: Throws Error('WALStore must implement {method}()') if any required method is missing. The required methods are: append, readUncommitted, commit, checkpoint, loadCheckpoint, truncate, getCurrentLsn, close.

Memory WAL Store

`MemoryWALStore`

new MemoryWALStore(): MemoryWALStore

In-memory WAL store for testing. Data is lost on process restart. Implements the full WAL store interface plus testing helpers.

Constructor: No arguments.

Additional methods (beyond interface):

Method	Returns	Description
`getEntryCount()`	`number`	Total entries stored
`getAllEntries()`	`WALEntry[]`	Shallow copy of all entries
`clear()`	`void`	Resets all internal state

File WAL Store

`FileWALStore`

new FileWALStore(directoryOrOptions: string | {
  dataDir: string,
  encoding?: 'jsonl' | 'binary',
  silent?: boolean,
  syncIntervalMs?: number,
  maxEntriesBeforeRotate?: number,
  enablePointLookup?: boolean,
  pausePointReadsDuringCompaction?: boolean,
  pausePointWritesDuringCompaction?: boolean,
  binaryFieldCacheMax?: number,
  compaction?: {
    autoCompaction?: boolean,
    compactionIntervalMs?: number,
    maxSegmentSize?: number
  }
}, options?: Object): FileWALStore

File-based WAL store with JSONL or binary encoding, point-lookup indexing via Maplet compaction manager, and blob storage for binary frame data. Supports both callback and synchronous APIs.

Constructor overloads:

new FileWALStore(directory, options) — shorthand for string path
new FileWALStore({ dataDir, ...options }) — full options object

Name	Type	Required	Default	Description
`dataDir`	`string`	yes	—	Directory for WAL files
`encoding`	`string`	no	`'jsonl'`	Encoding format: `'jsonl'` or `'binary'`
`silent`	`boolean`	no	`false`	Suppress console output
`syncIntervalMs`	`number`	no	`100`	Interval for periodic flush to disk
`maxEntriesBeforeRotate`	`number`	no	`10000`	Entries before segment rotate
`enablePointLookup`	`boolean`	no	`false`	Enable Maplet-based point-lookup index
`pausePointReadsDuringCompaction`	`boolean`	no	`true`	Skip point-lookup reads during compaction
`pausePointWritesDuringCompaction`	`boolean`	no	`true`	Skip point-lookup writes during compaction
`binaryFieldCacheMax`	`number`	no	`2048`	Max cache entries for binary encoding field deduplication

Error conditions (constructor): Throws Error('directory is required') if no directory provided.

Methods (interface implementations):

Method	Signature	Returns	Description
`append(entry, callback?)`	`(WALEntry, fn?) => number`	LSN	Synchronous append; increments LSN, writes to buffer
`readUncommitted()`	`() => WALEntry[]`	Array	Returns pending entries since last checkpoint
`readAll()`	`() => WALEntry[]`	Array	Returns all scanned entries
`readStateChanges(wrapperId, afterLsn?)`	`(string, number?) => WALEntry[]`	Array	All STATE_CHANGE entries for a wrapper after LSN
`readLatestStateChange(wrapperId, pointKey, afterLsn?)`	`(string, string, number?) => WALEntry\|null`	Entry or null	Latest STATE_CHANGE entry for wrapper+pointKey combo
`readByFrameId(frameId)`	`(string) => WALEntry[]`	Array	All entries matching frame ID
`commit(frameIdOrLsn, status, callback?)`	`(string\|number, string, fn?) => void`	—	Marks frame as COMMITTED/FAILED/CANCELLED
`checkpoint(state, callback?)`	`(PipelineState, fn?) => number`	LSN	Writes checkpoint; resets entry counter
`loadCheckpoint(callback?)`	`(fn?) => PipelineState\|null`	State or null	Loads latest checkpoint from disk. Returns `null` if file missing/corrupt
`truncate(beforeLsn, callback?)`	`(number, fn?) => number`	LSN	No-op (legacy support); resets entry counter
`getCurrentLsn()`	`() => number`	LSN	Current high-water LSN
`close(callback?)`	`(fn?) => Promise\|void`	—	Flushes buffer, closes file handle, stops sync timer, closes point lookup
`getCheckpoint()`	`() => PipelineState\|null`	State or null	Synchronous getter for loaded checkpoint

Blob methods:

Method	Signature	Description
`writeBlob(blobRef, data, callback?)`	`(string, Buffer, fn?) => void`	Writes binary blob to disk
`readBlob(blobRef, callback?)`	`(string, fn?) => Buffer`	Reads binary blob from disk
`deleteBlob(blobRef, callback?)`	`(string, fn?) => void`	Deletes binary blob

Lifecycle: Must call initialize(callback) or open() before use. open() returns a Promise.

Error conditions (methods): Throws Error('Store not initialized') if _ensureInitialized() is called before initialization.

`FileWALStoreLegacy`

Re-export alias for FileWALStore. Provided for backward compatibility.

Maplet Segment

`MapletSegment`

new MapletSegment(options?: {
  capacity?: number,
  remainderBits?: number,
  segmentId?: string,
  level?: number
}): MapletSegment

Immutable SSTable-style segment using Maplet quotient-filter indexing. Based on the Maplet paper (arxiv.org/html/2510.05518). Each segment is an on-disk binary file with a 64-byte header, hashtable slots, and frame data.

Name	Type	Required	Default	Description
`capacity`	`number`	no	`1024` (rounded to next power of 2)	Expected number of entries
`remainderBits`	`number`	no	`16`	Fingerprint remainder bits
`segmentId`	`string`	no	auto-generated	Unique segment ID
`level`	`number`	no	`0`	Compaction level (0 = active)

Instance methods:

Method	Signature	Returns	Description
`insert(key, frameData, lsn)`	`(string, Buffer, number) => boolean`	`true` / `false`	Insert a frame. Returns `false` if full. Uses Robin Hood hashing
`query(key)`	`(string) => { frameData, lsn, offset } \| null`	Result or null	Look up by key (quotient filter search)
`seal(directory)`	`(string) => string`	File path	Seal and write to disk; clears in-memory structures
`enumerate()`	`() => Generator`	Generator	Yield `{ fingerprint, frameData, lsn, offset }` for every entry
`getStats()`	`() => Object`	Stats object	Segment metadata: `segmentId`, `level`, `entryCount`, `tableSize`, `loadFactor`, `minLsn`, `maxLsn`, `fileSize`, `filePath`, `isSealed`
`delete()`	`() => void`	—	Delete segment file from disk

Static methods:

Method	Signature	Returns	Description
`MapletSegment.load(filePath)`	`(string) => MapletSegment`	Loaded segment	Load from disk (zero-copy via Buffer)
`MapletSegment.merge(segA, segB, outputDir)`	`(MapletSegment, MapletSegment, string) => MapletSegment`	Merged segment	Merge two segments with last-write-wins

Error conditions:

insert on sealed segment: throws Error('Cannot insert into sealed segment')
seal on already sealed segment: throws Error('Segment already sealed')
MapletSegment.load on invalid magic: throws Error('Invalid segment file: {path}')
MapletSegment.load on unsupported version: throws Error('Unsupported segment version: {version}')
Robin Hood insertion overflow: throws Error('Maplet table is full')

Binary format constants:

Constant	Value	Description
`SEGMENT_MAGIC`	`0x53544D4C`	'STML' magic bytes
`SEGMENT_VERSION`	`1`	Format version
`HEADER_SIZE`	`64`	Bytes
`SLOT_SIZE`	`16`	Bytes per maplet slot

Maplet Compaction Manager

`MapletCompactionManager`

new MapletCompactionManager(options?: {
  dataDir?: string,
  config?: Object,
  onCompactionComplete?: (compactedSegments: MapletSegment[], newSegment: MapletSegment) => void
}): MapletCompactionManager

Size-tiered compaction manager for Maplet segments. Manages a level-based tier (Level 0 = active mutable segment; Level 1+ = sealed immutable segments). When a level exceeds maxSegmentsPerLevel, merges the oldest segments into the next level.

Name	Type	Required	Default	Description
`dataDir`	`string`	no	`'./maplet-data'`	Segment storage directory
`config`	`Object`	no	`DEFAULT_CONFIG`	Compaction configuration
`onCompactionComplete`	`Function`	no	`null`	Callback after successful compaction

DEFAULT_CONFIG (MapletCompactionConfig):

Key	Default	Description
`levelSizeThresholds`	`[4MB, 40MB, 400MB, 4GB, 40GB]`	Size thresholds per level
`maxSegmentsPerLevel`	`[4, 10, 10, 10, 10]`	Max segments per level before compaction
`maxSegmentSize`	`4 * 1024 * 1024` (4MB)	Max single segment size
`maxEntriesPerSegment`	`10000`	Entries per segment
`autoCompaction`	`true`	Enable background compaction
`compactionIntervalMs`	`60000` (1min)	Background compaction interval
`compactionStrategy`	`'size-tiered'`	Strategy name
`mergeFactor`	`4`	Number of segments to merge at once

Methods:

Method	Signature	Returns	Description
`initializeSync()`	`() => void`	—	Synchronous initialization. Loads existing segments, creates active segment, starts compaction
`initialize()`	`() => Promise<void>`	Promise	Async wrapper around `initializeSync()`
`write(key, frameData, lsn)`	`(string, Buffer, number) => { segmentId, offset, lsn }`	Write result	Write to active segment; auto-seals and creates new if full
`read(key)`	`(string) => { frameData, level, segmentId, lsn } \| null`	Result or null	O(1) lookup via global index, fallback to full scan
`checkCompaction()`	`() => void`	—	Checks each level and triggers compaction if needed
`forceCompaction()`	`() => Promise<void>`	Promise	Compacts all levels until under threshold
`stopBackgroundCompaction()`	`() => void`	—	Stops the periodic compaction timer
`enumerateAll()`	`() => Generator`	Generator	Yields `{ key, frameData, segmentId, level, lsn }` for all entries
`getStats()`	`() => Object`	Stats	Full statistics including per-level segment counts, global index size, compaction history
`close()`	`() => Promise<void>`	Promise	Seals active segment, stops compaction, marks uninitialized

Error conditions:

write/read before initialization: throws Error('Compaction manager not initialized')
Critical segment load errors (invalid magic/version): throws Error('Failed to load segment {file}: {message}') — prevents store open
Non-critical load errors: logged to console, continues loading

Checkpoint Manager

`CheckpointManager`

new CheckpointManager(options?: {
  store: WALStore,
  checkpointEvery?: number,
  onCheckpoint?: (lsn: number, state: PipelineState) => void
}): CheckpointManager

Handles periodic checkpointing of pipeline state. Triggers after checkpointEvery operations.

Name	Type	Required	Default	Description
`store`	`WALStore`	yes	—	WAL store to persist checkpoints
`checkpointEvery`	`number`	no	`100`	Operations between checkpoints
`onCheckpoint`	`Function`	no	—	Callback after each checkpoint

Methods:

Method	Signature	Returns	Description
`registerWrapper(wrapperId, type, getState)`	`(string, string, () => Object) => void`	—	Register a wrapper for state capture during checkpoint
`unregisterWrapper(wrapperId)`	`(string) => void`	—	Remove a wrapper from state capture
`trackFrame(frameId, lsn, frame, wrapperId)`	`(string, number, Frame, string) => void`	—	Track an in-flight frame
`completeFrame(frameId)`	`(string) => void`	—	Mark frame as completed
`trackTimer(timerId, fireAt, wrapperId, context)`	`(string, number, string, Object) => void`	—	Track a pending timer
`removeTimer(timerId)`	`(string) => void`	—	Remove a timer from tracking
`recordOperation()`	`() => void`	—	Increment counter; triggers checkpoint if threshold reached (via `setImmediate`)
`checkpoint(externalState?)`	`(Object?) => number`	LSN	Force immediate synchronous checkpoint. Captures all wrapper states, in-flight frames, and pending timers. Thread-safe via `checkpointInProgress` flag
`truncateBeforeCheckpoint()`	`() => void`	—	Call `store.truncate(lastCheckpointLsn)`
`getStateSnapshot()`	`() => { wrappers, inFlight, pendingTimers }`	Snapshot	Get current state without writing to store

Checkpoint structure written to WAL:

{
  state: externalState,       // optional external state payload
  wrappers: { [id]: { type, state } },
  inFlight: [{ frameId, lsn, frame, wrapperId, enqueuedAt }],
  pendingTimers: [{ timerId, fireAt, wrapperId, context }]
}

Error conditions: Checkpoint is re-entrancy safe — if checkpointInProgress is true, returns lastCheckpointLsn without writing.

Recovery Manager

`RecoveryManager`

new RecoveryManager(options?: {
  store: WALStore,
  blobStore?: Object
}): RecoveryManager

Handles pipeline recovery from WAL and checkpoints after a crash.

Name	Type	Required	Default	Description
`store`	`WALStore`	yes	—	WAL store to read recovery data from
`blobStore`	`Object`	no	—	Blob store for restoring binary frame data

Methods:

Method	Signature	Returns	Description
`recover()`	`() => RecoveryResult`	RecoveryResult	Synchronous full recovery. Loads checkpoint, scans uncommitted entries, identifies frames to replay, adjusts timers for crash duration
`applyWrapperStates(recoveryResult, wrappers)`	`(RecoveryResult, Map<string, DurableWrapper>) => void`	—	Calls `setState(state)` on each wrapper with its recovered state
`restoreTimers(timersToRestore, timerCallback)`	`(Array, (timer) => void) => { immediate, scheduled }`	Counts	Schedules timers: fire-immediately if `fireImmediately`, otherwise `setTimeout` with adjusted delay

RecoveryResult shape:

Field	Type	Description
`recovered`	`boolean`	Whether recovery was performed (false if no checkpoint)
`checkpoint`	`PipelineState\|null`	Loaded checkpoint data
`wrapperStates`	`Object<string, WrapperState>`	Recovered wrapper states, merged with state changes from WAL
`framesToReplay`	`Array`	Frames that need re-enqueueing. Each: `{ frameId, frame, lsn, wrapperId, enqueuedAt }`
`timersToRestore`	`Array`	Timers to restore. Each: `{ timerId, wrapperId, context, originalFireAt, adjustedFireAt, fireImmediately }`
`crashDuration`	`number`	Duration of crash in ms (now - checkpoint time)

Recovery algorithm:

Load most recent checkpoint
Extract wrapper states and in-flight frames from checkpoint
Scan WAL for entries after checkpoint
Build sets of processed/failed frames, timer events, state changes
Frames to replay = (checkpoint in-flight + WAL enqueued) - (processed + failed)
Timers = checkpoint timers + WAL scheduled timers - fired/cancelled, with fireAt adjusted by crash duration
Wrapper states = checkpoint states merged with WAL state change deltas

Durable Pipeline

`createDurablePipeline(options)`

createDurablePipeline(options: {
  store: WALStore,
  provider: IFrameDuplex,
  wrappers?: WrapperConfig[],
  checkpointEvery?: number,
  recordResponses?: boolean,
  onReady?: (err: Error | null, recoveryResult: RecoveryResult) => void
}): DurablePipeline

Creates a durable pipeline — a chain of push-stream wrappers around a provider, with automatic WAL logging, checkpointing, and crash recovery. Pure push-stream: returns immediately, recovery runs asynchronously.

Name	Type	Required	Default	Description
`store`	`WALStore`	yes	—	WAL store (must have synchronous methods)
`provider`	`IFrameDuplex`	yes	—	Innermost duplex (adapter, etc.)
`wrappers`	`WrapperConfig[]`	no	`[]`	Wrapper chain configs, applied from innermost to outermost
`checkpointEvery`	`number`	no	`100`	Operations between checkpoints
`recordResponses`	`boolean`	no	`false`	If true, serialize response frames into WAL `FRAME_PROCESSED` entries
`onReady`	`Function`	no	—	Called after recovery completes

WrapperConfig:

Field	Type	Description
`type`	`'queue' \| 'rateLimit' \| 'circuitBreaker' \| 'retry'`	Wrapper type
`id`	`string`	Unique wrapper ID (auto-generated if missing)
`options`	`Object`	Wrapper-specific options
`limits`	`Object`	Rate-limit configuration (only for `rateLimit` type)

Return value: DurablePipeline object:

Property/Method	Type	Description
`duplex`	`IFrameDuplex`	The outermost wrapper duplex (wrapped with durability interceptor)
`checkpointManager`	`CheckpointManager`	The checkpoint manager instance
`recoveryManager`	`RecoveryManager`	The recovery manager instance
`wrappers`	`Map<string, { wrapper, type }>`	Map of wrapper ID to wrapper instance and type
`store`	`WALStore`	The WAL store
`checkpoint(callback?)`	`(fn?) => number\|void`	Force checkpoint. Sync return if no callback
`close(callback?)`	`(fn?) => true\|void`	Graceful shutdown: checkpoint then close store
`getStats()`	`() => { operationCount, inFlightFrames, pendingTimers, lastCheckpointLsn }`	Pipeline stats

Frame protocol: The durability interceptor wraps the pipeline with two hooks:

onRequest: On TEXT frames, writes a FRAME_ENQUEUED entry to WAL and tracks the frame in the checkpoint manager
onResponse: Intercepts durability META frames (from wrapper state changes, timers, etc.), writes them to WAL, and filters them from user output. On TEXT response frames, writes FRAME_PROCESSED. On ERROR response frames, writes FRAME_FAILED

Error conditions:

Unknown wrapper type: throws Error('Unknown wrapper type: {type}')
Store initialization failure: passed to onReady callback
checkpoint with callback can pass error via callback
close with callback can pass error via callback

Durable System

`createDurableSystem(innerDuplex, options?)`

createDurableSystem(innerDuplex: IFrameDuplex, options?: {
  dataDir?: string,
  autoRecover?: boolean,
  checkpointIntervalMs?: number,
  compaction?: boolean,
  onRecover?: (state: Object) => void,
  onPersist?: (entry: Object) => void
}): DurableSystem

High-level wrapper that automatically persists all state changes from a Polyweave system to a WAL. Intercepts durability META frames emitted by wrappers (dialog, voice, streaming, observational), persists them, and supports transparent crash recovery. Creates its own FileWALStore internally.

Name	Type	Required	Default	Description
`innerDuplex`	`IFrameDuplex`	yes	—	The Polyweave duplex to wrap
`dataDir`	`string`	no	`'./wal-data'`	WAL data directory
`autoRecover`	`boolean`	no	`true`	Auto-restore state on startup
`checkpointIntervalMs`	`number`	no	`30000`	Periodic checkpoint interval
`compaction`	`boolean`	no	`true`	Enable background compaction
`onRecover`	`Function`	no	`() => {}`	Called with recovered state
`onPersist`	`Function`	no	`() => {}`	Called with each persisted entry

Return value: DurableSystem object:

Property/Method	Type	Description
`sink`	`Sink`	Sink that intercepts frames
`source`	`Source`	Source from inner duplex
`walStore`	`FileWALStore`	The internal WAL store
`checkpointManager`	`CheckpointManager`	Checkpoint manager
`recoveryManager`	`RecoveryManager`	Recovery manager
`getStats()`	`() => Object`	Stats including WAL stats, wrapper count, recovery status
`checkpoint()`	`() => number`	Force checkpoint of all tracked wrapper states
`recover()`	`() => RecoveryResult`	Manual recovery
`close()`	`() => void`	Final checkpoint, closes WAL store and inner duplex

Delta application: The onResponse hook applies state deltas from META frames. Supported delta types:

STATE_SET — state[key] = value
ARRAY_PUSH — push item to state[arrayKey] array
ARRAY_REMOVE — remove item by id from state[arrayKey]
OBJECT_SET_PROPERTY — set property on state[objectKey]
INCREMENT — state[key] += amount
FULL_SNAPSHOT — replace entire state

`createDurableConversation(adapter, options?)`

createDurableConversation(adapter: IFrameDuplex, options?: {
  dataDir?: string,
  conversationId?: string,
  dialogOptions?: Object
}): DurableSystem

Convenience function for the most common use case: wraps an adapter in a createDialogWrapper and then wraps that in createDurableSystem. Uses require('@polyweave/dialog') at runtime.

Name	Type	Required	Default	Description
`adapter`	`IFrameDuplex`	yes	—	AI adapter
`dataDir`	`string`	no	`'./conversations'`	Base data directory
`conversationId`	`string`	no	`'conv-{Date.now()}'`	Unique conversation ID
`dialogOptions`	`Object`	no	`{}`	Options passed to `createDialogWrapper`

Durable Router

`createDurableRouter(options)`

createDurableRouter(options: {
  dataDir?: string,
  factory: (streamId: string) => IFrameDuplex,
  compactionInterval?: number,
  idleTimeout?: number,
  silent?: boolean
}): DurableRouter

Routes frames to isolated WAL-backed duplexes by streamId. Each stream gets its own FileWALStore, durability wrapper, and auto-recovery. Idle instances are unloaded after idleTimeout seconds.

Name	Type	Required	Default	Description
`dataDir`	`string`	no	`'./wal'`	Root data directory
`factory`	`(streamId) => IFrameDuplex`	yes	—	Creates the base duplex for each `streamId`. Must return synchronously
`compactionInterval`	`number`	no	`60`	Seconds between compactions per instance
`idleTimeout`	`number`	no	`300`	Unload instance after N seconds of inactivity
`silent`	`boolean`	no	`false`	Suppress console output

Return value: DurableRouter object:

Property/Method	Type	Description
`sink`	`Sink`	Accepts frames; routes by `frame.streamId`
`source`	`Source`	Joined output from all instances (adds `streamId` to frames that lack it)
`getStats()`	`() => { activeInstances, streamIds, schedulerStats }`	Router stats
`hasInstance(streamId)`	`(string) => boolean`	Whether an instance is loaded
`unloadInstance(streamId)`	`(string) => void`	Manually unload an idle instance
`close()`	`() => void`	Unloads all instances, stops scheduler

Idle management: Uses an internal LAWN-like scheduler. Each instance gets a compaction task (jittered within compactionInterval) and an idle timeout check. When idle time exceeds idleTimeout, the instance is checkpointed and unloaded (directory and files remain for future recovery).

Error conditions:

Missing factory: throws Error('factory function required')
streamId exceeding 128 characters: throws Error('streamId too long: max 128, got {length}')
Factory returns null: warns and skips instance creation
Factory throws: warns with error message and returns null (instance not created)
Invalid frame (missing streamId): warns and returns without routing

Snapshot and Persistence Helpers

`createTreeSnapshotClient(treeStore, mode?)`

createTreeSnapshotClient(treeStore: Store, mode?: string): { snapshot: (callback: (err: Error | null, state: Object) => void) => void }

Creates a snapshot client for tree stores. Sends TREE_SNAPSHOT frames to the tree store and returns the snapshot via callback.

Name	Type	Required	Default	Description
`treeStore`	`Store`	yes	—	Store with `source`/`sink` (must support `TREE_SNAPSHOT` frames)
`mode`	`string`	no	`'full'`	Snapshot mode passed in frame content

Frame protocol: Sends TREE_SNAPSHOT with content: { mode, streamId }. Pipes treeStore.source to a frame sink that listens for TREE_SNAPSHOT response frames and resolves pending callbacks by streamId.

`createFactSnapshotClient(factStore)`

createFactSnapshotClient(factStore: Store): { snapshot: (callback: (err: Error | null, content: Object) => void) => void }

Same pattern as tree snapshot but for fact stores. Sends and listens for FACT_SNAPSHOT frames.

`normalizeTreeCheckpoint(state)`

normalizeTreeCheckpoint(state: Map-based state | Array-based state): { nodes: Array, trees: Object } | null

Normalizes tree store state from Map-based internal representation to JSON-serializable format for persistence. Returns null if state is not valid.

`restoreStoreFromCheckpoint(options)`

restoreStoreFromCheckpoint(options: {
  walStore: WALStore,
  store: Store,
  wrapperId: string,
  wrapperType: string,
  normalizeState?: (state: Object) => Object | null,
  onSkip?: (reason: string) => void,
  onDone?: (err: Error | null, restored: boolean) => void
}): void

Loads a checkpoint from WAL, normalizes the state, and injects a META restore frame into the target store.

`checkpointTree(walStore, snapshotClient, meta, callback?)`

checkpointTree(walStore: WALStore, snapshotClient: Object, meta: { wrapperId: string, wrapperType: string }, callback?: (err: Error | null, lsn: number) => void): void

Takes a tree snapshot and writes it as a WAL checkpoint.

`checkpointFacts(walStore, snapshotClient, meta, callback?)`

checkpointFacts(walStore: WALStore, snapshotClient: Object, meta: Object, callback?: (err: Error | null, lsn: number) => void): void

Takes a fact snapshot and writes it as a WAL checkpoint.

`rebuildTextIndexFromTree(options)`

rebuildTextIndexFromTree(options: {
  treeStore: Store,
  textStore: Store,
  onNode?: (node: Object) => void
}): (callback?: (err: Error | null, leaves: Array) => void) => void

Returns a rebuild function. When called, snapshots the tree store, filters leaf nodes whose data.kind === 'paragraph' with docId and text, writes TEXT_INDEX frames to textStore, and returns the leaf array.

Frame protocol: Writes TEXT_INDEX frames with content: { docId, text } to textStore.sink.

Tests

npm run test:unit       # fast unit tests
npm run test:property   # property-based tests
npm test                # alias for test:unit

Tests cover WAL entry creation and serialization, MemoryWALStore CRUD and checkpoint lifecycle, FileWALStore initialization with all encoding formats, JSONL and binary WAL scanning and flushing, binary field cache, blob storage, MapletSegment insertion/query/enumeration/seal/merge/load with Robin Hood hashing, MapletCompactionManager write/read/compaction/stats, CheckpointManager wrapper registration and triggered checkpoints, RecoveryManager full recovery with crash-duration timer adjustment, durable pipeline creation with wrapper chains and recovery, durable system auto-recovery and delta application, durable router stream-id routing and idle unloading, snapshot client creation, and tree/fact checkpoint/restore cycle.

Requirements

Node.js 22 or later. ESM only. Depends on @polyweave/core.

Development Dependencies

ID	Version
@rigor/core	*
fast-check	^3.23.2

Peer Dependencies

ID	Version
@polyweave/core	*
@polyweave/dialog	*
@polyweave/openai	*

polyweave wal durability crash-recovery checkpoint maplet sstable

Details

npm

2026-07-11 15:23:55 +00:00

456

John Dvorak

SEE LICENSE IN LICENSE

latest

39 KiB

Assets (1)

durability-1.0.9.tgz 39 KiB

Versions (6) View all

1.0.9

2026-07-11

1.0.6

2026-07-10

1.0.5

2026-07-10

1.0.2

2026-07-10

1.0.1

2026-07-10

@polyweave/durability (1.0.9)

Installation

About this package

@polyweave/durability

Synopsis

Install

Why

API

WAL Types and Utilities

WALEntryType

CommitStatus

createWALEntry(type, payload, options?)

createFrameEnqueuedEntry(frame, wrapperId)

createFrameProcessedEntry(frameId, wrapperId, responseFrame?)

createFrameFailedEntry(frameId, wrapperId, error)

createFrameRetryingEntry(frameId, wrapperId, retryCount, nextRetryAt)

createTimerScheduledEntry(timerId, fireAt, wrapperId, context)

createTimerFiredEntry(timerId)

createTimerCancelledEntry(timerId)

createStateChangeEntry(wrapperId, wrapperType, stateDelta, options?)

createCheckpointEntry(pipelineState)

serializeFrame(frame)

deserializeFrame(serialized, blobStore?)

parseWALEntry(line)

stringifyWALEntry(entry)

WAL Store Interface

validateWALStore(store)

Memory WAL Store

MemoryWALStore

File WAL Store

FileWALStore

FileWALStoreLegacy

Maplet Segment

MapletSegment

Maplet Compaction Manager

MapletCompactionManager

Checkpoint Manager

CheckpointManager

Recovery Manager

RecoveryManager

Durable Pipeline

createDurablePipeline(options)

Durable System

createDurableSystem(innerDuplex, options?)

createDurableConversation(adapter, options?)

Durable Router

createDurableRouter(options)

Snapshot and Persistence Helpers

createTreeSnapshotClient(treeStore, mode?)

createFactSnapshotClient(factStore)

normalizeTreeCheckpoint(state)

restoreStoreFromCheckpoint(options)

checkpointTree(walStore, snapshotClient, meta, callback?)

checkpointFacts(walStore, snapshotClient, meta, callback?)

rebuildTextIndexFromTree(options)

Tests

Requirements

Dependencies

Keywords

`WALEntryType`

`CommitStatus`

`createWALEntry(type, payload, options?)`

`createFrameEnqueuedEntry(frame, wrapperId)`

`createFrameProcessedEntry(frameId, wrapperId, responseFrame?)`

`createFrameFailedEntry(frameId, wrapperId, error)`

`createFrameRetryingEntry(frameId, wrapperId, retryCount, nextRetryAt)`

`createTimerScheduledEntry(timerId, fireAt, wrapperId, context)`

`createTimerFiredEntry(timerId)`

`createTimerCancelledEntry(timerId)`

`createStateChangeEntry(wrapperId, wrapperType, stateDelta, options?)`

`createCheckpointEntry(pipelineState)`

`serializeFrame(frame)`

`deserializeFrame(serialized, blobStore?)`

`parseWALEntry(line)`

`stringifyWALEntry(entry)`

`validateWALStore(store)`

`MemoryWALStore`

`FileWALStore`

`FileWALStoreLegacy`

`MapletSegment`

`MapletCompactionManager`

`CheckpointManager`

`RecoveryManager`

`createDurablePipeline(options)`

`createDurableSystem(innerDuplex, options?)`

`createDurableConversation(adapter, options?)`

`createDurableRouter(options)`

`createTreeSnapshotClient(treeStore, mode?)`

`createFactSnapshotClient(factStore)`

`normalizeTreeCheckpoint(state)`

`restoreStoreFromCheckpoint(options)`

`checkpointTree(walStore, snapshotClient, meta, callback?)`

`checkpointFacts(walStore, snapshotClient, meta, callback?)`

`rebuildTextIndexFromTree(options)`