Run and scale the Flowcore Data Pump

@flowcore/data-pump is the lower-level TypeScript runtime for continuously reading retained events from Flowcore. It is the engine beneath virtual processing patterns: it walks event history, buffers events, hands them to application code, and persists a position from which processing can resume.

Use the package directly when you need control over fetching, acknowledgment, runtime state, notification transport, or horizontal worker distribution. If your application already uses @flowcore/pathways, prefer its higher-level pump integration unless you need an API exposed only by the Data Pump.

This guide reflects the behavior of @flowcore/data-pump 0.22.x. Some important runtime semantics—especially cluster mode, pulse reporting, and terminal failure handling—are implemented in the package but are not fully described by its README.

Understand the processing model

A pump has two positions. The fetch position moves as pages are read from Flowcore, while the durable processing position moves only when buffered events are acknowledged or removed as terminal failures. Keeping those positions separate lets the pump fetch ahead without claiming that application work has completed.

Events move through this lifecycle:

Flowcore history -> local buffer -> reserved batch -> handler
                                      |              |
                                      |              +-> acknowledge and advance
                                      |
                                      +-> timeout -> reopen -> reserve again

The buffer is bounded. Its default capacity is 1,000 events, and fetching pauses when it is full. Fetching resumes after enough entries leave the buffer to restore approximately ten percent of headroom. This hysteresis prevents a slow consumer from causing an unbounded in-memory queue.

Delivery is at least once. A process can commit an external side effect and then stop before acknowledging the event. On restart, the persisted position can cause that event to be delivered again. Durable idempotency is therefore part of the consumer contract, not an optional optimization.

Create a single-instance pump

Install the package with your normal Node.js-compatible package manager:

npm install @flowcore/data-pump

A production pump needs Flowcore authentication, a data-source selection, a durable state manager, and usually a processor. The current API key format contains its own key ID, so new code should pass only apiKey; the separate apiKeyId field is deprecated.

import { FlowcoreDataPump } from "@flowcore/data-pump"
 
const pump = FlowcoreDataPump.create({
  auth: {
    apiKey: process.env.FLOWCORE_API_KEY!,
  },
  dataSource: {
    tenant: "acme",
    dataCore: "commerce",
    flowType: "order.0",
    eventTypes: ["order.placed.0", "order.cancelled.0"],
  },
  stateManager: {
    getState: () => loadPumpState("orders-v1"),
    setState: (state) => savePumpState("orders-v1", state),
  },
  notifier: { type: "websocket" },
  bufferSize: 500,
  maxRedeliveryCount: 5,
  achknowledgeTimeoutMs: 30_000,
  processor: {
    concurrency: 10,
    handler: async (events) => {
      for (const event of events) {
        await projectEventIdempotently(event)
      }
    },
    failedHandler: async (events) => {
      await recordTerminalFailures(events)
    },
  },
  logger,
})
 
await pump.start((error) => {
  if (error) logger.error("Data Pump stopped with an error", { error })
})

The public option is currently spelled achknowledgeTimeoutMs; use that spelling until the package introduces a compatible replacement.

Passing a callback to start() starts the fetch loop in the background and enables its self-healing restart path. Transient fetch-loop failures are retried with exponential delays from one second up to thirty seconds. Calling await pump.start() without a callback instead waits for the long-running fetch loop and lets a fetch error reject that promise. The callback form is therefore the safer production pattern for a standalone pump.

The processor.concurrency name can be misleading. In 0.22.x it is the number of events reserved and passed to one handler invocation. It does not create that many simultaneous handler invocations. Treat it as batch size and add intentional parallelism inside the handler only when event ordering and downstream capacity allow it.

A batch is acknowledged only after the complete handler resolves. If the first events commit and a later event throws, the batch can be delivered again. Each event's side effect must be idempotent even when the handler iterates sequentially.

Persist a conservative checkpoint

The state manager reads and writes a timeBucket plus an optional eventId:

interface FlowcoreDataPumpState {
  timeBucket: string
  eventId?: string
}

Time buckets use the fourteen-digit UTC form yyyyMMddHH0000. The event ID refines the position within that hour.

