Pipelines

A pipeline is data: a JSON graph of nodes and wires, versioned in the database's x_blueprints table. publish stores it; the serve daemon executes it. The database stays passive — it holds the blueprint and emits change events; all execution lives in the runner.

{
  "version": 1,
  "name": "sensor-mirror",
  "nodes": [
    {"id": "trig", "type": "TableTrigger", "config": {"table": "sensor"}},
    {"id": "calc", "type": "CalcColumns",  "config": {"table": "sensor_calc"}},
    {"id": "sink", "type": "TableSink",    "config": {"table": "sensor_calc", "columns": [
      {"name": "entity"}, {"name": "ts", "type": "number"},
      {"name": "p00", "type": "number"}, {"name": "p00_dev", "type": "number"}
    ]}}
  ],
  "edges": [
    {"from": "trig.rows", "to": "calc.in"},
    {"from": "calc.rows", "to": "sink.in"},
    {"from": "trig.entity", "to": "sink.entity"},
    {"from": "trig.ts", "to": "sink.ts"},
    {"from": "trig.p00", "to": "sink.p00"},
    {"from": "calc.p00_dev", "to": "sink.p00_dev"}
  ]
}

Nodes

Node	Role
TableTrigger	the source: drains the table's sync feed from a cursor until dry — rows tables ride x_rev, timeseries ride x_seq; the family is auto-detected
CalcColumns	computes calculated columns from the formulas declared in x_calc_columns
LuaTransform	a free-form transform(row) script — sandboxed, isolated in a child process
Filter	drops rows by predicate
FieldMap	column ↔ field mapping and casts
OnValueChange	passes a row only when a watched value changed since last seen — the alarm gate
QuerySource	emits one SQL statement's answer — the engine aggregates, the pipeline moves it
Schedule	marks the blueprint periodic ({"every": "1h"}): the daemon re-runs it on cadence, no table event needed
TableSink	batched INSERT into the target; onStart/onDone/onError lifecycle events
EmptyTable	snapshot-then-truncate action, wired to a sink lifecycle event
CsvSource	local-file source for imports and backfills

How a run works

The daemon LISTENs x_data. When a subscribed table changes it re-runs the graph: the trigger drains pages from its stored cursor until dry, every page flows through the middle nodes into the sink, and — only when the whole run, sink included, succeeded — the cursor advances (in the x_re_cursors kv table). A failed run replays its pages on the next wakeup: at-least-once, converging on the sink's idempotent writes. A publish re-arms the running daemon through the registry's own x_data event; no restart, no redeploy.

If the table was rebuilt (TRUNCATE — the feed's epoch moved) or the cursor was refused, the trigger winds back to zero and replays from the start; the mirror converges again.

Scale-out

Run more daemons — same package, same conf, more boxes. N daemons against one database split the registry by leases: a kv table (x_re_leases, 45 s TTL) names each blueprint's owner; owners renew by running, everyone else skips. A crashed owner's lease expires and a survivor takes over within the TTL. No coordinator, no engine machinery — just the kv family doing its Redis-shaped job.

OnValueChange

{"id": "gate", "type": "OnValueChange",
 "config": {"key": "xid", "watch": ["status"]}}

Rows pass only when a watched column's value differs from the last this daemon saw for that key (key: "entity" for timeseries state). After a daemon restart the first drain primes the cache silently — an alarm pipeline never re-fires its whole history because a box rebooted ("prime": false floods the first run instead, for consumers that want the full state).

Scheduled pipelines

A blueprint carrying a Schedule node runs on cadence — the baseline recompute's shape:

{"id": "sched", "type": "Schedule", "config": {"every": "1h"}},
{"id": "src",   "type": "QuerySource",
 "config": {"sql": "SELECT entity, ts, p00 FROM sensor LATEST ON ts PARTITION BY entity"}},
{"id": "sink",  "type": "TableSink", "config": {"table": "sensor_latest", "columns": […]}}

The daemon fires it at tick granularity and keeps the last-run mark in the cursor store (mode schedule) — a restart neither re-runs a fresh schedule nor forgets a due one. Set the cadence above the query's own run time — a schedule shorter than the scan it fires runs back-to-back forever; operations has the rule and the numbers.

Watching it

The console's Pipelines page (database menu → Pipelines) shows the registry, every trigger cursor, the declared formulas and the live leases. The same tables answer over SQL, because they are just tables:

SELECT name, version, enabled FROM x_blueprints;
SELECT key, mode, epoch, cursor FROM x_re_cursors;
SELECT key, owner FROM x_re_leases;