# Architecture & package map

LLMS index: [llms.txt](/llms.txt)

---

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](../03-plugins/)); edges carry predicates (CEL or CESQL) and
delivery policies (retries, backoff, timeout, required/optional)
([Configuration](../04-config-pipeline/)).

## 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`.

| Package | Owns | Must not reach into |
|---|---|---|
| `internal/config` | Typed pipeline config, the strict loader (`LoadBytes`/`LoadBytesIn`), `${VAR}`/`${?VAR}`/`${constants.x}` substitution, `metadata.name` validation, sections/edges/hooks/parameters parsing, declaration order and materialized defaults | The registry or IR; plugin validity is decided later. It imports only the leaf packages `schema`/`fsname` |
| `internal/schema` | The single source of the framework vocabulary (candidate 06; renamed from `internal/framework`): per-section node fields, top-level keys, edge attributes, reserved plugin names, node-level default constants. config, registry and the LSP read it | Pipelines, plugins or parsing — a leaf with static lists only |
| `internal/ir` | The static IR: DAG nodes/edges, compiled CEL/CESQL predicates and Starlark programs, topology checks, plugin + codec resolution, lint, job semantics. `ir.Build` *is* the build stage (with the retaining `BuildForExplain` for explain, released through `Pipeline.Close` — candidate 09) — reached only through `internal/verify` in production code | `engine`, `store`, anything runtime. IR is built before an engine exists |
| `internal/verify` | The one verify-first composition (candidate 05): `File`/`Bytes` → a `Result` carrying the built IR, the merged diagnostics and the strict verdict; the two-stage `LoadBytes` → (parameters) → `Build` form; the explicit `baseDir` rule for relative paths | Running anything — it stops at the built IR (`ops`, `cli`, `lsp`, `jobs`, `testrun` enter it) |
| `internal/engine` | Spool admission, DAG execution, commit tracking, per-edge delivery retries, dead lettering, checkpointing, backpressure, transform/sink workers | Config parsing or expression compilation — it consumes a built `ir.Pipeline` only. It also never decodes payloads itself (codecs do) |
| `internal/registry` | The plugin registration model: four kinds, JSON Schema validation (santhosh-tekuri), version pins, the typed struct→schema generator, `Catalog`, and the reserved-name gate shared with config through `internal/schema` | Any host package. A leaf — plugins register *into* it; it imports only `internal/schema` |
| `internal/registry/builtin` | All compiled-in plugins: file/cron/http_server/kafka/sql sources; script/split/wasm transforms; file/http/kafka/drop sinks; json/raw/csv/avro/protobuf codecs | Anything beyond `registry` and `wasmhost` (for the wasm transform config) |
| `internal/lang/celhost` | CEL predicate host: env binding, compile, cost-limited eval | Nothing internal (leaf). It does not know about pipelines |
| `internal/lang/cesqlhost` | The CESQL edge dialect (official CloudEvents parser + the `data.*` rewrite) | Nothing internal (leaf) |
| `internal/lang/starhost` | The Starlark sandbox host: compile, frozen constants, lazy COW message bindings, step budget, backtraces | Nothing internal (leaf) |
| `internal/wasmhost` | wazero runtime host: wasip1 reactor compile (compiled modules cached by file identity + compile-affecting config, candidate 09), per-invoke budgets through the one `ResolveMode` resolution (fast mode default), guest ABI check, invokers | Nothing internal (leaf). `registry/builtin` adapts it |
| `internal/rpcplugin` | Out-of-process gRPC source/sink plugins: spawn, JSON handshake, auth metadata, restart supervision | `ir`/`engine`. It adapts `config.GrpcConfig` + `pkg/pluginproto` into registry-shaped sources/sinks |
| `internal/fsname` | The shared file-name rules: `Sanitize` (conservative base-name charset) and `WindowsReservedName` — used by config's `metadata.name` validation and the store owner's layout | Nothing internal — a leaf. One reserved-name list, no copies |
| `internal/store` | The durable spine, split into facets — `SpoolStore` (spool/checkpoint/source states), `DeadLetterStore`, `JobRunStore`, combined as `Store` — backed by SQLite (`modernc.org/sqlite`) and an in-memory implementation, plus the `Owner` that decides where a pipeline's store lives (`<data-dir>/stores/<sanitized name>.db`) and owns its handle lifetime (one cached handle per pipeline, closed together; memory owner for tests/`--ephemeral`) | Parsing or execution. It persists `registry.Message` but never interprets payloads |
| `internal/jobs` | Job pipelines: cron scheduling, catchup, overlap, run lifecycle, typed parameters (`cursor`/`now`), hooks, run history | Being a *caller* of the engine only — it must not reimplement admission or commit logic |
| `internal/ops` | The operations service: verify/test/explain/deploy/status/jobs/trigger/tail/dlq_query/dlq_replay/drain/pause/resume. The single implementation behind MCP and Admin REST | HTTP or protocol concerns. `admin`/`mcpserver`/CLI are thin shells over it |
| `internal/admin` | Admin REST + SSE + the embedded read-only UI + the security middleware (token, Host allowlist) | Business logic — everything delegates to `ops` |
| `internal/obs` | OpenTelemetry: one MeterProvider with Prometheus + OTLP readers, the instrument set, span helpers | Nothing internal (leaf); callers pass in pipeline/node names |
| `internal/explain` | Deterministic walkthroughs: symbolic and message-level traces (resolved semantics — shared wasm mode, per-edge delivery + batch rule, optional drop, decoder-aware sample), mermaid/ASCII topology | `engine` — explain runs compiled IR, never a live engine; it dry-runs the retained `ForExplain` instances, and the caller closes them (`Pipeline.Close`, candidate 09) |
| `internal/lsp` | The language server: minimal hand-written JSON-RPC 2.0, diagnostics from the real verify path (the document's directory is the relative-path base), completion, hover | Execution — it validates documents through `internal/verify`, never runs pipelines |
| `internal/mcpserver` | The 14 MCP tools over stdio or Streamable HTTP (official Go SDK) | Business logic — thin shells over `ops` |
| `internal/dlq` | The dead-letter strategy surface (candidate 05): where-filter compilation with the pipeline's constants, selection order (select before delete), the codec-carrying replay request | Transports — `ops` injects into a live engine, the CLI replay builds a local engine; both assemble the shared `Request` |
| `internal/runtimecfg` | Deployment-level `kind: Runtime` config: `storage.*`, `admin.*`, `mcp.*`, `telemetry.*` keys | Pipeline concerns; it never loads pipeline files |
| `internal/testkit` | Injection/capture/fault-injection primitives: `ManualSource`, `CaptureSink`, `FlakySink`, `StoreWrapper`, `RegisterFakeTransform` | Production behavior — nothing in `internal/` (outside tests) may import it |
| `internal/testrun` | The §3.2 contract-test runner: suite YAML → in-process real-engine runs with capture sinks | Only the public engine/testkit surfaces; it is itself consumed by `ops` and the CLI `test` verb |
| `internal/inttests` | Env-gated integration suites: `kafka/` (real broker via testcontainers) and `soak/` (long-run stability) | — test-only; skipped without their env vars |
| `pkg/pluginproto` | Generated gRPC stubs for `eventboat.plugin.v1` | Generated code — regenerate, never hand-edit. It lives in `pkg/` (not `internal/`) so third-party plugin modules can import it |
| `pkg/plugin` | The compile-time plugin ABI: aliases of the engine's plugin interfaces plus typed `RegisterSource`/`RegisterTransform`/`RegisterSink`/`RegisterCodec` into the process-wide registry | Importing the engine — plugin packages import only this, so their own test binaries stay light |
| root `package eventboat` | `RunCLI` — the library entry point for custom builds: blank-import plugin packages, then run the full CLI under your own `main` (Benthos's `service.RunCLI` role) | Business logic — it delegates to `internal/cli` |
| `internal/cli` | The CLI verb table and cmdX executors (verify/test/run/trigger/jobs/explain/replay/repl/lsp/plugin/mcp), dispatch on lynx-go/commands, help goldens; shared by `cmd/eventboat` and the root package's `RunCLI` | New business logic — verbs delegate to `ops`, `testrun`, `explain`; and no `os.Args`/`os.Exit` — `Run` returns the exit code, the caller owns exit |
| `cmd/eventboat` | The shipped binary: a thin `main` over `internal/cli` | Anything else |
| `tools/` | Support tools. `tools/sitegen` (in progress) is the documentation site generator: a separate Go module, run with `go run ./tools/sitegen`, no CDN dependencies | Importing `internal/` — it is a separate module |
| `examples/` | Shipped pipelines (linear, branching, fan-in, job-sync, codecs), the third-party gRPC plugin example (`examples/plugins/ticker-source`, its own Go module), the VS Code launcher, the k8s manifest | — covered by the CI examples gate: all examples are verified and contract-tested |

## 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](../06-observability/)). 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](../02-engine/) — the accept path, commit frontier,
  persistence and recovery in detail.
- [Plugin system & registry](../03-plugins/) — how to register, write and test
  plugins.
- [Configuration & diagnostics](../04-config-pipeline/) — the load pipeline and
  every diagnostic code.
- [Testing guide](../07-testing/) — where each guarantee is pinned.