Persist state in a database for production. When events remain buffered, the pump saves a conservative frontier based on the earliest remaining event. When the buffer becomes empty, it can save the last removed event. This design favors replay over skipping work, but it also means the saved event ID should not be interpreted as a perfect audit record of the last handler that completed.

The state write and your business transaction are separate operations. A robust projection stores the event ID in the same transaction as the projected data:

await db.transaction(async (tx) => {
  const claim = await tx.query(
    `INSERT INTO processed_events (consumer, event_id)
     VALUES ($1, $2)
     ON CONFLICT DO NOTHING
     RETURNING event_id`,
    ["orders-v1", event.eventId],
  )
 
  if (claim.rowCount === 0) return
 
  await tx.query(
    `INSERT INTO orders (id, status, source_event_id)
     VALUES ($1, $2, $3)
     ON CONFLICT (id) DO UPDATE
       SET status = EXCLUDED.status,
           source_event_id = EXCLUDED.source_event_id`,
    [event.payload.orderId, event.payload.status, event.eventId],
  )
})

Returning null from getState() starts at the current time, not at the beginning of retained history. Specifically, the pump uses the current hour and a time-based event ID representing now, so earlier events in the current hour are not part of that new live subscription. Supply an explicit historical state when building a projection or backfilling existing data.

Process history and bounded replays

To replay from a known position, return it from getState() before starting the pump:

stateManager: {
  getState: () => ({
    timeBucket: "20260101000000",
    eventId: undefined,
  }),
  setState: persistState,
},
stopAt: new Date("2026-02-01T00:00:00Z"),

The pump walks available time buckets, paginates through each bucket, and moves into live notification mode after catching up. stopAt places an event-ID boundary on the fetch and waits for the active buffer to drain before stopping.

You can reposition a running single-instance pump with restart():

pump.restart({
  timeBucket: "20260115120000",
  eventId: "optional-time-uuid",
})

The bucket must be exactly fourteen digits. Restarting clears the current in-memory buffer and refreshes the available bucket list before continuing from the closest bucket.

In 0.22.x, do not rely on the optional second restart(state, stopAt) argument to change an existing stop boundary: the option changes, but the internal stop state is not rebuilt during restart. Stop the old pump and create a new instance when the replay boundary must change.

Choose automatic or manual acknowledgment

Providing processor selects automatic, or push, mode. The pump reserves a batch, invokes the handler, and acknowledges every event after the handler resolves.

Omitting processor selects manual, or pull, mode. Your code calls reserve(), then explicitly acknowledges completed events:

const events = await pump.reserve(20)
 
for (const event of events) {
  try {
    await projectEventIdempotently(event)
    await pump.acknowledge([event.eventId])
  } catch (error) {
    logger.warn("Event will become eligible for redelivery", {
      eventId: event.eventId,
      error,
    })
    // Leave the reservation unresolved. It reopens after the acknowledgment timeout.
  }
}

The distinction between an unresolved reservation and fail() is important. In the current implementation, fail(eventIds) removes those events from the buffer as terminal failures; it does not schedule them for retry. To retry in pull mode, leave the event unacknowledged and let the acknowledgment timeout reopen it.

maxRedeliveryCount counts redeliveries after the initial reservation. The default value of three therefore permits four total reservation attempts. Set it to -1 only when infinite redelivery is intentional and a poison event cannot block useful work indefinitely.

Per-event retries wait for the fixed acknowledgment timeout. They do not use exponential backoff. Exponential backoff is used for failed fetch loops, process loops, leader pumps, and cluster reconnections. If business retries require a different schedule, publish work into a durable retry workflow instead of treating the pump's reservation timer as a scheduler.

processor.failedHandler runs after a terminal fail() and when timeout-based redelivery exceeds the configured maximum. The separately registered onFinalyFailed() callback—whose spelling is also preserved by the current API—runs only for failures that exhaust timeout-based redelivery.

Select a live notification transport

Notifications wake a caught-up pump so that it can query Flowcore storage again. They are acceleration signals, not the durable event source. The checkpoint and retained event history allow the pump to recover from a missed notification.

WebSocket is the implicit default. Each live wait has a twenty-second safety timeout, so a broken or half-open connection eventually releases the fetch loop to poll storage and create a fresh connection. A WebSocket error also releases the wait rather than leaving the pump stuck.

