Documentation

• 1: Introduction
  • 1.1: What is Eventboat?
  • 1.2: Architecture
• 2: Get Started
  • 2.1: Installation
  • 2.2: Your First Pipeline
• 3: Reference
  • 3.1: CLI Reference
  • 3.2: Configuration
• 4: Developer guide
  • 4.1: Architecture & package map
  • 4.2: Engine internals
  • 4.3: Plugin system & registry
  • 4.4: Configuration & diagnostics
  • 4.5: Expressions & scripting sandbox
  • 4.6: Observability & operations
  • 4.7: Testing guide
  • 4.8: Building & release
  • 4.9: Contributing

1 - Introduction

1.1 - What is Eventboat?

Eventboat routes events through directed acyclic graphs (DAGs) of transforms — from sources (Kafka, HTTP, cron, SQL, files) to sinks (Kafka, HTTP, files), with at-least-once delivery, dead-lettering, and replay.

It is agent-native: every capability is accessible through MCP (Model Context Protocol), CLI, and a language server — so AI agents can write, verify, deploy, and operate pipelines autonomously.

What makes it different

The one-line pitch

Eventboat lets AI agents build and run event pipelines that don’t lose messages — because machines verify every step before it goes live.

1.2 - Architecture

Three-layer model

YAML (+overlay) → Config (typed) → Static IR → Runtime Engine
                                          ↓
                              Source → Transform → Sink plugins

• Config layer: YAML parsing, strict schema validation, variable substitution
• Static IR: validated DAG + precompiled CEL programs + Starlark programs + schema
• Runtime: spool + settle + checkpoint engine, consuming only the IR

Reliability model

source → [spool: append-only durable queue] → in-memory DAG → sinks
                │                                  │
                └── checkpoint ←── settle tracker ←── terminal states

• Spool: every message hits SQLite before the DAG sees it (invariant 1)
• Settle: each message settles when all branches reach terminal state
• Checkpoint: advances only over settled prefix (invariant 2)
• Crash recovery: kill -9 → restart → replay from checkpoint, never lose (invariant 3)
• Dead letters: exhausted retries → DLQ store with query + replay CLI

Seven invariant tests

Each has a dedicated test that must pass in CI:

1. Spool before visible
2. Checkpoint advances only after settle
3. Kill -9 replay covers all unsettled
4. Dead-letter write failure blocks settle
5. required: false edges don’t block siblings
6. Redelivery keeps message ID stable
7. Cursor watermark never exceeds settled

Four machine gates

2 - Get Started

2.1 - Installation

Prerequisites

• Go 1.25+ (for building from source)
• No runtime dependencies — everything is compiled into one binary

Install

go install github.com/eventboat/eventboat/cmd/eventboat@latest

Verify:

eventboat help

You should see the help screen with 11 commands listed.

What’s in the binary

Next steps

• Write your first pipeline
• CLI reference

2.2 - Your First Pipeline

The three-section format

Every Eventboat pipeline is a single YAML file with three top-level sections: sources, transforms, and sinks — connected by from edges.

apiVersion: eventboat/v3
kind: Pipeline
metadata: { name: my-first-pipeline }

sources:
  ingest:
    cron: { expression: "*/5 * * * *" }   # every 5 minutes

transforms:
  hello:
    from: [ingest]
    script: |
      payload.greeting = "hello from eventboat"
      payload.timestamp = meta.ingest_time

sinks:
  console:
    from: [hello]
    drop: {}                              # discard (demo)

Verify (gate 1)

eventboat verify --config pipeline.yaml

Output: pipeline.yaml: 0 error(s), 0 warning(s)

Run

eventboat run --config pipeline.yaml

The pipeline starts; every 5 minutes the cron source fires, the Starlark script enriches the message, and the sink receives it.

Add branching (CEL predicates)

sinks:
  important:
    from: { hello: { when: 'payload.score > 100' } }
    http: { url: "https://api.example.com/alerts" }

  archive:
    from: [hello]                          # unconditional edge
    drop: {}

Agent mode

# Start the MCP server — AI agents connect via stdio
eventboat mcp --stdio

# Or with the admin UI
eventboat mcp --http

Next steps

• Configuration reference
• CLI reference
• Examples

3 - Reference

3.1 - CLI Reference

Overview

eventboat [--json] verify --config <pipeline.yaml> [--strict]
eventboat [--json] test <testfile-or-dir> [...]
eventboat run --config <pipeline.yaml> [--data-dir DIR] [--ephemeral]
eventboat run --config-dir <dir> [--runtime runtime.yaml]
eventboat [--json] trigger --config <job.yaml> [--parameters '{"from":"..."}']
eventboat [--json] jobs list --config <job.yaml> [--limit N]
eventboat [--json] jobs show <run-id> --config <job.yaml>
eventboat [--json] explain --config <pipeline.yaml> [--message f.json] [--topology]
eventboat [--json] replay --config <pipeline.yaml> (--dlq | --spool --from N | --job <run-id>) [--dry-run]
eventboat repl [--message sample.json] [--cel 'expr' | --script f.star]
eventboat lsp
eventboat [--json] plugin catalog
eventboat [--json] plugin schema <name>
eventboat mcp (--stdio | --http) [--config-dir <dir>] [--data-dir DIR]

verify

Statically validate a pipeline. Checks schema, topology invariants (no cycles, no orphans, source has no in-edges, sink has no out-edges, at least one source→sink path), CEL predicate compilation, Starlark script compilation, job configuration, and semantic lint.

test

Run contract test suites against the real in-process engine. Test files declare injection points, expected captures, and DLQ assertions.

suite: my-test
pipeline: ../pipeline.yaml
cases:
  - name: vip-order-routed-correctly
    inject: { at: ingest, messages: [fixtures/order.json] }
    expect:
      capture: { at: out }
      messages:
        - payload.total: 12000    # subset match

run

Execute a pipeline. Job pipelines (with run.mode: job) run under the jobs manager with scheduling and catchup. --config-dir starts a multi-pipeline daemon with the admin surface.

trigger

Manually fire a job pipeline once, optionally with parameters (backfill).

eventboat trigger --config sync.yaml --parameters '{"from":"2026-08-01","to":"2026-09-01"}'

explain

Deterministic walkthrough of a pipeline. With --message, performs real CEL evaluation and Starlark dry-run on the sample. With --topology, renders the DAG (mermaid + ASCII).

replay

Re-inject dead letters (--dlq), a spool window (--spool --from N), or one job run’s dead letters (--job <run-id>) into a live pipeline.

repl

Evaluate CEL predicates and Starlark scripts against one sample message without running a pipeline.

eventboat repl --message sample.json --cel 'payload.score > 100'
eventboat repl --message sample.json --script transform.star

mcp

Start the MCP server for AI agents.

3.2 - Configuration

Top-level sections

Three-section topology

Nodes are organized by section; edges declared on the downstream side via from:

sources:
  ingest:
    decoder: json
    kafka: { brokers: ["${KAFKA_BROKERS}"], topics: [orders] }

transforms:
  enrich:
    from: [ingest]                         # unconditional edge
    script: |
      payload.total = payload.price * payload.qty

sinks:
  eu-out:
    from: { enrich: { when: 'meta.region == "eu"' } }  # conditional edge
    kafka: { topic: orders-eu }

Edge attributes

Attributes on from elements:

Variable substitution

• ${VAR} — environment variable (unset = error)
• ${?VAR} — optional (unset = omit key)
• ${constants.name} — pipeline constant
• Applies to all string values

Built-in plugins

Sources

Sinks

Codecs

Job pipeline example

apiVersion: eventboat/v3
kind: Pipeline
metadata: { name: nightly-sync }

run:
  mode: job
  schedule: "0 1 * * *"
  overlap: skip
  catchup_window: 2h
  skip_if_successful: true
  retention: { history: 90d }

parameters:
  from: { type: string, default: cursor }
  to:   { type: string, default: now }

sources:
  pull:
    sql:
      driver: mysql
      query: |
        SELECT * FROM orders
        WHERE updated_at >= :from AND updated_at < :to
      args: { from: "${parameters.from}", to: "${parameters.to}" }
      cursor: { column: updated_at }
      pagination: { key: [updated_at, id], page_size: 5000 }

transforms:
  enrich:
    from: [pull]
    script: |
      payload.source_system = constants.source_system

sinks:
  out:
    from: [enrich]
    kafka: { brokers: ["${KAFKA_BROKERS}"], topic: orders-sync }

4 - Developer guide

These guides are rendered at build time from the eventboat/eventboat repository; edit them there, under docs/developer/.

4.1 - Architecture & package map

Eventboat is a single Go binary that routes events: messages arrive from sources (Kafka, HTTP, cron, file, SQL), flow through an explicit DAG of transforms and filtered edges, and land at sinks — at-least-once, verifiable before anything runs, and replayable after a crash. Pipelines are declared in YAML, statically verified against plugin JSON Schemas and topology rules, and then executed by an engine whose entire reliability model is pinned by eight dedicated invariant tests. Predicates are CEL, transforms are Starlark (or WASM), so there is no custom language to learn.

The processing model

A pipeline is a three-section YAML document — sources, transforms, sinks — joined by depends_on edges into a DAG. The runtime model is:

1. Admission: every inbound message is durably spooled (SQLite, or in-memory with --ephemeral) before it becomes visible to the DAG (internal/engine/admission.go, admit: live / replay / inject modes).
2. Execution: the message fans out along matched edges into transforms and sinks. Each execution branch must reach a terminal state: sink ack, dead letter, filtered (zero matching edges or zero transform outputs), or an optional-edge drop.
3. Commit: a commit tracker counts outstanding branches per message; the checkpoint only advances over the contiguous committed prefix (internal/engine/commit.go). Per-source watermarks are persisted by the async source committers (internal/engine/sourcecommit.go).
4. Recovery: on restart the engine starts its workers first, replays the spool beyond the checkpoint through admission, then starts sources — pull sources resume from their committed watermarks: duplicate delivery, never loss.

Sources, transforms and sinks are plugins in the registry (Plugin system); edges carry predicates (CEL or CESQL) and delivery policies (retries, backoff, timeout, required/optional) (Configuration).

Event lifecycle through the engine

 source node                 engine (internal/engine)                    store
 ------------   -----------------------------------------------------   --------
     │  emit          admit(msg) [live mode]                                │
     │  ◀───────      admission gate: one slot per uncommitted message      │
     │  error         (high watermark, default 10_000 in flight; blocks)    │
     │  = verdict     stamp meta (message_id, ingest_time, source)          │
     │                                          │                            │
     │                                          ├─ AppendSpool ────────────▶ │ seq
     │                                          │       (invariant 1: not    │
     │                                          │        visible before this)│
     │                                          ▼                            │
     │                                   commit.arrived(seq, src, srcSeq)    │
     │                                          │                            │
     │                                          ▼                            │
     │                 dispatchFrom: decode (codec) at entry; decode failure  │
     │                 = dead letter                                          │
     │                                          │                            │
     │                                          ▼                            │
     │                 fanOut: evaluate each outgoing edge predicate          │
     │                     │ matched                                          │
     │         ┌───────────────┴──────────────────┐                           │
     │         ▼ (transform node)                ▼ (sink node)               │
     │   runTransform worker               runSink: batch (size/timeout)     │
     │   Apply → 0 outputs = filtered      encode, order_key, Write          │
     │   Apply → N outputs = fan out       retries per edge delivery policy  │
     │   Apply error → retries then        exhausted: optional → drop,       │
     │   dead letter (backtrace)           required → dead letter            │
     │         │                                 │                           │
     │         └───────────────┬─────────────────┘                           │
     │                         ▼                                             │
     │                 commit.done(seq) per branch                           │
     │                         │                                             │
     │                         ▼                                             │
     │     commitTracker.advanceLocked: contiguous committed prefix          │
     │                         │                                             │
     │          onCommit: metrics, span end, admission slot release          │
     │          onAdvance: persistCheckpoint ── SetCheckpoint ─────────────▶ │
     │                       (spool retention trim ─ DeleteSpoolThrough ─▶ │)
     │                       (frontier ─ post ─▶ source committer ─         │
     │                        Source.Commit ── SetSourceState ────────────▶ │)

