Units and State

A pipeline turns each input into units at its store, turns each unit into items for its sender, and remembers what it needs between inputs in state. This page describes those three, and what a step can and cannot reach. For the methods a step uses, see Steps.

Units

A unit is one row that reaches a pipeline's .store(). Steps before the store may split an input into many rows; each of them becomes a unit. Anything before the store is transient: a unit exists from the moment the store commits it, and the sending side is answered only after that commit.

pipeline("lab-batch")
  .source(webhookSource({ port: env("LAB_PORT") }))
  .step((files) => files.flatMap(splitBatch).withAttrs((m) => ({ msg_id: controlIdOf(m) })))
  .store(postgresStore({ codec: "text" }), { index: { msg_id: "btree" } })
  .step(hl7v2Parser())
  .step(labToFhir)
  .step(fhirItems())
  .sender(toAidbox);

One HTTP request holding a batch of lab messages becomes one unit per message. splitBatch and controlIdOf stand for functions of your own: the first splits the request body into raw messages, the second reads a raw message's MSH-10. The rows stay strings up to the store, which keeps each message as text for hl7v2Parser() to parse after it.

• A unit is identified by its id alone. One sequence numbers the units of every pipeline, so an id names one unit and, through it, its pipeline.
• Every arrival is a new unit. Nothing deduplicates at the store: an MLLP message resent after a lost ACK becomes a second unit. For Aidbox that is harmless, because skipUnchanged() skips resources that have not changed.
• The store keeps what the last step before it produced, in the codec the store declares: text, json or bytes. Put .store() right after .source() to keep the raw input, which a Retry then processes again from the start. The store keeps every value; there is no retention yet.
• A unit skipped before the store has no stored value, and cannot be retried.

A unit carries:

Items: fan-out after the store

Units are created only at the store. A flatMap after it splits a unit's row into items, which stay the unit's. When the steps after the store finish, the engine writes each item into the pipeline's sender queue, in the order the Batch holds them, and the unit becomes processed.

• The unit is the unit of retry. If a step fails for one item, the whole unit fails, with error_step naming that step. A Retry runs the unit again from the store, for all its items. Only the sender retries per item.
• An item's key, set with withMeta or by fhirItems(), is its identity at the destination, such as Patient/123. The sender never holds two versions of one key at once, and sends them in the order of their units' order_at, then the order the units were created in.
• Skipped items are counted on the unit, under attrs._skipped, as { "unchanged": 12 }. A unit whose items were all skipped or filtered closes sent at once, with nothing to send.
• A finished item leaves a record for a retention window. Every item that leaves the sender queue keeps a summary record: its key, its resource type and id, how it finished and why, and when. Only an item the sender rejected keeps its body as well; a delivered, dropped or deleted one does not. The record is purged after INTERBOX_DONE_ITEMS_RETENTION_DAYS days, 7 by default, from one hour up to 36500, or never at 0 (see Operating Pipelines). The Unit page lists the items still waiting, each with its tries, its next try and its last error, and its FHIR resources tab lists both the waiting and the finished items carrying a FHIR resource, each finished one with how and when it finished. A rejected item is also named in its unit's error message, such as Patient/123: ….
• A unit's items are held in memory for the take, because they are written with the take's result. Fan-out after the store is for the resources of one message or document, not for millions of rows.
• Through a bridge, each item becomes a unit of the next pipeline instead (see Correlation and lineage).

Statuses

A unit has one of six statuses. Steps name none of them: the engine sets each at a durable point, never between two steps.

How a unit moves between them:

Each status a unit enters is also a webhook event of the same name, for a pipeline that subscribes to it — see Webhooks.

error_step says where a unit failed: the name of the step that threw (see defineStep for how steps are named), sender when the sender rejected it, store when its stored value could not be read, or engine when no single step is to blame. A Defer keeps the error fields; a Retry clears them. Every retry, automatic or an operator's, starts at the store and runs every step after it again, with the code the workspace has then.

Correlation and lineage

Work in several stages is several pipelines joined by a bridge: pipeline A ends in .sender(intoPipeline("b")), and pipeline B starts with .source(fromPipeline("a")). Each item A's steps produce is written straight into B's store as a unit of B, in the transaction that closes A's unit as sent.

pipeline("deliveries")
  .source(webhookSource({ port: env("DELIVERY_PORT") }))
  .store(postgresStore({ codec: "text" }))
  .step(splitDelivery)
  .sender(intoPipeline("practitioners"));