NATS mode subscribes to stored-event notification subjects for the selected event types. The wait also has a twenty-second timeout and unsubscribes before the next cycle. Use it when the deployment already has reliable access to NATS.

Poller mode avoids a notification connection. In the 0.22.x implementation, the wake delay is calculated with Math.min(intervalMs, 1000), so values above one second still wake after one second. This differs from the option's minimum-interval comment and README example; account for the resulting API traffic until the behavior is corrected.

Scale processing with cluster mode

FlowcoreDataPumpCluster scales handler execution while preserving one logical fetcher and one shared checkpoint. Every replica registers with a coordinator and participates in lease election. The lease holder becomes leader, starts the underlying pump, and distributes reserved batches to workers. Followers do not fetch from Flowcore, but they can process batches sent by the leader.

Cluster mode requires a FlowcoreDataPumpCoordinator. The package defines the contract but does not ship a production implementation:

interface FlowcoreDataPumpCoordinator {
  acquireLease(instanceId: string, key: string, ttlMs: number): Promise<boolean>
  renewLease(instanceId: string, key: string, ttlMs: number): Promise<boolean>
  releaseLease(instanceId: string, key: string): Promise<void>
  register(instanceId: string, address: string): Promise<void>
  heartbeat(instanceId: string): Promise<void>
  unregister(instanceId: string): Promise<void>
  getInstances(staleThresholdMs: number): Promise<Array<{
    instanceId: string
    address: string
  }>>
}

Back the lease, registry, and pump state with shared durable storage. Lease acquisition must be atomic, renewal must succeed only for the current holder, and stale instances must disappear from discovery. The repository's integration suite demonstrates this contract with PostgreSQL.

Use NATS distribution

When notifier.type is nats, cluster mode also uses NATS request/reply to distribute event batches. All replicas join the data-pump-workers queue group, so one subscriber handles each request. The leader can also receive work because it starts the same worker subscription.

import { FlowcoreDataPumpCluster } from "@flowcore/data-pump"
 
const cluster = new FlowcoreDataPumpCluster({
  auth: { apiKey: process.env.FLOWCORE_API_KEY! },
  dataSource: {
    tenant: "acme",
    dataCore: "commerce",
    flowType: "order.0",
    eventTypes: ["order.placed.0", "order.cancelled.0"],
  },
  stateManager: sharedStateManager,
  coordinator: postgresCoordinator,
  notifier: {
    type: "nats",
    servers: [process.env.NATS_URL!],
  },
  clusterKey: "orders-projection-v1",
  workerConcurrency: 10,
  processor: {
    handler: async (events) => {
      for (const event of events) {
        await projectEventIdempotently(event)
      }
    },
    failedHandler: recordTerminalFailures,
  },
  logger,
})
 
await cluster.start()

A NATS request waits up to thirty seconds for a worker reply. If distribution fails, the leader falls back to running the user handler locally. That preserves availability, but a lost reply can mean the remote worker committed before local fallback begins. Idempotency remains mandatory.

clusterKey scopes the NATS distribution subject. In 0.22.x it does not scope the internal leader lease key, which is fixed to flowcore-data-pump-leader. If unrelated logical clusters share one coordinator backend, namespace the coordinator's lease storage or wrap each coordinator so their lease keys cannot collide.

Use WebSocket distribution

All non-NATS cluster configurations use the WebSocket mesh. Each instance must expose a WebSocket server, pass accepted connections to cluster.handleConnection(), and register an advertisedAddress reachable from every peer.

The leader discovers live instances through the coordinator, connects to workers, and distributes whole batches round-robin. WebSocket connections use ping and pong frames to detect dead peers. A delivery waits up to thirty seconds for acknowledgment; disconnects and timeouts reject the delivery so the underlying pump can make it eligible for redelivery.

The default lease lasts thirty seconds and is renewed every ten seconds. Heartbeats run every five seconds, and WebSocket discovery treats an instance as stale after thirty seconds. Tune lease and heartbeat values together: a renewal interval must leave enough room for a temporary database pause without allowing two leaders to operate for an extended period.

