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.