# Plugin system & registry

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

---

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

| Kind | Interface | Factory signature |
|---|---|---|
| `sources` | `Source` (+ optional `PullSource`) | `func(cfg map[string]any) (Source, error)` |
| `transforms` | `Transform` (+ optional `TransformCloner`, `TransformFlavor`) | `func(cfg any, dir string) (Transform, error)` |
| `sinks` | `Sink` | `func(cfg map[string]any) (Sink, error)` |
| `codecs` | `Codec` | `func(cfg map[string]any, dir string) (Codec, error)` |

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):

```yaml
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:

```go
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

```go
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

```go
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

```go
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`:

| Field | Meaning |
|---|---|
| `Err` | the underlying error |
| `DiagCode` | at **verify** time, routes a factory failure to a specific diagnostic code (empty falls back to `plugin_schema`) — `expr_starlark_compile` and `expr_wasm_compile` survive this way, with a `Hint` |
| `Backtrace` | at **run** time, stored verbatim in the dead letter (Starlark backtraces; positions render as `script:L:C`) |
| `Kind` | the typed **failure kind** — `steps`, `timeout`, `guest`, `compile`, `runtime` or `other` (`registry.FailureKind`). It becomes the dead-letter class and drives the budget/timeout counters; classification is **never** derived from the message text |

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:

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