Package map

Every directory, what it owns, and what it must not reach into. The layering is enforced by Go imports — each “must not” below is checkable with go list -deps.

Design invariants

These are the properties the architecture is built around. Items 1–8 each have a dedicated test (TestInvariant_* in internal/engine/invariants_test.go); breaking any of them is a review blocker.

1. Spool before visible. A message must not become visible to the DAG until its spool append has succeeded. A failing append refuses the message; it never reaches any sink, and the refusal reaches the source through emit’s error (the source may safely re-emit it).
2. Checkpoint only over committed. The durable checkpoint may advance only across the contiguous prefix of messages whose every execution branch reached a terminal state.
3. Crash = replay, never loss. After a kill -9, the set replayed from the checkpoint is a superset of the uncommitted set. Duplicate delivery is the contract; loss is not.
4. Dead-letter write failure blocks commit. A dead letter that cannot be written is retried; the message stays uncommitted. Dead-letter unavailability slows the pipeline instead of losing messages.
5. Optional-edge failure isolates. A required: false edge failure terminates only its own branch and never blocks sibling branches from committing.
6. Duplicate delivery is safe by design. Re-delivery to an idempotent sink must be harmless; meta.message_id exists for dedup keys. This part is the user’s documented responsibility.
7. Cursor watermark follows commit. Pull sources’ watermarks advance only to the max cursor column of committed messages — job-pipeline resume correctness.
8. Fan-out branches are isolated. Sibling branches share the underlying msg.Decoded / msg.Meta maps (delivery shallow-copies the struct), so transforms must replace those values wholesale and never mutate them in place — an in-place write races a sibling branch’s encode and can crash the process with an unrecoverable concurrent-map fault. This is also the message-ownership contract documented on registry.Message and pkg/plugin.

Project-level invariants that live outside the engine:

• Verify-first for every write path. deploy, the admin REST write surface, the MCP deploy tool and eventboat run all verify through the same composition, internal/verify (load → build → judge). ops.Verify returns the same diagnostic sequence and strict verdict for the same input; ops.Deploy fails when verification fails. There is no bypass channel — the LSP, jobs, testrun and the CLI enter the same composition (internal/verify, candidate 05).
• The engine does not parse payloads. Raw bytes plus a codec marker are the spooled truth (internal/registry/registry.go, Message.Raw); decoding and encoding are codec plugins. Replay stays compatible with codec upgrades because the spool never stores the decoded form.
• Plugin isolation. Out-of-process plugins are separate processes speaking gRPC on loopback with a one-shot token (internal/rpcplugin); WASM guests run in a capability sandbox with memory caps (internal/wasmhost); Starlark scripts run without while/recursion/I/O under a step budget (internal/lang/starhost). A wedged plugin degrades one worker or one pipeline, not the process.
• Single binary. Everything — engine, store, language hosts, admin UI, MCP, LSP — compiles into one static binary with CGO disabled (every driver is pure Go; SQLite via modernc.org/sqlite), see the Dockerfile.

How a config becomes a running pipeline

The same path serves every entry point:

config.LoadBytes ──▶ ir.Build ──▶ engine.New ──▶ engine.Run(ctx)
     (diagnostics)     (static IR)   (plugins,       (replay, workers,
  cfg_* topo_* expr_*  + compiled     channels,        sources, commit)
  plugin_* codec_*     programs,      commit tracker)
  job_* lint_*         explain-safe
                       instances)

Entry points on top of this: eventboat run --config (one pipeline, OTLP-only telemetry), eventboat run --config-dir (the multi-pipeline daemon; every YAML in the directory loads, the admin surface serves on 127.0.0.1:7788 by default), ops.Deploy (the admin/MCP write path — verify-first, then drain-and-swap), and the one-shot verbs (verify, test, explain, replay, trigger, jobs). Job pipelines (run.mode: job) put internal/jobs between the config and the engine: one engine per run, sharing one admission pool across overlap: all runs.

Every runner reads one run outcome (candidate 02): the engine’s Wait/Outcome classifies the run as completed, partial, failed or interrupted, and the caller maps that onto its own process contract — job runs to success/partial/failed/canceled, the batch verb to exit 0/1, the long-lived continuous verb to exit 0 on a graceful stop and 1 on a failed run, the daemon to completed/failed (see Observability & operations). A source failure stops the engine in every mode; restart resumes from the source watermarks (duplicate delivery, never loss).

Every entry point opens the pipeline’s store through one store.Owner (internal/store/owner.go, candidate 04): the canonical file is <data-dir>/stores/<sanitized name>.db for the one-shot verbs and the daemon alike, one handle per pipeline per process, closed on shutdown. --ephemeral substitutes a cached in-memory owner, so all surfaces of one process see the same data. The old eventboat.db / stores/pipeline.db layouts are retired without migration (beta ruling). Modules depend on the store facets they use — engine.New takes the spool + dead-letter facets, jobs.New run history plus the engine’s facets, ops the combined Store.

Where to read next

• Engine internals — the accept path, commit frontier, persistence and recovery in detail.
• Plugin system & registry — how to register, write and test plugins.
• Configuration & diagnostics — the load pipeline and every diagnostic code.
• Testing guide — where each guarantee is pinned.

4.2 - Engine internals

The engine (internal/engine) executes one compiled pipeline against one store: source admission through a durable spool, in-memory DAG execution with commit tracking, per-edge delivery retries, dead lettering, checkpointing and backpressure. The package is five files plus tests: engine.go (lifecycle, persistence), admission.go (the inbound path), nodes.go (transform/sink workers, dead letters), commit.go (the commit frontier), plus sourcecommit.go (the per-source watermark committers). Every reliability property is phrased as one of the eight invariants (see Architecture), each with a dedicated test in internal/engine/invariants_test.go.

Startup sequence

engine.New resolves plugins and allocates structures; Run starts them (internal/engine/engine.go):

1. IR in, plugins instantiated. For each node in ir.Pipeline.Order: sources and sinks instantiate through the registry (reg.NewSource / reg.NewSink) or, for grpc: blocks, through rpcplugin.SpawnSource / SpawnSink; codecs resolve by name (named codecs: declarations come pre-instantiated on the IR, bare names instantiate config-less). Transforms instantiate through reg.NewTransform and are immediately Init-ed with a TransformEnv — constants, parameters, a node-scoped logger and the slow-call advisory — before workers exist. A transform Init failure closes the instance and aborts startup.
2. Commit tracker. newCommitTracker gets one srcTracker per source node and the two callbacks: onCommit (per-message bookkeeping) and onAdvance (checkpoint persistence).
3. Admission and committers. newAdmission builds the inbound path (below); one sourceCommitter is allocated per source node.
4. Run(ctx) phases, in this order (candidate 01; the ordering is an invariant, not a preference):
   1. committers + workers — one goroutine per source committer, then the transform and sink workers;
   2. crash replay — read the checkpoint and re-dispatch every spool row beyond it through admission (replay mode). Replay dispatches into node channels, so the consumers must already exist: a recovery with more uncommitted rows than a channel holds used to block forever on a channel nobody read. A cancelled context stops the scan without an error (a voluntary stop, v1.24);
   3. sources last — a fresh source restores its persisted state via Init(state) when one exists, then Runs (or Pulls for pull sources). Sources last so replayed rows are already registered before new emissions race them. replayDone flips only after the replay scan finishes; WaitCommit refuses to read “all committed” before that (an unregistered replay row momentarily looks like outstanding == 0).

The admission path

internal/engine/admission.go owns how a message enters the DAG: the backpressure gate, the acquired-slot ledger, stamping, the spool append, commit registration and the dispatch step. One entry (admit) has three modes that differ only in request fields:

Live admissions also record an accept time (commit latency) and a sampled span; replay and inject do not. Every mode takes admission quota — one slot per uncommitted message. The uncommitted set is ≤ HighWatermark by construction, so replay cannot wedge on the gate, and the quota is uniform across entries. The gate can be a shared pool (Options.Admission) so the jobs manager aggregates limits.max_in_flight across concurrent overlap: all runs instead of multiplying per run.

The verdict admit returns is the source contract’s emit error: nil = durably accepted; a ctx error = engine shutdown; anything else = refusal (not durable, never visible, safe to re-emit — see Plugin system). A refused live emission returns the slot it took; a refused injection returns the error to InjectAt/InjectReplay.

The accept path

accept is the thin live shell over admission (engine.go):

1. Admission gate. The caller blocks on the gate — one slot per uncommitted message. When the high watermark is reached, sources stop being served (backpressure; eventboat_backpressure_events_total). A blocked accept that races shutdown returns ctx.Err() and the message is not accepted — the source may safely re-emit it (at-least-once).
2. Stamping. message_id (preserved if the caller supplied one, so replays keep identity), ingest_time, source, plus any Options.MetaStamps (e.g. job_run_id). The source node’s decoder names the message’s codec (json default).
3. Durable spool append. Store.AppendSpool returns the spool sequence. On failure the admission slot is returned and the message is refused — never delivered (invariant 1).
4. Registration. acceptedAt[seq] records the accept time (commit latency), the acquired-slot ledger remembers the backpressure slot, and commit.arrived(seq, sourceNode, raw.SrcSeq) pre-registers exactly one outstanding branch and records the source emission for watermark tracking. Then dispatchFrom decodes at the source entry (decode failure = dead letter) and fans out.

Injection

InjectAt(node, msg) and InjectReplay(node, msg) take a registry.Message (Raw/Meta/Codec/ID); the message’s identity and codec travel with it. InjectReplay stamps meta.is_replay=true and meta.original_message_id from msg.ID, and preserves the ID — a csv dead letter replays as csv, never as json. InjectAt at a source node fans out of it (spooled and stamped like a live emission, but not attributable to the source’s watermark); at an internal node it is spooled and enters INTO that node (transform script, sink batch), which is what makes operator replay re-execute the failing step.

Per-edge delivery

Edges are resolved by the IR with defaults from edge_defaults and overrides per edge (internal/ir/ir.go): Required (default true), Retries (default 3), Backoff (exponential | constant, base Options.BackoffBase = 100ms, exponential doubling capped at 30s), TimeoutMs (per-attempt, Options.DefaultTimeout = 30s when unset), BufferMax (per-edge channel surge capacity, default 128).

• Sinks batch engine-side (runSink, nodes.go): batch.size and batch.timeout_ms control flush; writeBatch encodes each message, evaluates order_key, takes the strictest retry policy of the edges in a mixed batch, and on success commits every instance. After retries are exhausted, a required: false edge drops the message (counted, eventboat_optional_drops_total) while a required edge dead letters.
• Transforms retry per the incoming edge’s policy (review R6), then dead letter with the plugin’s backtrace if any. A transform failure never fails the node. Zero outputs filter the message (commit-as-filtered + NoMatch); N outputs expand the commit accounting by N-1 extra branches — the split plugin’s 1→N contract.
• Dead lettering (deadLetterMsg → writeDeadLetter) retries the store write forever (backoff Options.DLBackoff = 500ms) until it succeeds or the engine shuts down: a dead letter that cannot be written blocks the commit (invariant 4) — degraded, not lossy. Abandon calls the same writer with a caller-bounded ctx (candidate 02): a failed write stops the attempt and the message stays uncommitted for the next run’s replay. The record carries the full original message, node/edge attribution, reason, the dead-letter class recorded where the failure was produced (store.DeadLetter.Class: decode/codec/encode/delivery/canceled or the transform failure kind — never re-derived from the reason text), backtrace and run id.
• Delivery on shutdown is deliberately not committed: an instance that could not be queued before ctx.Done() stays uncommitted and is replayed from the spool on restart (invariant 3).