When no remote worker is available, the leader processes the batch locally. Cluster mode therefore continues to work as a one-replica deployment, provided the leader has the user handler and access to its dependencies.

Cluster mode supports the processor model only. It does not expose the single pump's manual reserve(), acknowledge(), fail(), or restart() methods.

Use dedicated-cluster optimizations deliberately

By default, the data-source configuration contains resource names. The pump translates and caches tenant, data-core, flow-type, and event-type IDs before querying events.

On a dedicated Flowcore installation, directMode sends SDK commands through the direct execution path. noTranslation skips lookup calls, but then every configured value—including tenant—must already be the corresponding resource ID:

dataSource: {
  tenant: process.env.FLOWCORE_TENANT_ID!,
  dataCore: process.env.FLOWCORE_DATA_CORE_ID!,
  flowType: process.env.FLOWCORE_FLOW_TYPE_ID!,
  eventTypes: process.env.FLOWCORE_EVENT_TYPE_IDS!.split(","),
},
noTranslation: true,
directMode: true,

Do not copy IDs between environments. Resolve and inject the IDs belonging to each deployment.

includeSensitiveData defaults to false and is passed to the Flowcore event-list request. Enabling it requires both an explicit application decision and an identity with the relevant permissions. The pump does not add a second payload-redaction layer, so logs and failed-event storage must still avoid exposing protected values.

Observe progress with metrics and pulses

The package exports a Prometheus registry. Expose it through your service using the asynchronous metrics() result:

import { dataPumpPromRegistry } from "@flowcore/data-pump"
 
Bun.serve({
  port: 9090,
  async fetch(request) {
    if (new URL(request.url).pathname !== "/metrics") {
      return new Response("Not found", { status: 404 })
    }
 
    return new Response(await dataPumpPromRegistry.metrics(), {
      headers: { "content-type": dataPumpPromRegistry.contentType },
    })
  },
})

The single-pump metrics report buffered and reserved events, payload bytes in the buffer, acknowledged and failed events, pulled bytes, and SDK command counts. Event metrics use tenant, data core, flow type, and event type labels. The SDK command counter uses only a command label.

Cluster mode adds leader status, active workers, distributed events, worker acknowledgments, and worker failures. These cluster metrics are process-local and do not carry tenant or event-type labels, so separate services or registries when one process hosts unrelated pumps.

The optional pulse configuration reports a richer status snapshot to a Flowcore control-plane endpoint:

pulse: {
  url: process.env.FLOWCORE_CONTROL_PLANE_URL!,
  pathwayId: process.env.FLOWCORE_PATHWAY_ID!,
  sourceId: process.env.PUMP_SOURCE_ID,
  intervalMs: 30_000,
  successLogLevel: "debug",
  failureLogLevel: "warn",
},

A pulse includes the current bucket and event ID, live status, buffer depth and reserved count, payload bytes, cumulative pulled, acknowledged, and failed counts, plus process uptime. The first emission is randomly staggered across the interval to avoid every replica reporting at once. Pulse failures are logged but do not stop event processing.

Metrics and pulses complement each other. Metrics support local scraping and alerting; pulses communicate pump state to the Flowcore control plane. Neither proves that a projection is correct, so also measure the freshness and integrity of the application state produced by the handler.

Shut down without hiding replay

stop() aborts live waiting, stops pulse emission, clears the local buffer, and marks the pump as stopped. It does not drain active work before clearing the buffer. On the next start, the durable checkpoint should cause unfinished events to be fetched again, but completed side effects can also be replayed.

Handle termination at the application boundary and stop accepting unrelated work before stopping the pump:

process.on("SIGTERM", () => {
  pump.stop()
})

For cluster mode, await cluster.stop(). It stops leader duties, closes worker connections, drains the NATS distribution connection, releases the lease, and unregisters the instance:

process.on("SIGTERM", async () => {
  await cluster.stop()
  await db.end()
  process.exit(0)
})

A clean shutdown reduces recovery delay, but it does not change the at-least-once contract. Keep event-ID idempotency in place across deployments, leader changes, worker timeouts, manual restarts, and bounded replays.

Continue with Build reliable handlers and read models for transaction and projection patterns, and Operate Data Pathways in production for progress-aware monitoring and recovery.