pipeline("practitioners")
  .source(fromPipeline("deliveries"))
  .store(postgresStore({ codec: "json" }))
  .step(buildPractitioner)
  .step(fhirItems())
  .sender(toAidbox);

• B has no steps before its store, since nothing would run them; the builder after fromPipeline offers only .store() and .webhook(). Each of A's items must fit B's codec, or A's unit fails with engine/codec_mismatch.
• The two must name each other. A pipeline that sends into, or reads from, a pipeline that does not answer in kind refuses the workspace load.
• origin_unit_id on each unit of B names the unit of A it came from.
• correlation_id is the id of the unit the work started from. A unit a source creates takes its own id, and a bridge copies it unchanged, so everything one delivery produced, across pipelines, shares one correlation id. The Unit page lists the units that share it.
• B's units start with A's order_at, so the versions of a resource B sends keep the order of A's events. They start with no attrs of their own.

Lineage lasts while an item is in flight. A delivered item's row is gone, so the lasting record of what a unit delivered is in the destination, when a step emits one: a Provenance with an id from provenanceId. What was sent can also be produced again by retrying the unit, since the store keeps its input.

Attrs and declared indexes

attrs holds a unit's facets: the dimensions the dashboard filters and counts on, and that steps look units up by. A step sets them with withAttrs; before the store a row inherits the attrs of the row it was split from.

The store declares which attrs are indexed:

.store(postgresStore({ codec: "text" }), {
  index: { patient_id: "btree", message_type: "btree" },
  search: "trgm",
  batch: 100,
})

• The engine builds the indexes when it loads the pipeline, without blocking writes to a pipeline that already has units. A pipeline gets new facets with no migration.
• The dashboard filters and counts on declared attrs only, beside status, error_kind and error_step. A step's ctx.units may filter on any attr, but an undeclared one reads every unit of the pipeline.
• Every index costs a write per status change. A unit changes status about three times, and each change rewrites each of its indexes. Declare the attrs you filter on, not every attr you set.
• Attrs are untyped. A misspelled attr name compiles, and its filter finds nothing.
• Attrs are shown to anyone who can read the dashboard. Record what an operator needs to find a unit.

search_text

withSearchText sets the text the dashboard's search reads. It is kept apart from attrs, so the facets never read it. With search: "trgm" declared, the Units page finds units whose text contains every word typed, each word at least three letters or digits long; without it, the pipeline has no search. A step's ctx.units does not read it.

Pipeline state

State is a small durable key/value store a pipeline keeps between inputs: a cursor, a "seen this already" marker, a cached lookup, a fingerprint of what the destination holds. It is declared as a typed handle, and any step that imports the handle can use it:

const lastSeen = defineState<{ orderAt: string }>("adt.last-seen");

• One scope, the pipeline. Each pipeline has its own copy of every namespace, and no pipeline reads another's.
• Writes commit with the work. A step's writes commit with its input or its take, and nothing of an input refused or a take abandoned is written. Other work sees only committed values. See State for the exact rules.
• A sender keeps state too. The Aidbox sender records a fingerprint of each resource it delivered, which skipUnchanged() reads. The admin library reads and clears them.
• No time to live. A namespace that would grow forever is cleaned with del.
• One writer per key is your convention. Two steps writing one key race, and the last write wins. State is for markers, cursors and caches, not for coordinating work.
• Names are what the rows are stored under. Renaming a namespace, or the pipeline, leaves the old rows behind. The engine logs pipelines whose stored data no declared pipeline names, and never drops them.

Why steps get no database handle

A step reaches the outside world through what its context offers — ctx.state, ctx.units, ctx.lookup, ctx.sender.queued, terminology — and through systems for external APIs. Reference data you load into the database, such as a provider registry or a code table, is read by key through ctx.lookup, from tables of the reference schema only. A step gets no database connection, by decision:

• Retries and deadlines stay safe. Steps may run more than once: every retry starts at the store, and a take that passes its deadline is abandoned and taken again. The engine writes a unit's status, its items, its state and its events in one transaction with the result, so abandoned work writes nothing. A step's own writes would be repeated by a retry, and kept by work that was abandoned.
• The engine's tables are not an interface. SQL a workspace runs against them breaks, with no compile error, on the release that changes them. ctx.units and ctx.state are versioned with the SDK, and so is the admin library that scripts use.
• A read-only handle could not be held to reads. Raw SQL on the same connection could write, and a database driver in the SDK would ride in every workspace bundle.

For the same reasons a sender's run gets no database handle: it returns an outcome for each item, and the engine records them.