The commit frontier

internal/engine/commit.go implements the heart of the reliability model:

• outstanding map[int64]int counts open execution branches per spool seq; openBranches is a running sum of positive values so snapshot() is O(1) (it is polled from the hot path — WaitCommit’s 2ms loop, ops status).
• arrived(seq) pre-registers one unit; fan-out add(seq, n-1) grows it; each terminal event calls done(seq) (i.e. add(seq, -1)).
• advanceLocked scans the contiguous prefix: from committedPtr up while entries are zero/closed; the scanned range is committed, and the just-committed seqs plus the per-source frontier snapshot become the callback payload. Callbacks run outside the tracker lock (they do store I/O; the lock must not convoy on fsync).
• The hole barrier (adversarial review 2026-09-26). A seq with no outstanding entry stops the scan unless it carries a removed mark — the record forceTerminal leaves when the abandon path deliberately terminates a message (its durable dead letter is already written). An unmarked absence is the AppendSpool → arrived window: the row is durable but not yet registered, so it is not committed and the checkpoint must not cross it. Treating that absence as committed let a crash replay from beyond a row that was never delivered, which no-cursor sources (cron, http_server) cannot re-emit. Marks are consumed by the sweep and bounded by the in-flight window. The straggler branch (a forceTerminal racing an in-flight arrived) still delivers the late terminal event and now also sweeps the source refs under the cursor.
• Restart seeding. resumeCheckpoint moves the cursor and the persistence barriers to the recovered checkpoint: rows at or below it are committed by definition and are never registered in the new run, so the barrier must start past them (otherwise the prefix would stall on the first pre-checkpoint seq and WaitCommit/Quiesced would wait for a flush that already happened). A fresh tracker starts its cursor at seq 1.
• Ordered srcRefs FIFO. Each source emission is recorded as (spool seq, node, srcSeq). Arrival is near-ordered (each source goroutine appends and registers back-to-back), but two sources racing through accept can invert; addSrcRef splices those into sorted position, so the sweep can pop a contiguous prefix off the head instead of scanning the whole in-flight window per message (the old map scan was O(high watermark) per message).
• Per-source frontiers. srcTracker keeps arrivedAt/committedAt sets per source seq and advances front over the contiguous prefix of emissions committed this run. Replayed spool rows are not attributable to a live source emission and never reach the tracker — sources re-pull their uncommitted tail instead, which is exactly the duplicate-delivery contract. Swept ranges are deleted so the maps stay bounded by the in-flight window.
• Commit(ctx, throughSrcSeq) contract (registry.Source): the engine calls it when the contiguous committed frontier advances, with the highest committed srcSeq for that source; the source returns its new durable state (Kafka offsets, file offsets, SQL watermarks), which the source committer persists via Store.SetSourceState.
• Source committers (sourcecommit.go, candidate 01). One worker per source receives coalesced maximum frontier advances from persistCheckpoint and calls Source.Commit off the committing goroutine — so a source may hold an internal lock across emit (the file source does) without deadlocking the pipeline. A failed commit or state write keeps its pending value and is retried on the next advance; the srcPersisted monotonic guard lives in the committer. After drain(), Run waits (bounded) for the workers to exit and flushes each pending frontier synchronously with context.WithTimeout(context.Background(), DrainTimeout) — the engine ctx is already cancelled, and a cancelled Commit would discard the last frontier. A source state that lags the checkpoint only widens the crash replay window: duplicate delivery, never loss.

Persistence

persistCheckpoint (engine.go) runs on the goroutine that advanced the prefix, outside the tracker lock. Concurrent advances can flush out of order, so monotonic guards under persistMu keep everything from regressing:

• persistedThrough — the highest checkpoint actually written (Store.SetCheckpoint); CheckpointPtr mirrors it for status. A failed write only widens the replay window; the next advance retries. The flush position flushAttempted advances even on failure so observers waiting on the durability barrier are not wedged by a permanently broken store.
• durableThrough() reports flushAttempted — the visibility barrier. WaitCommit and Quiesced require durableThrough >= committedThrough, so “committed” still implies “persistence attempted” now that flushing is asynchronous to the tracker lock. Quiesced additionally refuses to read “no work” while the crash-replay scan is in flight (replayDone) or while an admission holds a gate slot but has not registered yet (admitting) — both windows would otherwise look like an empty pipeline (candidate 02).
• Spool retention. One batched trim per window of durable progress: when persistedThrough >= retentionDue, rows at or below persistedThrough - SpoolRetention are deleted (Store.DeleteSpoolThrough) and the next due point moves a full window. The cutoff is anchored at persistedThrough, not the in-memory frontier — trimming ahead of what a restart would replay from would turn at-least-once into loss while checkpoint writes fail. DefaultSpoolRetention is 10_000 rows; storage.spool_retention in the Runtime config overrides it (internal/runtimecfg).
• Dead-letter retention (opt-in). Rides the same window: a pipeline that sets dlq.retention sweeps dead letters older than Clock() - DLQRetention once per retention window (Store.DeleteDeadLettersBefore, bounded 10,000-row batches, driven by the dead_letter(pipeline, created_at) index so a large backlog stays a range scan). Unset (the default) keeps everything forever — dead letters are operator data for replay, so the engine never deletes what the pipeline did not ask to delete; removed rows are gone from replay for good. A failed sweep is logged and retried by the next window, exactly like the spool trim — it never blocks the commit path (deleting terminal artifacts cannot affect the invariants).
• Per-source Commit states persist through the committers (above), with the monotonic guard inside each committer; the checkpoint stays the durable barrier.
• Single writer (candidate 08). A pipeline store is written by exactly one process at a time. A running engine holds an exclusive OS file lock on a sidecar next to the canonical database — <store>.lock — via internal/store/lease.go (flock on Unix, LockFileEx on Windows, both non-blocking); the kernel releases the lock when the process dies, so there is no TTL heartbeat and a crash leaves no stale lease. The daemon acquires the lease per deployed pipeline and releases it when the instance stops; run --config, trigger and a live replay acquire it for the duration of their run and refuse with a message pointing at the admin/MCP surface when they cannot (a second engine on the same spool would race the checkpoint). On a target without a file-lock implementation the run refuses loudly instead of silently skipping the single-writer guarantee. The lock is an advisory lock: the filesystem must honor it (local disk or a block-backed volume); on NFS/RWX it may not protect across nodes, so keep the k8s deployment on Recreate with one replica. The lock is a sidecar so SQLite’s own locking is untouched, and the file is never deleted — an unlocked leftover is expected and harmless.

Group-commit write path (P1, design 2026-09-24 §2.2)

Every store mutation runs on one writer goroutine over one dedicated connection (internal/store/writer.go); reads (Checkpoint, SourceState, replay, dead-letter and job-run queries) stay on the pool. There is exactly one write transaction at a time, so writer-writer SQLITE_BUSY contention is gone by construction. The pragma contract (WAL, synchronous=NORMAL, busy_timeout) is unchanged.

AppendSpool is a group commit:

• The writer drains the queued writes and commits one transaction per group — a single multi-row INSERT for the spool, capped at storage.write_batch.max_rows (default 256) spool rows per transaction. spool.seq is INTEGER PRIMARY KEY AUTOINCREMENT, so one statement assigns contiguous rowids: the group’s seqs come back as last_insert_rowid - n + 1 .. last_insert_rowid, in queue (admission) order.
• Every caller blocks until its own group has committed and returns the seq of its own row. A failed transaction fails every waiter it carried with the same error — the refusal contract is unchanged (nothing durable, the source re-emits). AppendSpool takes no ctx and has no “give up halfway” path: a source cancelled while its append is in flight still gets the row committed and registered, so a committed row can never be invisible to the commit tracker and pin the contiguous prefix (the orphan-row hazard, TestAppendCancelDuringAppendRegistersRow).
• SetCheckpoint and SetSourceState coalesce: one upsert per pipeline (the maximum seq) and one per (pipeline, source) (the request with the highest srcSeq, carrying its paired state — never a field-level merge) per group. Monotonic high-water marks make a later transaction unable to regress an earlier value; a caller whose value was superseded still succeeds, with its value or a larger one durable.
• Close stops the writer, refuses everything still queued with ErrClosed, refuses a transaction that was in flight (rollback — nothing landed), joins the writer goroutine and releases the dedicated connection: a caller blocked in AppendSpool is always woken, never left blocking.

No artificial linger. storage.write_batch.max_wait_ms (default 2) does not make the writer sleep: a timer’s floor is the OS quantum (>500 µs on Windows), which would cap a lone producer at ~1K rows/s while gaining nothing — a lone producer cannot enqueue a companion until its own append returns. Instead the value maps to a bounded probe budget (up to 16 scheduler yields, companionProbes): after a group commits, the writer yields to let the callers it just released re-enqueue, then commits whatever arrived. 0 disables companion collection (write-through). Observed batch sizes are available through the optional WriteOptions.OnBatch(rows, waited) hook; the store stays a leaf and never imports telemetry (§2.6.3 is the tuning surface).

Bounds and fairness. storage.write_batch.max_rows is capped at 2000 (a Runtime load error above it; the store clamps defensively): 2000 rows is 18000 SQL bindings, and a larger group would exceed the driver’s parameter limit, fail the whole transaction and refuse every waiter in a loop. A checkpoint or source-state request whose value is already durable (a later group carried a higher one) is resolved before its group’s transaction, so a failing sibling statement cannot refuse a caller whose value is safe.

Recovery

Crash recovery is Run’s second phase: read the checkpoint, replay every spool row beyond it (Store.ReplayFrom) through admission, re-dispatching each into the DAG — with the workers already running, so a backlog larger than a node channel cannot wedge. Rows whose source node no longer exists in the IR are committed immediately rather than wedging the contiguous prefix forever. Pull sources resume from their persisted watermark, so the uncommitted tail may arrive twice: once via spool replay, once via re-emission — duplicate delivery, never loss (invariant 3; TestInvariant_Kill9ReplayReplaysAllUncommitted).

Job pipelines resume runs found in pending/running on startup; internal/jobs drives each run through Engine.Wait and maps the Outcome onto the run status (below). Run admission — the overlap policy, the run record and the goroutine registration — is one critical section, and Manager.Stop waits until every run persisted its terminal state, so ops.Drain/Deploy really wait for drained runs and a restarted manager cannot pick up a run the previous instance was still driving (candidate 08). A canceled run that must stop immediately dead-letters its outstanding set through the bounded Abandon(ctx, reason): the durable record is written first and only then is the tracker cleared (force-terminate, which also releases the admission slot the old path leaked — candidate 01), so the checkpoint prefix never wedges (review R2). A ctx cancellation or store failure stops the abandon attempt, leaves every unrecorded message uncommitted and returns an error — the next run replays them (never loss, candidate 02). The normal dead-letter path keeps invariant 4: it retries forever on the engine ctx; only Abandon is bounded.

Run outcome

One decision point for “how did the run end” (candidate 02): every runner calls Engine.Wait(ctx, runDone, WaitOptions) and maps the returned Outcome onto its own process contract — none of them re-derives the terminal state.

• Statuses. RunStatus is completed | partial | failed | interrupted: completed = quiesced cleanly; partial = quiesced with dead letters; failed = the engine stopped itself (worker-fatal or a source failure); interrupted = the caller’s ctx was canceled (never the engine’s internal cancel, so an engine self-stop is never misreported as interrupted).
• Classification priority (one classifyOutcome switch): worker-fatal → source errors → caller cancellation → dead letters > 0 → completed.
• Outcome carries status, rows read, committed, dead-lettered, the SourceErrors snapshot, the worker-fatal error, and the abandon count / error. FailureText() renders the operator-facing cause.
• WaitOptions: AbandonOnCancel (jobs: R2 semantics; batch does not — uncommitted rows replay), AbandonReason, AbandonTimeout (bounded, and the caller ctx is already canceled on that path, so Wait derives a fresh context) and DrainTimeout (the post-cancel wait bound, defaulting to the engine’s DrainTimeout + 5s). Past that bound a non-cancellable writer may outlive the run by design; Wait then folds in the engine’s published fatal (failNode records it before it cancels) instead of waiting on Run.
• WaitQuiesced stays the low-level primitive (tests use it); Wait builds on the same loop.

A source failure (v1.24: Run/Pull returned a non-nil error) stops the engine in every mode: the error is recorded first, then the engine cancels — a continuous pipeline must not keep running with a dead source. Restart resumes from the source watermarks (duplicate delivery, never loss). An error returned under an already-cancelled engine ctx is a shutdown artifact and never a source failure. Options.OnSourceError was deleted; the SourceErrors snapshot carried by the outcome is the only channel.

Shutdown

Every pipeline runner registers signal.NotifyContext for SIGINT and SIGTERM (the run, config-dir daemon, mcp, replay and trigger verbs — commit a9250c8 closed the docker-stop gap). On ctx cancellation the engine:

1. stops accepting (accept returns ctx.Err()),
2. drains: waits up to Options.DrainTimeout (default 10s, overridable by limits.drain_timeout) for worker and source goroutines, then hard-cancels,
3. flushes pending sink batches (the sink worker’s ctx branch flushes before returning),
4. closes sinks and transform masters (clones close themselves),
5. stops the source committers and flushes their pending frontiers with a bounded, non-cancelled context (candidate 01).

Run then returns the fatal error, if any. The ops layer prints the final status line (the “settle status” report: counts + checkpoint).

The fatal path

Per-message failures dead-letter. A node that cannot run at all is fatal: failNode records the first error (first error wins), cancels the context, and Run returns it after draining — the run outcome is failed. The motivating case is a transform whose Clone fails: the master instance is precisely what the plugin declared unsafe to share, so the pipeline fails instead of racing workers on it (nodes.go, runTransform). A failed source stops the engine the same way without a worker-fatal error: the error lands in SourceErrors, the outcome is failed, and Run returns nil (candidate 02).

Transform workers

workers (default 1) spawns that many goroutines per transform node, all reading the node’s channel. A plugin implementing TransformCloner is cloned once per worker (Clone() after Init); stateless plugins share one instance — script programs are immutable and per-message state lives in copy-on-write bindings, split has no state. A failed Clone is worker-fatal (see above). TransformFlavor (“script”, “wasm”) selects the per-flavor duration histogram and the budget/timeout counters, which are driven by the typed failure kind (registry.FailureKind); any other flavor takes the generic branch (run accounting + dead-letter class, no per-flavor instrument).

Options and defaults

4.3 - Plugin system & registry

Every source, sink, codec and transform in eventboat is a plugin registered in a registry.Registry (internal/registry/registry.go). Registration carries a JSON Schema; configuration blocks are validated strictly against it (unknown fields are errors), which is the mechanism that keeps agents from inventing plugins or fields that do not exist (redesign-v3.md §5.6). The registry is a leaf package: it imports nothing internal, so any Go program — including your own build of eventboat — can register plugins into it.

The four kinds

Registration also pins an integer ABI version (builtins are version 1) and optional capabilities (e.g. "pull" for sources, "explain-safe" for transforms).

Plugin-name-as-key

In YAML, the plugin name is the key of a node’s block; everything else at node level must be a framework field from the per-section whitelist (internal/schema, the single source config, the registry and the LSP all read — candidate 06):

sources:
  ingest:
    decoder: json              # framework field
    file: { path: input.jsonl }  # "file" is the plugin name
transforms:
  enrich:
    depends_on: [ingest]             # framework field
    script: |                  # "script" is the plugin name
      payload.x = 1
sinks:
  out:
    depends_on: [enrich]
    encoder: json
    file: { path: out.jsonl }

Exactly one plugin key per node (cfg_missing_plugin / cfg_multiple_plugins otherwise). Names colliding with framework fields (depends_on, decoder, encoder, workers, order_key, batch, grpc, version, when, route, buffer, delivery, required, plus from for the rename migration) are rejected at registration (registry review R5, candidate 06: the reserved set is the union of the framework fields, so grpc/version can no longer register a plugin that could never load). script, split and wasm are ordinary registrable names. A version: pin on the node is checked against the registered/manifest version at verify (plugin_version_mismatch).

The typed schema system

Hand-writing a schema string and a map[string]any factory invites drift. The typed layer (internal/registry/typed.go) derives both from one struct:

type kafkaSourceConfig struct {
    Brokers []string `json:"brokers" schema:"minItems=1,desc=Kafka broker addresses"`
    Topics  []string `json:"topics" schema:"minItems=1"`
    GroupID string   `json:"group_id" schema:"default=eventboat"`
}

registry.RegisterSourceT(reg, "kafka", 1, nil, func(c kafkaSourceConfig) (registry.Source, error) {
    return newKafkaSource(c), nil
})

• json tags name the config keys; schema tags carry the constraint grammar: optional, default=, enum=a|b|c, min=, max=, minLen=, minItems=, maxItems=, desc= (must come last so descriptions may contain commas).
• Fields are required by default; additionalProperties: false is emitted always; properties render in declaration order.
• Defaults are injected, not just documented: the same default= tag is applied to zero-valued fields after decode, recursing into nested structs and slices of structs — schema-documented defaults are the runtime behavior.
• Generated schemas compile through the same santhosh-tekuri pipeline as hand-written ones, so plugin_schema diagnostics and the external gRPC manifest path are byte-compatible.
• Golden tests pin every builtin schema: internal/registry/builtin/testdata/schemas/*.json; regenerate with go test ./internal/registry/builtin -update-schemas after changing a config struct. Every schema change becomes a reviewable diff.
• The string-based RegisterSource/RegisterSink/RegisterCodec/ RegisterTransform API remains as the escape hatch (it is what testkit.RegisterFakeTransform uses).

The SCALAR ROOT asymmetry for transforms

RegisterTransformT[T, C] allows C to be a scalar, not only a struct: transform plugin blocks are not forced to mappings. The one deliberate asymmetry exists because the script plugin’s config is the Starlark source text itself (script: | — the block value is a string). A null script: is therefore a type error against the string schema, not an empty config (registry.NewTransform does not normalize nil to an empty object for transforms). The factory also receives dir — the pipeline file’s directory — so configs carrying file paths (wasm module:) resolve against it.

Lifecycle contracts per kind

Source

type Source interface {
    Init(state []byte) error
    Run(ctx context.Context, emit func(Message) error) error
    Commit(ctx context.Context, throughSrcSeq int64) (state []byte, err error)
    Close() error
}

• Init receives the persisted state the source previously returned from Commit — the engine calls it only when state exists. Restore your offset here.
• emit’s error is the admission verdict (candidate 01):
  • nil means the message was durably accepted — it is spooled before it becomes visible to the DAG (invariant 1);
  • an error satisfying errors.Is(err, context.Canceled) / context.DeadlineExceeded means the engine is shutting down: the message was not accepted, and Run/Pull must return nil (a voluntary stop, never a failure);
  • any other error is a refusal: the message is not durable and never became visible, so it is safe to re-emit. The source owns its input semantics (Kafka offset, file offset, HTTP response) and decides whether to retry internally or return the error. The builtin default is to report the refusal as a failed source — the source watermark is the no-loss safety net, and a failed run is louder than a silent zero-output run. (http_server answers 503 instead: one client’s retry is the source policy there.) The engine retries nothing on its own.
  • emit blocks under backpressure (the admission gate), so no buffering is needed. Set Message.SrcSeq to a per-source monotonic sequence — it advances the commit watermark.
• Run returns the source’s completion signal (v1.24): nil = exhausted (a finite source read to its end — batch runs exit, job runs commit) or a voluntary stop; a non-nil error is a failed source: the engine records it in SourceErrors and stops in every run mode (candidate 02), so the run outcome is failed (a failed job, a non-zero exit) instead of a live pipeline with a dead source. ctx cancellation is a voluntary stop — return nil, never ctx.Err(). An infinite source (tailer, broker consumer) simply never returns until ctx is cancelled.
• Commit(ctx, throughSrcSeq) is called whenever the contiguous committed frontier advances; commit your offsets here (Kafka offsets, file offsets, SQL watermarks). The builtin sources use a watermark-bounded scan: the scan starts at the last folded watermark and deletes every pending entry it visits, keeping each call O(new work) — see kafkaSource.Commit / fileSource.Commit in internal/registry/builtin/. Cursor on emitted messages is what job pipelines persist as the sql watermark. Commit runs on the engine’s per-source committer goroutine, not on the committing goroutine — a source may hold an internal lock across emit without deadlocking the pipeline; watermark persistence merely lags the checkpoint (duplicate delivery on crash, never loss).
• PullSource adds Pull(ctx, emit) error for job pipelines: emit rows synchronously, return nil on exhaustion (the run commits) or an error (the run fails — distinct from per-message dead letters). Sources declaring the "pull" capability SHOULD implement it; a source without PullSource is still job-eligible when Run itself terminates (v1.24: the file source with on_eof: stop works exactly this way — the job run drives Run and its nil return IS the exhaustion signal). The "finite" capability declares “Run can exhaust”; the batch mode’s batch_no_finite_source verify warning keys off it.

Sink

type Sink interface {
    Write(ctx context.Context, msgs []Message) error
    Close() error
}

Batching is engine-owned (§6.4): the sink receives one batch, writes it, returns nil or an error; the engine handles retries, optional drops and dead letters. Success = commit of every message in the batch. There is no Init on sinks — out-of-process sink adapters self-init with their config for that reason (docs/plugins.md).

Codec

Decode(raw []byte) (any, error) and Encode(v any) ([]byte, error). Codecs referenced by bare name (decoder: json) instantiate config-less; named codecs: declarations validate and instantiate at verify with the pipeline directory for relative paths. Declaration names and registered codec names are disjoint namespaces (cfg_codec_shadow otherwise).

Transform

type Transform interface {
    Init(env *TransformEnv) error
    Apply(msg *Message) ([]*Message, error)
    Close() error
}

• Init(TransformEnv) runs once before workers start: the pipeline’s frozen Constants and the run’s Parameters, a node-scoped Logf, and SlowCallWarn — the advisory slow-call threshold for plugins hosting heavyweight engines (the wasm watchdog reads it).
• Apply turns one message into zero or more messages:
  • zero outputs = the message is committed as filtered and counted (eventboat_fanout_no_match_total) — the same semantics as an edge predicate with no matching edge;
  • one or more outputs fan out downstream; the engine expands the commit accounting for the extra branches (the split contract);
  • a non-nil error retries per the incoming edge’s delivery policy, then dead letters — it never fails the node.
• Message ownership (invariant 8). Fan-out sibling branches share the underlying msg.Decoded / msg.Meta maps (the engine shallow-copies the struct per edge), so Apply must never mutate those maps in place — replace the fields wholesale: assign a fresh map/value to msg.Decoded / msg.Meta (the script plugin’s copy-on-write binding does exactly this) or return fresh messages. An in-place write races a sibling branch’s encode and can crash the process with an unrecoverable concurrent-map fault (TestInvariant_BranchIsolation). The same contract is documented on registry.Message and pkg/plugin.
• TransformCloner — implement it when your execution state is not goroutine-safe (wasm module instances die on traps). The engine clones once per worker goroutine and clones must be independent. If Clone fails, the engine treats it as node-fatal: failNode cancels the pipeline and Run returns the error — the master instance is exactly what the plugin declared unsafe to share, so it must not degrade into being shared across workers (internal/engine/nodes.go, runTransform).
• TransformFlavor — return "script" or "wasm" to feed the per-flavor duration histograms and budget/timeout counters; any other flavor (including split’s "split") takes the engine’s generic branch: the execution is still counted by the run accounting and a failure’s kind still becomes the dead-letter class, but no per-flavor instrument exists for it.
• Declare the "explain-safe" capability (script, split) if explain may dry-run your transform on scratch messages; wasm deliberately does not (explain never executes guest code). The instance lifecycle is the build’s choice: a verify-only build (verify, LSP, jobs, testrun) calls Init for validation and closes the instance immediately — failure paths included — while an explain build (verify.Options.ForExplain) retains it and closes it through ir.Pipeline.Close when the walkthrough is done. Explain-safe plugins must therefore tolerate Init/Close once per build and a Close after any validation outcome; the pipeline’s Close is idempotent.

Error classification across plugin boundaries

A plugin cannot raise diagnostics itself, so failures carry classification in registry.TransformError:

The hosts define their own kinds where the failure is created — starhost.ScriptError.Kind (compile/steps/runtime; budget exhaustion is set by the Starlark step-limit hook, not sniffed from "too many steps") and wasmhost.Error.Kind (compile/timeout/guest/trap) — and the built-in adapters map them onto the registry enum: script → compile/steps/runtime, wasm → compile/timeout/guest/runtime (a trap is the guest’s runtime failure). A custom in-process transform just sets Kind (or leaves it empty for other).

internal/ir.addFactoryDiags unwraps TransformError at verify; the engine unwraps Kind/Backtrace at run time (internal/engine/nodes.go). Flavor is not carried on errors: it is an instance property read through TransformFlavor.

Writing a custom in-process transform

For tests and third-party compile-in transforms, internal/testkit provides the fastest path:

package mypipeline_test

import (
    "testing"

    "github.com/eventboat/eventboat/internal/registry"
    "github.com/eventboat/eventboat/internal/testkit"
)

func TestUppercaseTransform(t *testing.T) {
    reg := registry.New()
    // RegisterFakeTransform uses the string-schema API with an
    // accept-anything schema, so the fake works with any plugin block.
    err := testkit.RegisterFakeTransform(reg, "upper", testkit.TransformFunc(
        func(msg *registry.Message) ([]*registry.Message, error) {
            if s, ok := msg.Decoded.(map[string]any); ok {
                s["shout"] = "hi"
            }
            return []*registry.Message{msg}, nil
        }))
    if err != nil {
        t.Fatal(err)
    }
    // ... build a pipeline YAML whose transform node uses `upper: {}`
    // and run it through testrun.RunFile or the engine directly.
}

TransformFunc adapts a plain function to registry.Transform (identity Init/Close); returning a nil slice filters, an error dead-letters after the edge’s delivery retries — both engine contracts, exercised by tests. For the full out-of-process story — gRPC sources/sinks in any language, the JSON handshake, manifests, restart policies — see docs/plugins.md; note that out-of-process gRPC transforms are future work (transforms are registered plugins, but compiled-in only).

Where the builtins live

internal/registry/builtin/register.go wires everything: source-file/cron/http_server/kafka/sql, transform-script/split/fields/wasm, sink-victorialogs/file/http/kafka/drop/debug, codec-json/raw/csv/avro/protobuf. Each registration is a ~30-line typed function — the best templates for writing your own.

4.4 - Configuration & diagnostics

A pipeline is one YAML file loaded by internal/config and compiled by internal/ir. Loading is deliberately strict: unknown keys, unknown plugins and un-compilable expressions are errors, not warnings — the config is a contract with agents as much as with humans. Every finding surfaces as a config.Diagnostic with a stable code, severity, file/line, message and hint (internal/config/config.go); --json output, the LSP, the MCP verify tool and the admin UI all render the same structs.

The load pipeline

LoadBytes(file, data) (internal/config/loader.go) runs four passes:

1. Parse + env substitution. The document parses into a yaml.Node; every string scalar goes through ${VAR} / ${?VAR} substitution (unset ${VAR} is a cfg_env_unset error; unset ${?VAR} omits the key; substituted values are never re-scanned — single-pass, ruling D1).
2. Structural decode into generic maps.
3. Constants substitution. ${constants.x} expands (unknown constant = error); ${parameters.x} passes through untouched only in job pipelines (run.mode: job) — anywhere else it is a cfg_scope_unknown error. The optional marker ? is only legal for plain environment variables.
4. Structural validation with whitelists: top-level keys, metadata.name validation (metadata is a strictly checked mapping: unknown keys and a non-mapping value are errors), the limits/run/parameters/hooks/telemetry/codecs/dlq sections, the three node sections with their framework-field whitelists, and finally manifest reads for every external (grpc:) node. The whitelists, the top-level keys, the edge attributes, the reserved plugin names and the node-level default constants all come from the single leaf package internal/schema (candidate 06) — config, registry and the LSP read one vocabulary, so a name like grpc or version cannot register as a plugin that config would parse as a framework field.

metadata.name is a conservative identifier (cfg_name_invalid): 1–64 characters of [a-zA-Z0-9._-], starting alphanumeric, no .. (path traversal), and no Windows reserved device name (CON, PRN, AUX, NUL, COM1-9, LPT1-9) — the name becomes the deployed file name and the store key.

Pipeline.Order is the YAML document order of node declarations (candidate 06), preserved from the node stream: jobs binds cursor to the first source in that order, explain picks its default entry node from it and the diagnostics iterate it, so a shuffled document still behaves like the one a human reads.

The three-section topology

sources, transforms, sinks — each node is name: {plugin block, ...framework fields}. depends_on edges join them. Sources and sinks are required; the transforms section is optional. Exact node-level whitelists (internal/schema, read by internal/config/sections.go) —

— everything else at node level must be exactly one plugin key. workers is transform-only: sinks.workers was accepted and ignored before and is now rejected with cfg_sink_workers (sink concurrency is engine-owned; implement or refuse). Edge attributes on a depends_on element: when, route, buffer, delivery, required. when accepts a string (CEL) or {lang: cel|cesql, expr: "..."}; route is sugar compiled to meta.route == "<name>" and is mutually exclusive with when. edge_defaults accepts only delivery, required and buffer — a global default predicate is a footgun and a default route is meaningless, so when/route there are cfg_edge_defaults_field errors (candidate 06). Pipeline-level defaults (decoder/encoder json, delivery retries: 3 / backoff: exponential, required: true, buffer max_events: 128, workers 1) are materialized into the typed config at load: downstream layers read the typed field instead of re-applying a default (the json fallback used to be copied in ten places).

Runtime configuration

Deployment-level settings live in a separate kind: Runtime document (internal/runtimecfg/runtimecfg.go), resolved from --runtime, then ./eventboat.yaml, then defaults; CLI flags override. The decode is typed and strict (candidate 06): unknown keys AND type errors are errors (data_dir: 123, enable: "yes", sample_ratio: -1 used to fall through to the defaults), and sample_ratio must be in [0, 1] (0 = tracing off).

Verify and IR resolution

Production code builds pipelines only through internal/verify (candidate 05): verify.File/verify.Bytes load, build and judge in one call and return a Result with the built *ir.Pipeline, the merged diagnostics and the strict verdict; verify.LoadBytes + verify.Build is the two-stage form for consumers with a middle step (jobs substitutes parameters between the stages). ir.Build(cfg, reg, starOpts, parameters) is the build stage it wraps. Per node, in order:

1. Plugin lookup — the plugin key must resolve in the registry (plugin_unknown) or, for external nodes, match its manifest (grpc_manifest_name, grpc_builtin_conflict for compiled-in names).
2. Version pin check — a node version: must equal the registered or manifest version (plugin_version_mismatch); this is the §6.5 rule that makes “documented but absent” impossible.
3. JSON Schema validation — the plugin block validates against the generated/manifest schema (plugin_schema).
4. Factory instantiation — the registry factory builds a real instance at verify time: the script plugin compiles the Starlark program (expr_starlark_compile), the wasm plugin compiles the module and checks ABI exports (expr_wasm_compile), sources/sinks run their typed build functions (cross-field rules JSON Schema cannot express live here).

Transform instances declaring explain-safe are retained on the IR for explain dry-runs — but only in the retaining build (verify.Options.ForExplain, used by the CLI/ops explain entries and replay --dry-run); the default verify-only build validates every instance and closes it immediately, failure paths included, so nothing outlives a verify. An explain caller owns the pipeline and closes the retained instances with ir.Pipeline.Close when the walkthrough is done. Others (wasm — explain never executes guest code) are always closed immediately. Then: codec resolution (codec_unknown, codec_config), order-key compilation, topological sort, job semantics, telemetry pattern checks and lint.

explain renders the resolved semantics (candidate 09), never a per-plugin re-derivation: the sample message is decoded with the entry source’s declared decoder (the same codec instance the engine resolves — a raw/csv source is not fed JSON), each edge predicate is evaluated once per walk, every inbound edge of a sink is rendered with its own delivery policy plus the engine’s mixed-batch rule (max retries, max explicit timeout; the default timeout applies only when no edge in the batch sets one), and a required: false edge shows its terminal drop — the engine drops and commits the message instead of dead-lettering it. A wasm symbolic line shows the resolved per-invoke mode (fast mode (no per-invoke kill switch) when timeout_ms is unset).

The diagnostics table

Every diagnostic code that exists in the code, by emitting layer. Severity error fails verify; warning is lint (upgraded to error by verify --strict). Example messages are abbreviated.

Loader: file, syntax, document shape

Loader: sections

Loader/sections: nodes and edges

IR: topology, plugins, expressions, jobs, lint

Retired codes

Spec v1.19 retired the hand-parsed transform diagnostics — cfg_transform_main_field, cfg_script_type, cfg_split_type, cfg_wasm_type, cfg_wasm_module, cfg_wasm_range, cfg_wasm_allow — their checks now surface as cfg_missing_plugin/cfg_multiple_plugins or plugin_schema issues like every other kind. expr_starlark_compile, expr_wasm_compile and wasm_no_kill_switch survive via error classification (see Plugins).

Where diagnostics are consumed

• verify / run / test CLI verbs print severity[code] file:line: message plus hints; --json emits the structs.
• The LSP converts the same diagnostics into editor squiggles (internal/lsp/diagnostics.go).
• The MCP verify tool returns them with an ok verdict so agents iterate; deploy fails the tool call when verification fails — verify-first has no bypass.

4.5 - Expressions & scripting sandbox

Eventboat hosts three languages, each with one job: CEL is the default edge predicate dialect, CESQL is an opt-in CloudEvents-flavored alternative for edges, and Starlark is the scripting language of the built-in script transform. All three are sandboxed, deterministic and side-effect free — a property explain relies on to dry-run pipelines, and the reason replay produces the same answers as production. The WASM tier covers compute-heavy transforms that outgrow Starlark.

Edge predicates

An edge condition is declared on depends_on:

sinks:
  eu-out:
    depends_on:
      - enrich: { when: 'payload.region == "eu"' }        # string form = CEL
      - warm:
          when: { lang: cesql, expr: "severity >= 5" }    # object form

The string form is CEL. The object form (internal/config/sections.go) takes lang: cel (default) or lang: cesql plus a non-empty expr. Both hosts implement one interface (internal/ir/ir.go, WhenPredicate) with one error contract: an evaluation error means the condition does not pass, plus a counter (eventboat_cel_eval_errors_total) — never a panic, never a silent pass. A fan-out where no edge matches commits the message as filtered.

CEL host

internal/lang/celhost hosts CEL exactly as-is: no custom functions, no syntax extensions (§4.2). The activation binds four dyn-typed variables (celhost.NewEnv / Predicate.evalValue):

The same env compiles sink order_key expressions (result must be a string).

The cost limit. Every predicate program is built with cel.CostLimit(1_000_000) (maxEvalCostUnits). The rationale from the code: the runtime cost model charges ~1 per arithmetic/logical step, per element for comprehensions, and input length × 0.1 for size-driven operations (equality/regex/contains), so a realistic predicate over a KB-scale message lands in the low hundreds; 1e6 leaves an order of magnitude of headroom while cancelling payload-driven blowups. Exceeding the limit surfaces as an ordinary EvalError (“operation cancelled: actual cost limit exceeded”) — so a too-expensive predicate makes the edge not match and increments eventboat_cel_eval_errors_total; it never aborts the message. Compile errors (CompileError) are verify findings instead.

CESQL dialect

internal/lang/cesqlhost hosts the opt-in CESQL dialect (§4.7) by reusing the official CloudEvents SDK parser (cloudevents/sdk-go/sql/v2); the official TCK is vendored (275 cases) and must stay green in CI. Dialect rules, per the code:

• Identifiers are alphanumeric only — CESQL’s lexer has no . token and no underscore, mirroring CloudEvents attribute names. Meta keys containing underscores (kafka_offset, message_id) are therefore unreachable in this dialect; use CEL for those.
• data.* payload extension is a string-literal-aware pre-parse rewrite to synthetic camelCase identifiers: data.amount → dataAmount, data.a.b → dataAB (rewriteDataPaths). Nested objects flatten up to depth 4; identifiers starting with data (case-insensitive) are reserved for the extension. Nulls are not injected (they read as missing attributes); arrays and non-scalar leaves become JSON strings.
• Meta maps to context attributes: id, source, type, subject, time route through the event writer (required context attributes); everything else becomes an extension. Values map string→string, bool→bool, integral numbers→int32, other numbers→decimal string, arrays/objects→JSON string.

The extension mode is “CESQL subset + documented extension” — never advertised as full CESQL compatibility.

Starlark sandbox

internal/lang/starhost runs transforms.<node>.script — a statement sequence bound against payload, meta, constants, parameters. Guarantees, all visible in starhost.go:

• Syntax restrictions (fileOptions()): syntax.FileOptions is built with Set: true, TopLevelControl: true (scripts are statement sequences, not function bodies — review R2), and While: false, Recursion: false, GlobalReassign: false. No while loops, no recursion, no global reassignment.
• Step budget: thread.SetMaxExecutionSteps(100_000) (DefaultOptions). Exhaustion kills the execution; the host types it at the step-limit hook (starhost.KindSteps — never sniffed from the "too many steps" text) and the engine counts it (eventboat_script_step_budget_exhausted_total). A script that exceeds the budget dead letters after the incoming edge’s retries, with the steps failure kind as its dead-letter class.
• Load allowlist: exactly json and math (allowedModules); there is no loadable strings module in go-starlark — string methods are built into the string type. No time, no I/O, no entropy: evaluation is deterministic and replayable.
• Predeclared bindings: payload, meta, constants, parameters (frozen), plus safe_json_decode(s, default) — the one sanctioned escape hatch, returning the default instead of aborting on bad JSON — and remove(dict, key).
• Per-message Thread: each execution gets a fresh starlark.Thread (RunWithParams); payload/meta bind lazily with copy-on-write semantics, so a failed attempt’s writes never leak into a retry.
• Precise dirty tracking (beta round): mutations mark through a marker threaded into the converted dicts, so only written subtrees materialize back into the message. Container reads no longer dirty; list-bearing trees dirty conservatively (native Starlark list mutators cannot be intercepted — the boundary is locked by a dedicated test in internal/lang/starhost, and a future list wrapper would flip exactly that test). After Apply, only dirty states are written back to the message (internal/registry/builtin/transform_script.go).
• fail() behavior: Starlark has no exceptions to swallow; a runtime failure becomes ScriptError{Msg, Backtrace, Line} (asScriptError — the innermost frame carries the user-visible line). The engine dead letters the message with the backtrace verbatim; positions render as script:L:C. Compiled programs are immutable, so one program serves all workers concurrently (script registers without TransformCloner).

WASM transforms

The third tier (internal/wasmhost, guest ABI in docs/wasm.md) runs wasip1 reactor modules under wazero. Guests build with the standard Go toolchain (GOOS=wasip1 go build -buildmode=c-shared); the ABI is _initialize, eb_alloc(len), transform(ptr, len) (plus optional eb_last_error). Wire format: payload as JSON bytes in/out; metadata passes through untouched (chain a script: node to touch meta).

Resource model — wazero has no instruction metering, so budgets are:

• Memory cap: max_memory_pages, default 512 pages = 32 MiB (DefaultMaxMemoryPages), always on.
• Wall-clock budget: timeout_ms is opt-in. When set, wazero’s CloseOnContextDone kills runaway guests — measured at ~5x throughput cost on loop-heavy guests. When unset (fast mode, the default) there is no kill switch: a runaway guest wedges its one worker until the pipeline restarts. This is a documented tradeoff (M3-audit J2 ruling): a tier whose reason to exist is performance cannot default to slower than Starlark; the risk is availability, not correctness — no data is lost and the eight invariants hold. Guardrails: the wasm_no_kill_switch verify warning (escalated by --strict), and the zero-interference slow-call watchdog (engine WasmSlowCallWarnMs, default 5000ms) logs a wedged invoke once.
• Traps/timeouts in protected mode kill the instance, which is transparently re-instantiated per worker (TransformCloner — module instances are not goroutine-safe). explain does not dry-run wasm guests.

Which layer for which job

• Edge predicate (when:) for routing decisions: cheap, per-edge, evaluated before delivery. CEL first; CESQL only for CloudEvents interop or teams that already speak it.
• script: transform for mapping, enrichment, filtering-by-computation and routing via meta.route. The default choice — the ladder’s trigger standard (§4.6) is performance or dependencies, never “complex logic”.
• wasm: transform only when the ladder justifies it: measured heavy per-message computation (the shipped benchmark: heavy aggregation 2.5ms WASM vs 5.9ms Starlark, ~30,000x fewer allocations in fast mode) or a dependency that only exists as a guest. For light mapping Starlark wins outright.

Scratch evaluation: repl and explain

Two tools run the same hosts outside a live pipeline:

• eventboat repl --message sample.json --cel 'payload.total > 100' (or --script fix.star) evaluates one expression/script against a sample; the interactive loop re-executes the accumulated script against the original sample each line — deterministic session semantics, the §4.3 guarantee.
• eventboat explain --message sample.json really executes scripts and evaluates each outgoing edge against the transformed payload (the sandbox is deterministic and side-effect free, review R10), so the walkthrough shows the answer production would give. A failing script renders its backtrace and the dead-letter consequence. Without --message, nothing executes (symbolic summary only).

The deterministic-replay caveat

The sandboxes are deterministic on purpose: no clocks, no randomness, no I/O in Starlark; WASI provides fake clocks and no filesystem. This is what lets explain --message execute scripts for real, lets repl re-execute a session deterministically, and makes contract tests reproducible. The caveat runs the other way too: anything nondeterministic you need must enter through the message — stamps like ingest_time in meta, or parameters resolved at trigger time — because the pipeline itself will never draw entropy at run time, and a script that tried to (no such builtin exists today) would break the replay guarantees the docs make.

4.6 - Observability & operations

Eventboat exposes three operations surfaces that share one implementation: the MCP tools, the Admin REST endpoints (plus SSE and an embedded read-only UI), and the --json CLI outputs are all thin shells over internal/ops.Service — REST, MCP and CLI emit the same Go structs and JSON shapes. Telemetry is OpenTelemetry with dual export (Prometheus pull + OTLP/HTTP push); the metric set is fixed and enumerated in internal/obs/obs.go.

Metrics

One MeterProvider, two readers (internal/obs/obs.go): Prometheus exposition at /metrics and, when telemetry.otlp_endpoint is set, an OTLP/HTTP push reader (10s interval). Fully disabled telemetry costs nothing (noop providers; every Obs helper is nil-receiver safe). Every instrument carries the eventboat_ prefix (the static set below, plus the per-source counters created lazily on first use):

Counters

reason_class is the dead-letter class the engine recorded with the message (store.DeadLetter.Class) — decode, codec, encode, delivery, canceled, or a transform failure kind (steps, timeout, guest, compile, runtime, other). It is never re-derived from the reason text: candidate 07 deleted the prefix-matching obs.ReasonClass, so the label values changed ("script" became the failure kind, "encoder" became codec, and an "encode:" failure is now encode, never other).

Histograms (unit: seconds)

Gauges

Gauges are pushed on every ops.Status() snapshot (the SSE stream and UI poll status; /metrics scrapes the recorded values).

Tracing

• One span per job run: eventboat.job.run, attributes pipeline/run_id/trigger, ended with the terminal status and recorded errors (internal/jobs). Script backtraces ride dead-letter records and span error events.
• Per-message spans are opt-in: telemetry.span_sample_rate (default 0 = none, zero cost; 1 = all). Roots are named eventboat.message with pipeline/message_id/source/spool_seq attributes and end at a terminal state — eventboat.terminal_state = committed or dead_letter (internal/engine/engine.go, startMessageSpan/finishSpan). Rationale (review R16): no per-message spans at routing rates. The span context is deliberately not propagated through the DAG — correlation rides the attributes.
• Trace export goes through OTLP/HTTP when telemetry.otlp_endpoint is configured, sampled by telemetry.sample_ratio (default 0.1); otherwise the global noop tracer.

OTel wiring

obs.Setup(ctx, cfg) builds the providers from the Runtime config (internal/runtimecfg): telemetry.prometheus (default true) registers the Prometheus reader served by the admin listener at /metrics; telemetry.otlp_endpoint adds the OTLP metric reader (push every 10s) and the trace exporter. Single-pipeline eventboat run gets OTLP only (no HTTP surface); run --config-dir and mcp --http serve both readers.

The admin surface

internal/admin serves the whole surface on one mux — Admin REST + SSE + UI, /metrics, /mcp — behind one security middleware. Auth: an optional bearer token resolved as --admin-token flag > EVENTBOAT_ADMIN_TOKEN env

admin.token in the Runtime config (internal/cli/mcp.go, internal/runtimecfg). With a token set, every request needs Authorization: Bearer <token> (the ?token= query form exists only for the UI’s EventSource and first navigation — it then appears in browser history/proxy logs; use the header from curl/agents). A non-loopback admin.listen refuses to start without a token — the write surface can deploy pipeline YAML whose grpc plugin command: executes on the host (internal/admin/security.go). Loopback binds additionally enforce a Host allowlist (DNS-rebinding defense); wildcard binds skip it and rely on the mandatory token. JSON request bodies are capped at 8 MiB (maxBodyBytes, internal/admin/admin.go).

The admin UI is a single embedded HTML page (internal/admin/ui.go): read-only status console fed by the SSE stream; on a 401 it serves a sign-in prompt that verifies the token against /admin/status.json and keeps it in sessionStorage.

The same operations exist as 14 MCP tools (internal/mcpserver): catalog, verify, test, explain, status, jobs, tail, dlq_query, deploy, trigger, dlq_replay, drain, pause, resume — over stdio (eventboat mcp --stdio) or HTTP (--http). verify returns the same composition the CLI runs (candidate 05: one diagnostic sequence, one strict verdict), and explain accepts at — the entry node behaves identically on CLI, MCP and Admin.

Log tailing

ops.Tail samples the most recent deliveries per sink node (bounded ring of 100 entries, payloads truncated at 512 bytes, is_replay flagged) via a sink wrapper installed on every managed engine. Entries show the payload document as delivered (msg.Out, falling back to Raw). Consume it via GET /admin/tail/{node}, the MCP tail tool, or the UI.

Redaction

The pipeline-level telemetry.redact section lists dot-separated field paths where every segment is a path.Match glob (payload.user.email, meta.authorization). Compilation and matching live in internal/ops/redact.go. Coverage is deliberately presentation-only:

• Tail entries are masked ("***") before the 512-byte truncation;
• DLQ display masks both roots — payload patterns against the raw document, meta.* patterns against the meta map — before letters cross the admin REST or MCP dlq_query surface (ops.redactDeadLetters).

The stored rows stay raw: the spool, dead letters and deliveries are the data path and are never altered, so DeadLetterReplay re-injects the original bytes. Non-JSON tails pass through unmasked. A redact pattern that cannot compile is a verify error (telemetry_redact_pattern) — a bad pattern must not silently never match.

Engine counters (the non-OTel layer)

Beside the OTel instruments, the engine keeps atomic counters in engine.Metrics (internal/engine/engine.go): MessagesIn, CommittedCount (split from the checkpoint pointer, review R5), CheckpointPtr, DeadLettered, CelEvalErrors, NoMatch, Retries, DlqFailures, OptionalDrops, DecodeErrors, TransformRuns, Backpressured, SpoolFailures. These are the M1 bookkeeping layer and the source of the --json CLI/status output — the OTel instruments record the same events for Prometheus/OTLP consumers. ops.Status() reads them per pipeline (messages_in, committed, dead_lettered, checkpoint, in_flight, messages_per_sec as a delta over the last snapshot window).

Run outcomes and exit codes

Every runner derives its terminal state from one engine decision point (candidate 02): Engine.Wait classifies the run — completed (quiesced cleanly), partial (quiesced with dead letters), failed (the engine stopped itself: worker-fatal or a source failure) or interrupted (the caller canceled) — and each process maps that onto its own contract:

The exit-code split follows the process shape, not the status: a one-shot run that did not complete exits non-zero, while a long-lived process treats SIGTERM as a graceful stop. A failed source stops the engine in every mode (no silent “alive but not consuming”); restart resumes from the source watermarks — duplicate delivery, never loss. The daemon’s failed status carries the outcome’s failure text (worker-fatal first, then the first source error); a batch that fails no longer sits at running with an error.

SSE event types

GET /admin/sse emits hello on connect, then typed events — deploy (pipeline deployed, replaced instance), job (job run state), status (pause/resume/drain and a full status snapshot every second, which is also what pushes the gauges). Slow consumers drop events; snapshots repeat.

HTTP server posture

The admin listener (admin.Serve) sets ReadHeaderTimeout 10s, ReadTimeout 60s, IdleTimeout 120s — and a deliberately loose WriteTimeout (15m) because SSE responses are long-lived streams; it caps their lifetime instead of erroring mid-request (the UI’s EventSource reconnects).

Operations runbook pointers

• Status polling: GET /admin/status.json or watch /admin/sse; job runs via GET /admin/jobs/{pipeline} (the status embeds the 5 most recent runs, surfacing run:<status> for job pipelines).
• A wedged WASM worker (fast mode) shows as one transform node stalled with a slow-call watchdog log line; restart clears it.
• Backpressure is visible as eventboat_backpressure_events_total climbing with eventboat_in_flight_messages pinned at the high watermark.
• Checkpoint lag is eventboat_spool_depth; a growing lag with a failing store indicates checkpoint writes failing (replay window widening — the engine keeps committing in memory and retries durability).

4.7 - Testing guide

Eventboat’s testing philosophy: every reliability property is a named, retrievable test; every externally visible surface (help screens, plugin schemas) is pinned by a golden file; and anything that claims to work against the real world (Kafka, long runs) is proven against the real thing in CI. The full suite is go test ./...; CI runs it with -race.

The test taxonomy

The eight engine invariants

internal/engine/invariants_test.go pins redesign-v3.md §6.2 — these names are load-bearing; do not rename them:

• TestInvariant_SpoolBeforeVisible — a failing spool append refuses the message; it never reaches a sink.
• TestInvariant_CheckpointAdvancesOnlyAfterCommit — the checkpoint trails committed messages.
• TestInvariant_Kill9ReplayReplaysAllUncommitted — crash recovery replays a superset of the uncommitted set.
• TestInvariant_DeadLetterWriteFailureBlocksCommit — a failing dead-letter write never drops the message.
• TestInvariant_RequiredFalseFailureDoesNotBlockSiblingBranches — optional edge failure isolates to its branch.
• TestInvariant_RedeliveryKeepsMessageIdStable — duplicate delivery keeps meta.message_id, so idempotent sinks can dedup.
• TestInvariant_CursorWatermarkNeverExceedsCommitted — pull-source watermarks only advance over committed messages.
• TestInvariant_BranchIsolation — fan-out siblings share the underlying msg.Decoded/msg.Meta maps, so a transform mutating them in place (the script binding’s lazy remove(), pre-fix) must never be able to race a sibling branch’s encode: the copy-on-write binding isolates the branches, and the process must survive a wide-map fan-out under -race.

The -race policy: the engine and store packages must stay race-clean — CI runs go test -race ./... on every push. If you add concurrency to either package, run the race detector locally first; a data race here is a correctness bug in the durability model, not a hygiene issue.

Golden tests

Three golden surfaces, each with its own regenerate flag:

• CLI help snapshots: internal/cli/testdata/help/*.txt pin every verb’s help screen. After changing a flag or usage string: go test ./internal/cli -update and commit the diff.
• Plugin schema goldens: internal/registry/builtin/testdata/schemas/*.json pin the JSON Schema generated from every builtin config struct. After changing a config struct: go test ./internal/registry/builtin -update-schemas. The test failure text tells you the flag.
• Protobuf descriptor golden: internal/registry/builtin/testdata/example.descr, regenerated with -update-descr (the protobuf codec test’s fixture; no protoc involved).

A golden diff in a PR must be reviewable line by line — that is the point.

Integration tests (need Docker)

Both suites skip silently without their env var; CI sets it:

# real Kafka via testcontainers (KRaft, confluent-local):
# roundtrip through the engine, dead-letter path, consumer-group rebalance
EVENTBOAT_KAFKA_TEST=1 go test ./internal/inttests/kafka/ -v -timeout 8m

# long-run stability: mixed load + injected faults; asserts at-least-once,
# full checkpoint advance and no goroutine leaks
EVENTBOAT_SOAK_TEST=1 EVENTBOAT_SOAK_DURATION=2m go test ./internal/inttests/soak/

EVENTBOAT_SOAK_DURATION bounds the soak run (CI default 25m; nightly plus manual dispatch via .github/workflows/soak.yml). The wasm tests optionally honor EVENTBOAT_WASM_GUEST — CI rebuilds the guest from source and points the env at the fresh artifact; locally the committed .wasm under internal/wasmhost/testdata/ is used, so plain go test ./... needs no wasm toolchain.

Benchmarks

bash scripts/bench-gate.sh   # what CI enforces

scripts/bench-gate.sh is a loose order-of-magnitude gate, not a noise gate — limits sit ~5–15x above the reference dev machine (i5-14600KF) to absorb shared-runner variance:

It covers the per-message hot paths (predicate eval, script exec, commit throughput against the memory store) and deliberately does not cover WASM (runner variance too high — BenchmarkHeavyTransform in internal/wasmhost stays informational) or SQLite-backed throughput. What a gate failure means: an algorithmic regression landed; profile before optimizing.

testkit and testrun

internal/testkit provides the deterministic primitives engine and contract tests build on:

• ManualSource — a source whose Emit blocks until the engine accepted the message (spooled + dispatched), making tests deterministic; Commit records the cursor watermark like a real pull source would.
• CaptureSink/Recorder — capture delivered messages; DiscardSink; FlakySink — fault-inject sink writes by attempt count.
• StoreWrapper — hook AppendSpool/WriteDeadLetter/SetCheckpoint to inject store faults.
• FixedClock/CounterID — frozen clock and deterministic IDs.
• RegisterFakeTransform/TransformFunc — register a function as a transform plugin with an accept-anything schema (see Plugins).

internal/testrun is the contract-test runner: it builds the pipeline IR, runs each case against a real engine with an ephemeral memory store, fixed clock, capture-wrapped sinks (the plugin sink never touches the real world), injects at the named node, waits for commit, and evaluates expectations (in-order subset matching on payload.*/meta.* dotted paths, DLQ count/reason). Job pipelines run with sources disabled so injections at the pull node are deterministic.

The suite format

The project convention is <root>/pipeline.yaml + <root>/tests/<suite>.yaml (+ tests/fixtures/); the suite references its pipeline as pipeline: ../pipeline.yaml (path resolution is contained within the suite root — an escaping path is an error):

suite: order-branching
pipeline: ../pipeline.yaml
cases:
  - name: eu-order-routed-to-eu-only
    inject: { at: ingest, messages: [fixtures/eu-order.json] }
    expect:
      capture:
        at: eu-out
        messages:
          - payload.total: 12000     # subset match
            meta.tier: vip
  - name: malformed-json-to-dlq
    inject: { at: ingest, raw: "{not json" }
    expect:
      dlq: { count: 1, reason_contains: decode }

eventboat test <path> consumes it: a file recurses (.yaml files without a top-level suite: key are skipped and counted; unreadable/unparseable YAML is a hard error, ruling D3). Exit codes: failing cases → 1; a directory with no valid suites but broken files → 2 (ruling D4).

Practical recipes

Run the fast loop (seconds, no Docker):

go build ./... && go test ./... 

Run everything locally before a PR:

go test -race ./...                          # full suite, race on (engine+store policy)
bash scripts/bench-gate.sh                   # perf gate
EVENTBOAT_KAFKA_TEST=1 go test ./internal/inttests/kafka/   # needs Docker

Add a regression test for a delivery bug: reproduce with the smallest pipeline YAML (see invYAML in invariants_test.go for the minimal shape), drive it with testkit.ManualSource + CaptureSink + StoreWrapper fault hooks, assert on Recorder.Captured() or the store’s dead letters. If it touches commit/checkpoint ordering, check whether one of the eight invariants already covers the class — extend that test’s scenario instead of asserting something weaker.

Add a contract test for a user-visible pipeline behavior: write the suite under examples/<pipeline>/tests/, fixtures under tests/fixtures/, and rely on the CI examples gate (internal/cli/examples_test.go) to keep it green.

4.8 - Building & release

Eventboat builds with the standard Go toolchain and no code generation step in the normal path: go build ./... is the whole story for contributors. This page covers the local build, the container image, the CI workflows, and how releases are cut.

Prerequisites

• Go 1.25 (matches go.mod; the CI jobs pin the same version).
• Docker — only for the env-gated integration suites (the Kafka testcontainers test) and for building/running the container image.
• Nothing else: all drivers are pure Go (SQLite via modernc.org/sqlite, Postgres via pgx, Kafka via kafka-go), so CGO is never required.

Local build and run

go build -o eventboat ./cmd/eventboat     # the single binary
./eventboat                               # bare invocation prints help, exit 0
./eventboat verify --config examples/linear/pipeline.yaml
./eventboat test examples
./eventboat run --config examples/linear/pipeline.yaml
./eventboat run --config my.yaml --ephemeral   # in-memory store, no ./data

The example job pipeline’s sqlite source database regenerates with go run ./examples/job-sync/seed. The WASM test guest is committed as a built artifact (internal/wasmhost/testdata/aggregate.wasm), so plain go test ./... needs no wasm toolchain; rebuild it with GOOS=wasip1 GOARCH=wasm go build -buildmode=c-shared -o ../aggregate.wasm . from internal/wasmhost/testdata/guest (CI does exactly this).

The container image

The Dockerfile is two stages:

1. Build: golang:1.25-alpine, CGO_ENABLED=0 go build -trimpath — the pure-Go dependency set is what makes the static build possible.
2. Runtime: gcr.io/distroless/static-debian12:nonroot, binary at /usr/local/bin/eventboat, running as nonroot (uid 65532).

/pipelines and /data are the conventional mount points (config dir and SQLite volume — examples/k8s/deployment.yaml mounts exactly these; .keep placeholders carry the nonroot-owned directories through COPY and keep unmounted smoke runs working). No CMD: a bare docker run prints the help screen and exits 0.

docker build -t eventboat:dev .
docker run --rm -v "$PWD/examples/linear:/work" -w /work eventboat:dev \
  run --config /work/pipeline.yaml

Development data layout

Where the entry points put things on disk (all under --data-dir, default data; --ephemeral replaces the SQLite store with a cached in-memory one):

The canonical layout is data/stores/<sanitized pipeline>.db (candidate 04): the one-shot verbs (trigger, replay, jobs, run --config) and the daemon (run --config-dir, mcp --http) all open the SAME per-pipeline file, so a run history written by trigger is the one jobs list and the daemon’s Status read. The name is sanitized to a conservative charset and checked against the Windows reserved device names (internal/fsname, the same rules the loader applies). The pre-candidate-04 layouts — data/eventboat.db and data/stores/pipeline.db — are retired without migration (beta ruling): the old files are simply no longer read.

The per-pipeline name is the store file namespace, which is why metadata.name is validated so strictly (cfg_name_invalid).

Regenerating protocol code

pkg/pluginproto is generated from proto/eventboat/plugin/v1/plugin.proto. Regenerate it only with the pinned protoc-gen-go toolchain the existing stubs were built with (see the git history for the toolchain version), and expect a matching ABI bump ripple: external plugins import the generated stubs, so a regeneration that changes the surface means external plugins must be rebuilt — record that in the CHANGELOG entry.

GHCR publishing

.github/workflows/docker.yml builds linux/amd64 + linux/arm64 (QEMU + buildx; CGO off makes cross-builds plain Go cross-compilation) and pushes ghcr.io/eventboat/eventboat:

There is deliberately no moving 0.x tag while pre-1.0: a floating 0.2 must not silently point at an rc of a later patch line.

CI workflows

.github/workflows/ci.yml runs on every push to main and every PR:

• test — build, vet, rebuild the WASM guest from source, then go test -race ./... (includes the eight invariants, the examples verify/test gate, the gRPC plugin acceptance, the CESQL TCK, the wasm tests; the env-gated integration packages skip without their env).
• lint — golangci-lint v2 (v2.13.2) with the repo config (.golangci.yml): standard set + misspell + gofmt; exclusions are justified in the config.
• kafka-integration — the real-broker suite (EVENTBOAT_KAFKA_TEST=1, testcontainers).
• bench — scripts/bench-gate.sh (the loose threshold gate) plus the informational Starlark-vs-WASM benchmark.

.github/workflows/soak.yml — nightly (02:30 UTC) plus manual dispatch: the soak suite under mixed load with injected faults, duration controlled by the workflow input (default 25m).

The docs site is published by a GitHub Pages workflow (.github/workflows/pages.yml): it builds and deploys the site from docs/ changes — content pages and the generator — so docs edits go live without a release. The rendered result lives at eventboat.dev/eventboat/ (the project-pages subpath under the organization domain).

The public docs site (eventboat.dev, the eventboat/eventboat.github.io Hugo repo) consumes this repo’s developer guides at ITS build time: its Pages workflow checks out eventboat/eventboat and runs tools/sitegen -hugo-out to render docs/developer/*.md into Hugo content under content/docs/developer/, then builds Hugo as usual — plus a daily scheduled rebuild to pick up upstream doc changes. Nothing is copied into that repo; this repo stays the single source of truth, so a guide edit here is live there after the next org-site build (push-triggered or daily).

The editor story

The LSP (eventboat lsp) works with any editor; a minimal VS Code launcher lives in examples/editors/vscode — it spawns the binary over stdio and wires diagnostics, completion and hover through. There is no marketplace extension; the example is the integration.

Release process

• CHANGELOG discipline: every user-visible change lands under ## Unreleased in CHANGELOG.md at the time of the commit (Keep a Changelog format; the repo’s entries carry the rationale, not just the diff summary). Look at the git log — commits like the Settled→Commit rename ship with their full old→new mapping in the entry.
• Cutting a release: consolidate Unreleased into a dated version section, tag v<semver>, push the tag. The tag triggers docker.yml, which publishes :<version> to GHCR. Pre-releases use rc suffixes following semver (v0.1.0-beta, v0.2.0-rc1 in the tag history); dependency bumps or contract changes that would break plugins ride an rc first (the v0.2.0-rc1 release bumped lynx-go/commands for a contract fix).
• The binary is the release artifact for non-container users: CGO_ENABLED=0 go build -trimpath -o eventboat ./cmd/eventboat on the supported targets (linux, macOS, the BSDs, Windows); the only platform-specific code paths are the store lease’s file locks (flock/LockFileEx), and a target without a lock implementation refuses to run a pipeline loudly rather than skipping the single-writer guarantee.

tools/sitegen — the docs site generator

tools/sitegen (in progress) is the documentation site generator: a separate Go module under tools/, so the main module’s dependency tree stays untouched. Run it with go run ./tools/sitegen from within tools/. It renders the Markdown under docs/ into a static site with no CDN dependencies — no external JS or fonts — because the site must render anywhere, including air-gapped environments. Frontmatter (title, order) drives page titles and ordering; relative links within docs/ resolve as site links; code blocks are plain ASCII-safe fences.

4.9 - Contributing

This repo has a strong house style — it is what keeps a codebase with this many guarantees reviewable. Read this page before your first PR; most of it is enforceable by CI, but the commit message convention and the comment voice are reviewed by humans.

Dev workflow

• Branch from main; keep one logical change per commit — the git log reads as a series of self-contained rationales, and bisectability depends on it.
• Run the fast loop before pushing: go build ./..., go test -race ./... (CI runs the race build on every push), and bash scripts/bench-gate.sh if you touched a hot path.
• User-visible changes get a CHANGELOG.md Unreleased entry at commit time, not at release time. Entries carry a Changed/Added/Fixed/ Removed heading and the same paragraph voice as the commit — the Settled→Commit rename entry, for example, includes the complete old→new mapping because external consumers needed it to migrate.

The commit message convention

This repo’s commits are single, detailed, paragraph-led messages written in the imperative that explain WHAT and WHY, with concrete file and identifier references. No one-line subjects, no bullet laundry lists (a short list is acceptable inside the paragraph when enumerating parallel changes). Read a few from git log — e.g. the Rename Settled → Commit everywhere commit or the Promote transforms to registered plugins commit — they record the motivation, the ruling that decided contested points, the migration/compat story, and the verification done (which tests were run). A good test: could a reviewer reconstruct the change from the message alone in a year?

Rules of thumb distilled from the log:

• First sentence = the change, imperative (“Add …”, “Fix …”, “Remove …”, “Rename …”).
• Then the why, naming the ruling or review finding that decided it (e.g. “review R6”, “M3-audit J2 ruling”) — see Design discussion below.
• Then the mechanics, with real file paths and identifiers (internal/engine/commit.go, RegisterTransformT, plugin_schema).
• End with what was verified (“full suite passes incl. -race on the engine”).

Code conventions

• gofmt and golangci-lint clean. .golangci.yml is deliberately small: the standard set (errcheck, govet, ineffassign, staticcheck, unused) plus misspell and gofmt — a hygiene floor, not a style machine. Exclusions (generated stubs in pkg/pluginproto, the vendored CESQL TCK, a quickfix class) must be justified in the config; new linters join with a ruling.

• Comment voice: comments state constraints and rationale, never narrate the obvious. A representative example, internal/registry/builtin/kafka.go:

  Commit runs on every frontier advance (~per message), so the scan must start at the watermark, not at 1: the scan itself deletes every seq it visits, which keeps each call O(new work) instead of O(all emissions).

  The comment gives the constraint (called per message), the decision (scan from the watermark), and the consequence (O(new work)). “increments the counter” comments do not exist here. If a comment only restates the code, delete it; if a constraint lives only in a review document, move the essential sentence into the code.

• The full-name principle for YAML vocabulary. Config keys are full words in snake_case — max_in_flight, drain_timeout, skip_if_successful, catchup_window, span_sample_rate — never abbreviations. The single recorded exception is dlq (retained industry-wide, spec v1.6). New keys follow the same rule; if you are tempted to abbreviate, the key is probably too long.

• Error wrapping with %w so classification survives: the engine unwraps registry.TransformError with errors.As, and store/registry errors are matched by identity, not string comparison. Wrap, never re-format, an error you pass along.

The “when you touch X you must also Y” checklist

Adding a builtin plugin

The checklist for a new compiled-in plugin (source, sink, codec or transform) — every step has a test that fails until you do it:

1. Define one config struct: json tags name the keys, schema tags the constraints (Plugin system).
2. Write the register function (typed: RegisterSourceT/RegisterSinkT/ RegisterCodecT/RegisterTransformT) and wire it into RegisterAll (internal/registry/builtin/register.go). Cross-field rules JSON Schema cannot express stay in the build function.
3. Regenerate the schema goldens (go test ./internal/registry/builtin -update-schemas) and review the diff.
4. Catalog surfaces are automatic — plugin catalog, plugin schema, the MCP catalog tool and the LSP completion read the registry — but their tests may need the new entry.
5. Codecs add a conformance case (codecs_conformance_test.go); the protobuf codec’s descriptor fixture regenerates with -update-descr.
6. Remember the name rules: the plugin name must not collide with framework fields (registration rejects it) and should follow the full-name principle below.

PR expectations

• Full suite green (go test ./...), with -race on the engine and store packages (CI does this; run it locally first).
• The examples gate stays green: every pipeline under examples/ must verify and pass its contract suites (internal/cli/examples_test.go runs it in CI).
• Golden diffs (help screens, plugin schemas) present and reviewed line by line — a schema golden diff that “just regenerates” without discussion is a red flag.
• New public surface (CLI verb, metric, config key, diagnostic code) follows the checklist above.

Where design discussion lives

redesign-v3.md is the spec and the single source of truth. Design discussion has a discipline: contested points are resolved as numbered rulings recorded in the spec’s revision log (and echoed in README decision ledgers), citing the review round that raised them (redesign-v3-review*.md). If your change settles a design question, add the ruling — reviewers should be able to trace every non-obvious decision to a dated entry. Dated review documents keep their historical wording even after later renames (the Settled→Commit rename explicitly left them alone).

Licensing

Eventboat is Apache-2.0 (LICENSE). By contributing you agree your contributions are licensed under the same terms. Keep the generated protocol stubs in pkg/pluginproto regenerable from proto/eventboat/plugin/v1/plugin.proto — do not hand-edit them.