conduit

package module
v0.19.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 24, 2026 License: Apache-2.0 Imports: 14 Imported by: 0

README

Conduit

Logo

Data Integration for Production Data Stores. 💫

scarf pixel License Test Go Report Card Discord Twitter Go Reference Conduit docs API docs

Overview

Conduit is a data streaming tool written in Go. It aims to provide the best user experience for building and running real-time data pipelines. Conduit comes with common connectors, processors and observability data out of the box.

Conduit pipelines are built out of simple building blocks which run in their own goroutines and are connected using Go channels. This makes Conduit pipelines incredibly performant on multi-core machines. Conduit guarantees the order of received records won't change, it also takes care of consistency by propagating acknowledgements to the start of the pipeline only when a record is successfully processed on all destinations.

Conduit connectors are plugins that communicate with Conduit via a gRPC interface. This means that plugins can be written in any language as long as they conform to the required interface.

Conduit was created and open-sourced by Meroxa.

Quick start

The fastest way to see Conduit working — scaffold and run a demo pipeline with a single command:

conduit quickstart

Sample records flow to your console within seconds. State is in-memory and nothing is written to your working directory; press Ctrl-C to stop. For a full walkthrough, see https://conduitdata.io/docs/getting-started.

Installation guide

Download binary and run

Download a pre-built binary from the latest release and simply run it!

./conduit run

Once you see that the service is running, the configured pipeline should start processing records automatically. You can also interact with the Conduit API directly, we recommend navigating to http://localhost:8080/openapi and exploring the HTTP API through Swagger UI.

Conduit can be configured through command line parameters. To view the full list of available options, run ./conduit run --help or see configuring Conduit.

Homebrew

Make sure you have homebrew installed on your machine, then run:

brew update
brew install conduit
Debian

Download the right .deb file for your machine architecture from the latest release, then run:

dpkg -i conduit_0.17.0_Linux_x86_64.deb
RPM

Download the right .rpm file for your machine architecture from the latest release, then run:

rpm -i conduit_0.17.0_Linux_x86_64.rpm
Build from source

Requirements:

git clone [email protected]:ConduitIO/conduit.git
cd conduit
make
./conduit run
Docker

Our Docker images are hosted on GitHub's Container Registry. To run the latest Conduit version, you should run the following command:

docker run -p 8080:8080 conduit.docker.scarf.sh/conduitio/conduit:latest

Preflight checks

Before running Conduit for the first time (or after changing configuration), run conduit doctor to check whether your environment is ready:

conduit doctor

It runs a set of offline, non-destructive checks — config resolution and validation, database reachability, API address availability, plugin directories, the built-in plugin registry, and whether a running engine is reachable — and reports pass/warn/fail for each, grouped by category. It does not start a Runtime; this is distinct from the running server's /readyz and /healthz endpoints, which answer "is the running engine serving?" instead of "would conduit run succeed here?".

Useful flags: --json for machine-readable output, --check <name> (repeatable) to run only specific checks, --deep to additionally verify standalone connector plugin binaries in an isolated subprocess, --require-server to fail instead of warn when no Conduit server is reachable, and -q/--quiet to only print warnings, failures, and the summary. See conduit doctor --help for the full list of checks and exit codes, and the design doc for the rationale.

Configuring Conduit

Conduit accepts CLI flags, environment variables and a configuration file to configure its behavior. Each CLI flag has a corresponding environment variable and a corresponding field in the configuration file. Conduit uses the value for each configuration option based on the following priorities:

  • CLI flags (highest priority) - if a CLI flag is provided it will always be respected, regardless of the environment variable or configuration file. To see a full list of available flags run conduit run --help.

  • Environment variables (lower priority) - an environment variable is only used if no CLI flag is provided for the same option. Environment variables have the prefix CONDUIT and contain underscores instead of dots and hyphens (e.g. the flag -db.postgres.connection-string corresponds to CONDUIT_DB_POSTGRES_CONNECTION_STRING).

  • Configuration file (lowest priority) - By default, Conduit loads a configuration file named conduit.yaml located in the same directory as the Conduit binary. You can customize the directory path to this file using the CLI flag --config.path. The configuration file is optional, as any value specified within it can be overridden by an environment variable or a CLI flag.

    The file must be a YAML document, and keys can be hierarchically structured using .. For example:

    db:
      type: postgres # corresponds to flag -db.type and env variable CONDUIT_DB_TYPE
      postgres:
        connection-string: postgres://localhost:5432/conduitdb # -db.postgres.connection-string or CONDUIT_DB_POSTGRES_CONNECTION_STRING
    

    To generate a configuration file with default values, use: conduit init --path <directory where conduit.yaml will be created>.

This parsing configuration is provided thanks to our own CLI library ecdysis, which builds on top of Cobra and uses Viper under the hood.

Storage

Conduit's own data (information about pipelines, connectors, etc.) can be stored in the following databases:

  • BadgerDB (default)
  • PostgreSQL
  • SQLite

It's also possible to store all the data in memory, which is sometimes useful for development purposes.

The database type used can be configured with the db.type parameter (through any of the configuration options in Conduit). For example, the CLI flag to use a PostgreSQL database with Conduit is as follows: -db.type=postgres.

Changing database parameters (e.g. the PostgreSQL connection string) is done through parameters of the following form: db.<db type>.<parameter name>. For example, the CLI flag to use a PostgreSQL instance listening on localhost:5432 would be: -db.postgres.connection-string=postgres://localhost:5432/conduitdb.

The full example in our case would be:

./conduit run -db.type=postgres -db.postgres.connection-string="postgresql://localhost:5432/conduitdb"

Connectors

For the full list of available connectors, see the Connector List. If there's a connector that you're looking for that isn't available in Conduit, please file an issue .

The quickest way to add a connector is the signed connector registry: conduit connectors install <name> — see the registry install guide.

Conduit loads standalone connectors at startup. The connector binaries need to be placed in the connectors directory relative to the Conduit binary so Conduit can find them. Alternatively, the path to the standalone connectors can be adjusted using the CLI flag -connectors.path.

Conduit ships with a number of built-in connectors:

  • File connector provides a source/destination to read/write a local file (useful for quickly trying out Conduit without additional setup).
  • Kafka connector provides a source/destination for Apache Kafka.
  • Postgres connector provides a source/destination for PostgreSQL.
  • S3 connector provides a source/destination for AWS S3.
  • Generator connector provides a source which generates random data (useful for testing).
  • Log connector provides a destination which logs all records (useful for testing).

Additionally, we have prepared a Kafka Connect wrapper that allows you to run any Apache Kafka Connect connector as part of a Conduit pipeline.

If you are interested in writing a connector yourself, have a look at our Go Connector SDK. Since standalone connectors communicate with Conduit through gRPC they can be written in virtually any programming language, as long as the connector follows the Conduit Connector Protocol.

Processors

A processor is a component that operates on a single record that flows through a pipeline. It can either change the record (i.e. transform it) or filter it out based on some criteria.

Conduit provides a number of built-in processors, which can be used to manipulate fields, send requests to HTTP endpoints, and more, check built-in processors for the list of built-in processors and documentations.

Conduit also provides the ability to write your own standalone processor, or you can use the built-in processor custom.javascript to write custom processors in JavaScript.

More detailed information as well as examples can be found in the Processors documentation.

API

Conduit exposes a gRPC API and an HTTP API.

The gRPC API is by default running on port 8084. You can define a custom address using the CLI flag -api.grpc.address. To learn more about the gRPC API please have a look at the protobuf file.

The HTTP API is by default running on port 8080. You can define a custom address using the CLI flag -api.http.address. It is generated using gRPC gateway and is thus providing the same functionality as the gRPC API. To learn more about the HTTP API please have a look at the API documentation, OpenAPI definition or run Conduit and navigate to http://localhost:8080/openapi to open a Swagger UI which makes it easy to try it out.

Embed Conduit in your Go app

You can run a full Conduit engine in-process, inside your own Go application, and drive its lifecycle with ordinary function calls — no subprocess, no socket. The engine that powers the CLI is the one you embed.

go get github.com/conduitio/conduit
e, _ := conduit.New(ctx, conduit.Options{DB: conduit.DBOptions{Type: "badger"}})
h, _ := e.Run(ctx)
defer h.Stop(ctx)

_ = e.ImportPipeline(ctx, conduit.NewPipeline("hello").
    WithConnector(conduit.NewSourceConnector("src", "builtin:generator")).
    WithConnector(conduit.NewDestinationConnector("dst", "builtin:log")))
_ = e.StartPipeline(ctx, "hello")

The exported API (Options, Engine, Handle, New, and the pipelines-in-code builder) is a semver-committed, frozen import path. See the Embedding Conduit (Go) guide for the full lifecycle, options (logger/metrics injection, storage), deployment and state considerations, and known limitations. Non-Go bindings (Python, Node) are planned as gRPC client libraries — see the roadmap.

Exit codes

Every conduit CLI invocation — one-shot commands (conduit pipelines list, ...) and the long-running conduit run — exits with one of a small set of deterministic codes, so scripts and agents can branch on failure kind without parsing error text:

Code Meaning Examples
0 Success The command succeeded, or conduit run shut down gracefully on a single SIGINT/SIGTERM.
1 Runtime error An internal or unclassified error (a bug, an unexpected failure).
2 Validation The request or config was rejected: invalid argument, not found, already exists, failed precondition.
3 Environment A required external dependency is unreachable: the server, the database, a rate limit, or an already-bound listen address.

conduit run treats a second SIGINT/SIGTERM (received after the first one has already started a graceful shutdown) as a forced kill and exits with the POSIX 128+signum convention instead (SIGINT130, SIGTERM143) — the same value a shell reports for a process killed by that signal, so a forced kill is distinguishable from an ordinary classified exit code.

A multi-result command like conduit doctor (many checks, one process) reduces its checks to a single exit code by taking the worst bucket across every failing check — a database that can't be reached (environment, 3) always outranks an invalid config field (validation, 2) in the final exit code, regardless of which check ran first. Warnings never affect the exit code.

See ADR: deterministic CLI exit codes for the full mapping and rationale, pkg/conduit/exitcode for the implementation, and pkg/conduit/check for the multi-result aggregation doctor (and future multi-check commands) share.

Validating pipeline configs

conduit pipelines validate <path> (alias: conduit pipeline validate) statically checks one pipeline config file, or every .yml/.yaml file in a directory (not recursed), without starting Conduit or contacting a running instance — it's fully offline, so it's safe to run in CI or before conduit run with no server anywhere nearby.

$ conduit pipelines validate pipelines/orders.yaml
✗ pipelines/orders.yaml
✗ config.field_required   /connectors/0/plugin
    connector "orders:postgres": "plugin" is mandatory
    → set connectors[0].plugin (e.g. "builtin:postgres")

Summary: 1 files · 0 passed · 1 failed · 1 problems
Fix the ✗ items above, then re-run.

(the connector's ID is shown enriched — <pipelineID>:<connectorID> — since validation runs after the same default-enrichment step conduit run applies; --json's configPath still points at your original connectors[0].)

Every problem in every file is reported — a bad file never stops the rest from being checked. Exits 0 if every pipeline is valid, 2 otherwise (see Exit codes above). Add --json for the structured {command, ok, summary, result, error} envelope, or -q/--quiet to suppress passing () lines and show only failures and the summary.

lint (advisory warnings, e.g. deprecated fields) and dry-run (shows the enriched pipeline graph and optionally resolves builtin plugin references) share the same engine and are planned as follow-ups — see the design doc.

Documentation

To learn more about how to use Conduit visit Conduit.io/docs.

If you are interested in internals of Conduit we have prepared some technical documentation:

llms.txt

llms.txt and llms-full.txt at the repo root are a machine-readable index for agents and LLM tools, generated straight from the pipeline config schema, the built-in connector registry, the stable error-code registry, and the conduit mcp tool catalog — see cmd/conduit/internal/llmsgen. They are never hand-edited: make generate regenerates both, and CI fails the build if the committed files drift from source.

Contributing

For a complete guide to contributing to Conduit, see the contribution guide.

We welcome you to join the community and contribute to Conduit to make it better! When something does not work as intended please check if there is already an issue that describes your problem, otherwise please open an issue and let us know. When you are not sure how to do something please open a discussion or hit us up on Discord.

We also value contributions in the form of pull requests. When opening a PR please ensure:

  • You have followed the Code Guidelines.
  • There is no other pull request for the same update/change.
  • You have written unit tests.
  • You have made sure that the PR is of reasonable size and can be easily reviewed.

Documentation

Overview

Package conduit is Conduit's embeddable Go API — the frozen, semver-committed import path a Go application uses to run a Conduit engine in-process, without touching os.Exit, a global logger, or the process-wide default Prometheus registry.

Scope (v1 / "B1" DX-hardened + "B2")

This package covers the engine lifecycle: New validates Options and constructs an Engine, but opens nothing — the database (and every service built on it) is opened lazily, on whichever of Engine.Run or Engine.Import is called first (see ensureRuntime, and Engine's "Lifecycle contract" doc). Run starts the engine and returns a *Handle once it is ready to accept work; Handle.Stop drains and shuts it down; Engine.Close releases whatever was actually opened, whether or not Run was ever called, and is a safe no-op if nothing was — see Engine's "Lifecycle contract" doc for the exact New/Run/Import/Stop/Close state matrix. Engine.Import lets a host create or update a pipeline from a PipelineConfig value.

PipelineConfig values don't have to come from YAML: NewPipeline starts a fluent, pipelines-in-code builder (PipelineBuilder, with ConnectorBuilder/ProcessorBuilder/DLQBuilder for nested entities) that produces the exact same PipelineConfig a YAML parse of the equivalent document produces — see builder.go and this package's round-trip tests (TestPipelineBuilder_RoundTrip, TestPipelineBuilder_RoundTrip_Fixtures) for the property this is verified against. Engine.ImportPipeline builds and imports in one call, so the common case never needs a separate provisioning/config import:

err = e.ImportPipeline(ctx, conduit.NewPipeline("hello").
	WithConnector(conduit.NewSourceConnector("src", "builtin:generator")).
	WithConnector(conduit.NewDestinationConnector("dst", "builtin:log")))

Language bindings (Python, Node) are a later workstream, delivered as gRPC client libraries rather than a C ABI — see docs/design-documents/20260724-embed-grpc-client-libraries.md.

The database open was eager in the original B1 merge; a DX audit (zero external adoption yet, so the right time to fix a frozen-contract mistake) found this made a New'd-and-discarded Engine leak a Badger file lock or a Postgres/SQLite pool unless Close was called even when the caller never ran anything. It is lazy as of this fast-follow — see New's doc and engineLeakCheckFinalizer for the GC-time safety net that remains for any discard-without-Close path this doesn't otherwise catch.

Why this exists, not pkg/conduit directly

pkg/conduit (docs/package_structure.md) is documented internal: its NewRuntime/Entrypoint.Serve API prints a splash to os.Stdout, calls os.Exit on error, writes logs unconditionally to os.Stdout, and registers metrics into the process-global Prometheus DefaultRegisterer — all correct defaults for a CLI process, all wrong for a library an application embeds alongside its own logging and metrics. This package wraps pkg/conduit with a handful of additive seams (see pkg/conduit.RuntimeOption) so the CLI's behavior is byte-for-byte unchanged while an embedder gets full control: every log line goes through a host-supplied *slog.Logger (or slog.Default(), an explicit, documented fallback — never os.Stdout), every metric goes through a host-supplied prometheus.Registerer (nil disables metrics entirely — never the default registry), and every failure is returned as an error, never an os.Exit or a panic.

Invariants

  • Invariant 7 (graceful shutdown): Handle.Stop reuses Runtime.Run's existing ctx-cancellation-driven drain (registerCleanup, dispatching to registerCleanupV1/V2 — unchanged by this package) unchanged. Stop supplies a different *trigger* (an explicit call cancelling Run's context) for the identical mechanism `conduit run`'s SIGTERM handling already exercises via Entrypoint.CancelOnInterrupt — it does not reimplement or duplicate drain logic.
  • Invariants 1-6 (ack/position/ordering/checkpoint/schema): not implicated. This package does not touch the record data path. Engine.Import delegates to provisioning.Service.Import unchanged; PipelineConfig is a type alias (not a copy) of the exact struct the YAML provisioner produces, so no new serialized shape is introduced.

Known limitation: metrics cross-talk across engines

Two Engines in the same process, each with its own MetricsRegisterer, are genuinely isolated as distinct prometheus.Registerer/Registry *objects* — but pkg/foundation/metrics keeps process-global metric *definitions*, so a metric defined via that package's package-level constructors (everything in pkg/foundation/metrics/measure) still fans its *values* out to every registry ever registered in the process, including a second Engine's. This is a known, accepted limitation for v1 (not fixed by this package), tracked as a follow-up; see pkg/conduit.WithMetricsRegisterer's doc and TestTwoEngines_MetricsCrossTalk_KnownLimitation.

Two sharper instances of the same root cause are tracked separately, also accepted for v1: a failed metrics registration during ensureRuntime (first Run or Import, not New — see this package's lazy-open fast-follow above) still leaks a registry into pkg/foundation/metrics' process-global bookkeeping (https://github.com/ConduitIO/conduit/issues/2669, see pkg/conduit.configureEmbeddedMetrics' doc), and the HTTP /metrics route serves promclient.DefaultGatherer rather than a host-supplied MetricsRegisterer (https://github.com/ConduitIO/conduit/issues/2670, see pkg/conduit.(*Runtime).newHTTPMetricsHandler's doc).

Package boundary / deprecation policy

This package's exported API (Options, Engine, Handle, PipelineConfig, New, NewPipeline/PipelineBuilder and its ConnectorBuilder/ProcessorBuilder/ DLQBuilder companions) is a public contract, versioned like the connector protocol, pipeline config schema, and error codes: a breaking change is announced (CHANGELOG + a `Deprecated:` godoc comment) in one monthly release, kept working with a warning for at least one more minor release, and removed no earlier than the third minor release after announcement. PipelineConfig extends this policy by name to provisioning/config.{Pipeline,Connector,Processor,DLQ} — all four are constrained by the single doc comment above provisioning/config.Pipeline in that package's parser.go (Connector, Processor, and DLQ have no separate per-type comment of their own; they are covered by name in that one shared comment, not individually annotated).

Example (HelloPipeline)

Example_helloPipeline demonstrates the three-call embed lifecycle (New -> Run -> Stop) plus ImportPipeline, end to end: an in-memory database, a generator source, and a log destination — the pipeline defined with NewPipeline, not parsed from YAML. This is the literal example a `go get github.com/conduitio/conduit` embedder copies first.

package main

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"time"

	conduit "github.com/conduitio/conduit"
)

// helloPipeline runs the full New->Run->ImportPipeline->Stop lifecycle with
// a single generator->log pipeline, entirely in-memory, defining the
// pipeline in code via NewPipeline/ImportPipeline rather than parsing YAML —
// the common case that needs no provisioning/config import at all. It is
// the shared logic behind Example_helloPipeline (compiled+run by
// `go test`, checked against captured Output) and
// TestExampleHelloPipeline_WithinBudget (a CI-timed wall-clock budget
// assertion — deliberately separate from the human-facing "15 minutes to a
// running pipeline" doc claim, which is a UX statement, not something a
// test can measure; see the embed workstream plan §12 open question 1).
//
// New below does not open anything (B1 DX-hardening fix: the database opens
// lazily, on this very Run call — see New's and Engine's "Lifecycle contract"
// doc); Run's own success-then-Stop path already releases it via Runtime's
// cleanup, which is why this example never calls Engine.Close explicitly —
// see Close's doc for when it would still be needed (e.g. if Run below had
// failed, or if the engine were discarded before ever being run).
func helloPipeline(ctx context.Context, dir string) error {
	e, err := conduit.New(ctx, conduit.Options{
		Logger:       slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})),
		DB:           conduit.DBOptions{Type: "inmemory"},
		PipelinesDir: dir,
	})
	if err != nil {
		return err
	}

	h, err := e.Run(ctx)
	if err != nil {
		return err
	}

	err = e.ImportPipeline(ctx, conduit.NewPipeline("hello-pipeline").
		WithName("hello-pipeline").
		WithConnector(
			conduit.NewSourceConnector("src", "builtin:generator").
				WithSetting("format.type", "raw").
				WithSetting("recordCount", "5"),
		).
		WithConnector(conduit.NewDestinationConnector("dst", "builtin:log")),
	)
	if err != nil {
		_ = h.Stop(ctx)
		return err
	}

	if err := e.StartPipeline(ctx, "hello-pipeline"); err != nil {
		_ = h.Stop(ctx)
		return err
	}

	time.Sleep(50 * time.Millisecond)

	return h.Stop(ctx)
}

func main() {
	ctx := context.Background()
	dir, err := os.MkdirTemp("", "conduit-embed-example-*")
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	defer os.RemoveAll(dir)

	if err := helloPipeline(ctx, dir); err != nil {
		fmt.Println("error:", err)
		return
	}
	fmt.Println("pipeline stopped cleanly")
}
Output:
pipeline stopped cleanly

Index

Examples

Constants

View Source
const (
	// StatusRunning is a pipeline's default desired status, applied by
	// Engine.Import's enrichment when WithStatus is never called.
	StatusRunning = provisioningconfig.StatusRunning
	// StatusStopped marks a pipeline as provisioned but not started.
	StatusStopped = provisioningconfig.StatusStopped

	// ConnectorTypeSource marks a connector as a pipeline source.
	ConnectorTypeSource = provisioningconfig.TypeSource
	// ConnectorTypeDestination marks a connector as a pipeline destination.
	ConnectorTypeDestination = provisioningconfig.TypeDestination
)

Pipeline/connector field-value constants, re-exported so a builder-only caller never needs to import provisioning/config directly just to spell "running" or "source" correctly. Each is the same value as provisioning/config's own constant (validate.go) — not a new one — so a literal string and the constant are always interchangeable.

Variables

This section is empty.

Functions

This section is empty.

Types

type APIOptions

type APIOptions struct {
	// Enabled turns on Conduit's HTTP and gRPC API. Default: false.
	Enabled bool
	// GRPCAddress is the bind address for the gRPC API (e.g. ":8084", or
	// "127.0.0.1:0" for an OS-assigned port). Required if Enabled.
	GRPCAddress string
	// HTTPAddress is the bind address for the HTTP API/gateway. Required if
	// Enabled.
	HTTPAddress string
}

APIOptions configures Options.API.

type ConnectorBuilder

type ConnectorBuilder struct {
	// contains filtered or unexported fields
}

ConnectorBuilder builds one connector (source or destination) to attach to a PipelineBuilder. Obtain one with NewSourceConnector or NewDestinationConnector — there is deliberately no bare NewConnector constructor, so a connector's type can never be left empty or misspelled.

func NewDestinationConnector

func NewDestinationConnector(id, plugin string) *ConnectorBuilder

NewDestinationConnector starts a ConnectorBuilder for a destination connector with the given id and plugin.

func NewSourceConnector

func NewSourceConnector(id, plugin string) *ConnectorBuilder

NewSourceConnector starts a ConnectorBuilder for a source connector with the given id and plugin (e.g. "builtin:postgres").

func (*ConnectorBuilder) WithName

func (c *ConnectorBuilder) WithName(name string) *ConnectorBuilder

WithName sets the connector's display name. Leave unset to have Engine.Import's enrichment default it to the connector's id.

func (*ConnectorBuilder) WithProcessor

func (c *ConnectorBuilder) WithProcessor(p *ProcessorBuilder) *ConnectorBuilder

WithProcessor appends a processor scoped to this connector only. Call order is preserved. A nil p is recorded as a Build-time error (surfaced once this connector is attached to a PipelineBuilder and Build is called) rather than panicking — see PipelineBuilder's doc.

func (*ConnectorBuilder) WithSetting

func (c *ConnectorBuilder) WithSetting(key, value string) *ConnectorBuilder

WithSetting sets a single connector configuration key. A later call with the same key overwrites the earlier value.

func (*ConnectorBuilder) WithSettings

func (c *ConnectorBuilder) WithSettings(settings map[string]string) *ConnectorBuilder

WithSettings merges settings into the connector's configuration — the same as calling WithSetting for every entry. Keys already set and not present in settings are left untouched.

type DBOptions

type DBOptions struct {
	// Driver, when set, is used directly and takes precedence over Type and
	// the type-specific fields below — the same precedence
	// pkg/conduit.Config gives a caller-supplied database.DB.
	Driver database.DB

	// Type selects the storage backend: pkgconduit.DBTypeBadger,
	// DBTypePostgres, DBTypeInMemory, or DBTypeSQLite. Defaults to
	// DBTypeInMemory when empty.
	Type string

	Badger struct {
		Path string
	}
	Postgres struct {
		ConnectionString string
		Table            string
	}
	SQLite struct {
		Path  string
		Table string
	}
}

DBOptions configures Options.DB.

type DLQBuilder

type DLQBuilder struct {
	// contains filtered or unexported fields
}

DLQBuilder builds a pipeline's dead-letter-queue configuration. Obtain one with NewDLQ and attach it with PipelineBuilder.WithDLQ.

WindowSize and WindowNackThreshold are *int fields on the underlying config.DLQ specifically so "never set" (nil, matching an omitted YAML window-size/window-nack-threshold key — Engine.Import's enrichment applies Conduit's own default) is distinguishable from "explicitly set to zero" (a non-nil pointer to 0). WithWindowSize/WithWindowNackThreshold are the only way to set either field, and only ever produce the explicit-value form — there is no way to construct the nil form other than never calling them, matching the enrichment path's own semantics exactly.

func NewDLQ

func NewDLQ(plugin string) *DLQBuilder

NewDLQ starts a DLQBuilder for the given DLQ plugin (e.g. "builtin:log").

func (*DLQBuilder) WithSetting

func (d *DLQBuilder) WithSetting(key, value string) *DLQBuilder

WithSetting sets a single DLQ configuration key. A later call with the same key overwrites the earlier value.

func (*DLQBuilder) WithSettings

func (d *DLQBuilder) WithSettings(settings map[string]string) *DLQBuilder

WithSettings merges settings into the DLQ's configuration — the same as calling WithSetting for every entry. Keys already set and not present in settings are left untouched.

func (*DLQBuilder) WithWindowNackThreshold

func (d *DLQBuilder) WithWindowNackThreshold(threshold int) *DLQBuilder

WithWindowNackThreshold sets the DLQ's nack threshold explicitly, including to 0 — see DLQBuilder's doc on the nil-vs-explicit-zero distinction.

func (*DLQBuilder) WithWindowSize

func (d *DLQBuilder) WithWindowSize(size int) *DLQBuilder

WithWindowSize sets the DLQ's nack window size explicitly, including to 0 — see DLQBuilder's doc on the nil-vs-explicit-zero distinction.

type Engine

type Engine struct {
	// contains filtered or unexported fields
}

Engine is a constructed, not-yet-running Conduit instance. Obtain one with New; start it with Run.

Lifecycle contract

New only validates Options — it does not open Options.DB, dial anything, or construct pkg/conduit's Runtime. That happens lazily, in ensureRuntime, the first time either Run or Import is called (B1 DX hardening fix: v1 opened the database eagerly in New, so a New'd-and-discarded Engine — constructed speculatively, or abandoned after a later precondition check failed — leaked a Badger file lock or a Postgres/SQLite pool for the rest of the process's life unless Close was called). Concretely:

  • New → discard (Run and Import never called): nothing was ever opened. Calling Close is still safe but no longer required to avoid a leak — there is nothing to release.
  • New → Close (Run/Import never called): same as above; Close observes that ensureRuntime never ran and returns nil without touching anything.
  • New → Run or Import (opens the database) → Close: Close releases the database ensureRuntime opened.
  • New → Run (fails before Ready, including the already-done-context guard) → Close: ensureRuntime already opened the database before Runtime.Run's guard trips (started still flips true on the Run call, see below), and it was never handed to Runtime's own cleanup path (registerCleanup) — Close is still required to release it.
  • New → Run (succeeds) → Stop → Close: Stop's drain already closed the database via Runtime's cleanup path. Close is safe to call anyway — it is idempotent with that path, not merely with itself — and returns nil.
  • Close is safe to call more than once from the same or different goroutines; every call after the first observes the same result.
  • New → Close → Run or Import (no prior Run/Import call — ensureRuntime never ran before Close): ensureRuntime checks the closed flag before ever opening anything, so this never silently opens a fresh runtime the caller believed was already released. It returns a coded error (conduiterr.CodeInvalidArgument) instead.

Safety net: if an Engine that DID open resources (Run or Import actually ran) is garbage-collected without Close ever being called, a runtime.SetFinalizer hook logs a leak warning through Options.Logger (or slog.Default() if none was supplied) instead of leaking silently — see engineLeakCheckFinalizer. This is belt-and-suspenders, not a substitute for Close: a finalizer's timing is unpredictable (GC-dependent, and finalizers are not guaranteed to run at all before process exit), and it deliberately only logs — it never closes the database itself, since running arbitrary cleanup from a finalizer goroutine with no ordering guarantee relative to the rest of the program is its own footgun.

started flips to true the moment Run is *called*, not when it succeeds: a failed Run permanently retires this Engine for running (a second Run attempt — even after a failure — returns the same conduiterr.CodeInvalidArgument "called more than once" error Run's own doc describes). Close is unaffected by started and may be called in any state.

func New

func New(_ context.Context, opts Options) (*Engine, error)

New validates opts and returns a not-yet-running Engine. It does not open Options.DB, dial anything, or construct pkg/conduit's Runtime — that is deferred to the first call to Run or Import (see ensureRuntime and Engine's "Lifecycle contract" doc). ctx is not used beyond this call (there is nothing left to bound: validation does no I/O); Run's and Import's own ctx bounds the later database dial.

Every failure is returned as an error — a *conduiterr.ConduitError where classifiable — never an os.Exit and never a panic on a caller-supplied misconfiguration.

func (*Engine) Close

func (e *Engine) Close(_ context.Context) error

Close releases whatever ensureRuntime actually opened — currently, the configured database.DB — regardless of whether Run was ever called on this Engine. See Engine's "Lifecycle contract" doc for the states Close must cover.

Close is a safe no-op if nothing was ever opened: an Engine that never had Run or Import called on it holds no resources at all (Options.DB opens lazily — see New's and ensureRuntime's doc), so there is nothing to release and Close returns nil without touching anything.

Close is idempotent: it is safe to call more than once, from any goroutine, and safe to call after a successful Run→Stop cycle (where the database was already released via Runtime's own cleanup path) — every call after the first returns the same result. When resources were opened, it delegates to Runtime.CloseDB, whose sync.Once is what actually makes double-release across both paths safe; Close itself does not need a separate guard for that.

Close does not stop a running Engine — an Engine with Run in flight must be stopped via Handle.Stop first (Invariant 7: that is the graceful-drain path). Calling Close while Run is running closes the database out from under the running pipelines, which is a caller error, not something Close detects or prevents. Likewise, calling Close concurrently with the very first Run or Import call on this Engine (i.e. while ensureRuntime is still opening the database for the first time) blocks until that open attempt finishes (Close and ensureRuntime share the same mutex) and then releases whatever it produced — but starting that race at all is caller-orchestrated concurrency this method does not otherwise guard against.

ctx is accepted for interface symmetry with Stop and for future resources that may need a bounded release; the current database.DB.Close call is not itself context-aware.

func (*Engine) Import

func (e *Engine) Import(ctx context.Context, cfg PipelineConfig) error

Import creates or updates a pipeline from cfg, delegating to the same provisioning path the HTTP/gRPC API and CLI use (provisioning.Service.Import) — no divergent logic. cfg is tagged as programmatically provisioned internally, so it is not removed by config-file reconciliation on a future Run (see pkg/provisioning/import.go's doc on Import vs. the config-file path).

Import enriches cfg with the same defaults the file-based provisioning path applies (provisioning/config.Enrich — DLQ defaults, connector/processor Settings, ID namespacing) and validates the result (provisioning/config.Validate) before handing it to the provisioning service, exactly mirroring provisioning.Service's own file-based provisionPipeline. This matters for a hand-built PipelineConfig literal specifically: provisioning.Service.Import assumes an already-enriched config (e.g. it dereferences cfg.DLQ.WindowSize/ WindowNackThreshold directly, which are nil on a zero-value DLQ) — skipping Enrich here would crash on the exact construction pattern Import exists for. A malformed cfg (after enrichment) is rejected as a coded error from Validate, not a later, harder-to-diagnose provisioning failure.

Import may be called before Run — a pure-embed, Import-driven caller that never needs a Handle need not call Run at all. It is (with Run) one of the two calls that can trigger ensureRuntime and lazily open Options.DB, if neither has run yet on this Engine — see Engine's "Lifecycle contract" doc.

func (*Engine) ImportPipeline

func (e *Engine) ImportPipeline(ctx context.Context, b *PipelineBuilder) error

ImportPipeline builds b and imports the result in one call — the common case for a host that constructs its pipeline with NewPipeline instead of parsing YAML. It is exactly:

cfg, err := b.Build()
if err != nil {
	return err
}
return e.Import(ctx, cfg)

b.Build()'s error (a coded *conduiterr.ConduitError — missing id/plugin/ type, a duplicate connector/processor id, a nil nested builder passed to one of PipelineBuilder's With... methods, ...) is returned as-is, before Import is ever called; Import's own enrichment/validation still runs against the built PipelineConfig exactly as it would for any other caller of Import. This is the one call an embedder needs for the common "define a pipeline in code, run it" path — see the package Example.

func (*Engine) Run

func (e *Engine) Run(ctx context.Context) (*Handle, error)

Run starts the engine: on first call, it lazily opens Options.DB and constructs every service (ensureRuntime — a no-op if Import already did this), then initializes all services and, if Options.API.Enabled, begins serving the HTTP/gRPC API. It returns as soon as either the engine becomes ready to accept work (a non-nil *Handle) or startup definitively fails (a non-nil error) — it never blocks indefinitely.

ctx bounds the database dial if this is the first Run or Import call on this Engine (see ensureRuntime), and then governs the engine's entire run: cancelling it starts the same graceful drain Handle.Stop triggers (Invariant 7) — Run reuses pkg/conduit's Runtime.Run and its existing ctx-cancellation-driven tomb drain unchanged, the same mechanism `conduit run`'s SIGTERM handling already drives from an OS signal via Entrypoint.CancelOnInterrupt. Run does not store ctx beyond deriving a cancelable child from it; a caller-supplied deadline propagates normally.

Run must be called at most once per Engine. A second call returns a coded error (conduiterr.CodeInvalidArgument) rather than racing a second tomb against the first — this is deliberately not a new conduiterr code: a double Run is caller misuse that has never been observed in practice, not a distinct, documented failure mode that needs its own machine-actionable identity.

A caller-configured Options.PipelinesDir that fails to provision (a missing directory, malformed pipeline YAML, ...) is a startup failure Run surfaces as a returned error — not merely a log line, unlike `conduit run`'s own default behavior for the identical failure (see pkgconduit.Runtime.ProvisioningErr's doc, which this reads: that field is purely additive and `conduit run` never consults it). B1 DX hardening: the CLI's swallow-unless-exit-on-degraded default, reused unchanged on the embed path, violated New's "every failure returned as an error" promise. Run detects this right after Ready fires (the same point ProvisionService.Init has already run), gracefully stops what it just started (Invariant 7 — the identical drain Handle.Stop would trigger), and returns the provisioning error instead of a Handle. This never fires when Options.PipelinesDir is empty: file provisioning is Disabled entirely in that case (see New's doc), not attempted against a possibly-nonexistent path.

func (*Engine) StartPipeline

func (e *Engine) StartPipeline(ctx context.Context, pipelineID string) error

StartPipeline starts a previously imported/provisioned pipeline by ID, delegating to the orchestrator unchanged. Calling it before Run or Import has ever run on this Engine still lazily opens Options.DB (ensureRuntime) so this returns the orchestrator's own not-found error rather than a nil-pointer panic — though no pipeline could exist yet on such an Engine.

func (*Engine) StopPipeline

func (e *Engine) StopPipeline(ctx context.Context, pipelineID string, force bool) error

StopPipeline stops a running pipeline by ID, delegating to the orchestrator unchanged. force, when true, skips the graceful drain for that single pipeline (mirrors the orchestrator's own Stop semantics) — an embedder that wants Invariant-7 graceful shutdown for everything should prefer Handle.Stop (which always drains) over StopPipeline(force=true) for individual pipelines. See StartPipeline's doc on why this also routes through ensureRuntime.

type Handle

type Handle struct {
	// contains filtered or unexported fields
}

Handle is returned by Engine.Run once the engine has successfully started. Use Stop to drain and shut it down. A Handle is only ever constructed inside Engine.Run — there is no supported way to obtain one otherwise.

func (*Handle) Stop

func (h *Handle) Stop(ctx context.Context) error

Stop drains in-flight records and checkpoints, then shuts the engine down.

Invariant 7: Stop triggers the identical ctx-cancellation-driven drain Runtime.Run's registerCleanup dispatches to (registerCleanupV1/V2, unchanged by this package) — the same path `conduit run`'s SIGTERM handling already exercises. Stop does not reimplement or duplicate that logic; it only supplies a different trigger (an explicit call instead of a signal).

Stop is idempotent: concurrent or repeated calls return the same result, with no panic and no double-close. Calling Stop on a nil *Handle is a caller bug (see Handle's doc); Stop reports it as a coded error (conduiterr.CodeInvalidArgument — again deliberately not a new code, see Run's doc) rather than panicking.

If ctx is done before the drain completes, Stop returns a coded timeout error without blocking forever; the underlying drain continues in the background (Runtime.Run's own registerCleanup timeout, exitTimeout, still applies) — a subsequent Stop call returns the same timeout result rather than re-waiting, per the idempotency guarantee above.

type Options

type Options struct {
	// Logger receives every log line Conduit produces, with level and
	// structured fields preserved as slog attributes. nil defaults to
	// slog.Default() (an explicit, documented choice, not an error). Two
	// concurrent Engines constructed with distinct Loggers never cross-talk:
	// Conduit never writes to os.Stdout/os.Stderr or a process-global logger
	// on this path. The resolved logger also backs the leak-check finalizer
	// (see Engine's "Lifecycle contract" doc) if Close is never called.
	Logger *slog.Logger

	// MetricsRegisterer receives Conduit's Prometheus metric families. nil
	// means "do not expose metrics" — metrics are silently disabled, not an
	// error. It is never prometheus.DefaultRegisterer implicitly: Conduit's
	// metrics never reach the process-global default registry through New.
	//
	// Known limitation: see this package's doc.go "metrics cross-talk"
	// section — two Engines each get an isolated Registerer/Registry
	// *object*, but pkg/foundation/metrics' process-global metric
	// definitions still fan values out across every registry in the process.
	MetricsRegisterer promclient.Registerer

	// DB configures the embedded storage backend. The zero value (Type =="")
	// defaults to an in-memory store — convenient for examples and tests, but
	// every pipeline configuration is lost once the Engine stops. A
	// production embedder should set Type explicitly.
	DB DBOptions

	// PipelinesDir optionally points at a directory (or a single file) of
	// pipeline YAML configs Conduit provisions on Run. Leave empty to skip
	// file-based provisioning entirely and manage pipelines exclusively
	// through Engine.Import — New never falls back to a CLI-style
	// cwd/pipelines default in that case (B1 DX hardening fix: an earlier
	// version of this package fell back to scanning the working directory
	// anyway, and — because that directory usually doesn't exist for a
	// pure-embed caller — provisioning then failed silently, as a swallowed
	// ERROR log, while Run still returned a healthy Handle). When
	// PipelinesDir is set and provisioning genuinely fails (a missing
	// directory, malformed YAML, ...), Run returns that failure as an error
	// instead of only logging it — see Run's doc.
	PipelinesDir string

	// API optionally exposes Conduit's HTTP/gRPC API. Disabled by default: an
	// embedder that only needs in-process orchestration is not forced to bind
	// a socket.
	API APIOptions
}

Options configures a New Engine. Logger and MetricsRegisterer have deliberately different zero-value semantics — read both before assuming symmetry: a nil Logger is a safe, explicit default; a nil MetricsRegisterer disables metrics.

type PipelineBuilder

type PipelineBuilder struct {
	// contains filtered or unexported fields
}

Passing a nil *ConnectorBuilder/*ProcessorBuilder/*DLQBuilder to a With... method never panics: it records a conduiterr.CodeInvalidArgument error (the same code Run/Stop misuse already uses, see conduit.go — no new code minted) that surfaces from the next Build call, alongside any other accumulated or validation errors. The pipeline being built is otherwise unaffected by the nil argument (no zero-value entry is appended).

func NewPipeline

func NewPipeline(id string) *PipelineBuilder

NewPipeline starts a PipelineBuilder for a pipeline with the given id. id is required; Build returns a coded error if it's ever empty when called.

func (*PipelineBuilder) Build

func (b *PipelineBuilder) Build() (PipelineConfig, error)

Build validates the pipeline and returns the resulting PipelineConfig.

Validation runs Validate against an enriched copy of the pipeline — the same config.Enrich-then-config.Validate sequence `conduit pipelines validate` runs on a parsed YAML document — so a Build-time error catches the same class of mistake the CLI would catch (missing id/plugin/type, an out-of-range name/description, a negative worker count, a duplicate connector/processor id, ...). Any nil-nested-builder misuse recorded by a With... method (see PipelineBuilder's doc) is joined into the same returned error, so a single Build call surfaces everything wrong at once. The returned PipelineConfig, however, is the raw, unenriched value this builder constructed: final namespaced IDs and injected DLQ defaults are Engine.Import's job at import time, not Build's — which is what keeps a hand-built PipelineConfig indistinguishable from a parsed one, the property this package's round-trip tests assert.

A validation failure is returned as the same coded, config-path-scoped *conduiterr.ConduitError provisioning/config.Validate already produces for the CLI and API; a nil-nested-builder misuse is a conduiterr.CodeInvalidArgument error (see PipelineBuilder's doc) — Build never panics and never returns an uncoded error. Calling Build on a nil *PipelineBuilder itself is likewise reported as a coded error rather than a nil-pointer panic.

func (*PipelineBuilder) WithConnector

func (b *PipelineBuilder) WithConnector(c *ConnectorBuilder) *PipelineBuilder

WithConnector appends a connector built by NewSourceConnector or NewDestinationConnector. Call order is preserved. A nil c is recorded as a Build-time error rather than panicking — see PipelineBuilder's doc.

func (*PipelineBuilder) WithDLQ

func (b *PipelineBuilder) WithDLQ(d *DLQBuilder) *PipelineBuilder

WithDLQ sets the pipeline's dead-letter-queue configuration. Omitting this call leaves DLQ at its zero value — matching a YAML pipeline with no dead-letter-queue block — and Engine.Import's enrichment fills in Conduit's default DLQ plugin/settings/window at import time, exactly as it does for a parsed YAML pipeline that also omits the block. Calling WithDLQ a second time replaces the previous value rather than merging. A nil d is recorded as a Build-time error rather than panicking — see PipelineBuilder's doc.

func (*PipelineBuilder) WithDescription

func (b *PipelineBuilder) WithDescription(description string) *PipelineBuilder

WithDescription sets the pipeline's human-readable description.

func (*PipelineBuilder) WithName

func (b *PipelineBuilder) WithName(name string) *PipelineBuilder

WithName sets the pipeline's display name. Leave unset to have Engine.Import's enrichment default it to the pipeline's id.

func (*PipelineBuilder) WithProcessor

func (b *PipelineBuilder) WithProcessor(p *ProcessorBuilder) *PipelineBuilder

WithProcessor appends a pipeline-scoped processor — one that runs on every connector's records, as opposed to a processor scoped to a single connector via ConnectorBuilder.WithProcessor. Call order is preserved. A nil p is recorded as a Build-time error rather than panicking — see PipelineBuilder's doc.

func (*PipelineBuilder) WithStatus

func (b *PipelineBuilder) WithStatus(status string) *PipelineBuilder

WithStatus sets the pipeline's desired status: StatusRunning or StatusStopped. Leave unset to have Engine.Import's enrichment default it to StatusRunning.

type PipelineConfig

type PipelineConfig = provisioningconfig.Pipeline

PipelineConfig is Conduit's pipeline configuration shape: a type alias (not a copy) of provisioning/config.Pipeline, the exact struct the YAML provisioner both produces and consumes. Aliasing guarantees Engine.Import accepts precisely what a hand-built PipelineConfig value produces, with no conversion step to drift out of sync — see provisioning/config/parser.go's matching doc comment for the constraint this places on that package.

type ProcessorBuilder

type ProcessorBuilder struct {
	// contains filtered or unexported fields
}

ProcessorBuilder builds one processor. The same ProcessorBuilder type attaches at pipeline scope (PipelineBuilder.WithProcessor) or connector scope (ConnectorBuilder.WithProcessor) — config.Processor carries no scope field of its own; where a built processor is attached is what determines its scope.

func NewProcessor

func NewProcessor(id, plugin string) *ProcessorBuilder

NewProcessor starts a ProcessorBuilder with the given id and plugin (e.g. "js", "builtin:field.rename"). There is deliberately no WithType: config.Processor has no Type field — "type" is a deprecated YAML-only wire name for what this package (and config.Processor) always calls Plugin, mechanically renamed by the parser's own linter (provisioning/config/ yaml/v2's Changelog) — the builder never exposes the deprecated spelling.

func (*ProcessorBuilder) WithCondition

func (p *ProcessorBuilder) WithCondition(condition string) *ProcessorBuilder

WithCondition sets a CEL expression gating whether this processor runs on a given record. Leave empty (the default) to run unconditionally.

func (*ProcessorBuilder) WithSetting

func (p *ProcessorBuilder) WithSetting(key, value string) *ProcessorBuilder

WithSetting sets a single processor configuration key. A later call with the same key overwrites the earlier value.

func (*ProcessorBuilder) WithSettings

func (p *ProcessorBuilder) WithSettings(settings map[string]string) *ProcessorBuilder

WithSettings merges settings into the processor's configuration — the same as calling WithSetting for every entry. Keys already set and not present in settings are left untouched.

func (*ProcessorBuilder) WithWorkers

func (p *ProcessorBuilder) WithWorkers(workers int) *ProcessorBuilder

WithWorkers sets the processor's worker count. Leave unset (0) to accept Engine.Import's enrichment default of 1. Invariant 4 (per-partition ordering): more than one worker can reorder records within a key, so raise this only deliberately.

Directories

Path Synopsis
cmd
conduit command
conduit/api/mock
Package mock is a generated GoMock package.
Package mock is a generated GoMock package.
conduit/internal/deploy
Package deploy is the engine behind `conduit pipelines deploy|apply` (see docs/design-documents/20260708-cli-pipeline-deploy-apply.md): parsing a single pipeline config file, wiring a pkg/provisioning.Service to plan/ apply against, and rendering the resulting Diff for both --json and human output.
Package deploy is the engine behind `conduit pipelines deploy|apply` (see docs/design-documents/20260708-cli-pipeline-deploy-apply.md): parsing a single pipeline config file, wiring a pkg/provisioning.Service to plan/ apply against, and rendering the resulting Diff for both --json and human output.
conduit/internal/generate
Package generate is the eval-harness half of `conduit generate` (docs/design-documents/20260722-conduit-generate.md).
Package generate is the eval-harness half of `conduit generate` (docs/design-documents/20260722-conduit-generate.md).
conduit/internal/llmsgen command
Package main implements the llms.txt/llms-full.txt generator, per docs/design-documents/20260712-llms-txt-generation.md.
Package main implements the llms.txt/llms-full.txt generator, per docs/design-documents/20260712-llms-txt-generation.md.
conduit/internal/llmsgen/allcodes
Package allcodes is a blank-import barrel: importing it, and only it, guarantees every conduiterr.Code in the engine has been registered (conduiterr.Register runs at package-level var-init time — see conduiterr.Register's doc — so a package's codes only exist in the registry once something imports it).
Package allcodes is a blank-import barrel: importing it, and only it, guarantees every conduiterr.Code in the engine has been registered (conduiterr.Register runs at package-level var-init time — see conduiterr.Register's doc — so a package's codes only exist in the registry once something imports it).
conduit/internal/mcp
Package mcp is the adapter layer behind `conduit mcp` (see docs/design-documents/20260708-mcp-server.md): it registers Conduit's operations as MCP tools that are 1:1 with the CLI verbs and call the exact same cobra-free engines — the "three faces" rule (human CLI, agent MCP, embedder library are renderings of one operation set).
Package mcp is the adapter layer behind `conduit mcp` (see docs/design-documents/20260708-mcp-server.md): it registers Conduit's operations as MCP tools that are 1:1 with the CLI verbs and call the exact same cobra-free engines — the "three faces" rule (human CLI, agent MCP, embedder library are renderings of one operation set).
conduit/internal/repair
Package repair is the engine behind `conduit pipelines repair` and the MCP `repair`/`repair_apply` tools (see docs/design-documents/20260712-repair-command.md).
Package repair is the engine behind `conduit pipelines repair` and the MCP `repair`/`repair_apply` tools (see docs/design-documents/20260712-repair-command.md).
conduit/internal/scaffoldcmd
Package scaffoldcmd implements the shared command body behind `conduit connector new` and `conduit processor new` (cmd/conduit/root/connectors and cmd/conduit/root/processors each register a two-line wrapper that sets Kind and the command-specific Usage/Docs/Example — see NewCommand).
Package scaffoldcmd implements the shared command body behind `conduit connector new` and `conduit processor new` (cmd/conduit/root/connectors and cmd/conduit/root/processors each register a two-line wrapper that sets Kind and the command-specific Usage/Docs/Example — see NewCommand).
conduit/internal/ui
Package ui is the single place that owns human-output presentation rules for the v0.17 "CLI as product" commands (`pipelines validate|lint|dry-run`, `doctor`, `connector|processor new`), per docs/design-documents/20260707-cli-output-conventions.md §2.
Package ui is the single place that owns human-output presentation rules for the v0.17 "CLI as product" commands (`pipelines validate|lint|dry-run`, `doctor`, `connector|processor new`), per docs/design-documents/20260707-cli-output-conventions.md §2.
conduit/internal/validate
Package validate is the shared, offline engine behind `conduit pipelines validate` (and, in a future PR, `lint` and `dry-run`), per docs/design-documents/20260707-cli-pipeline-validate.md.
Package validate is the shared, offline engine behind `conduit pipelines validate` (and, in a future PR, `lint` and `dry-run`), per docs/design-documents/20260707-cli-pipeline-validate.md.
conduit/root/doctor
Package doctor is the cobra-facing wrapper around doctorcheck's pure check set: `conduit doctor`, per docs/design-documents/20260707-cli-doctor.md.
Package doctor is the cobra-facing wrapper around doctorcheck's pure check set: `conduit doctor`, per docs/design-documents/20260707-cli-doctor.md.
conduit/root/doctor/doctorcheck
Package doctorcheck is `conduit doctor`'s check SET: the concrete pkg/conduit/check.Check implementations that answer "would `conduit run` succeed here?" (config resolution/validation, database reachability, API address availability, plugin directories, the built-in plugin registry, and whether a running engine is reachable), per docs/design-documents/20260707-cli-doctor.md and the ADR docs/architecture-decision-records/20260707-check-engine-shared-by-doctor-and-scaffold.md.
Package doctorcheck is `conduit doctor`'s check SET: the concrete pkg/conduit/check.Check implementations that answer "would `conduit run` succeed here?" (config resolution/validation, database reachability, API address availability, plugin directories, the built-in plugin registry, and whether a running engine is reachable), per docs/design-documents/20260707-cli-doctor.md and the ADR docs/architecture-decision-records/20260707-check-engine-shared-by-doctor-and-scaffold.md.
conduit/root/mcp
Package mcp implements `conduit mcp`: the ecdysis command that runs the agent-native MCP server (see docs/design-documents/20260708-mcp-server.md).
Package mcp implements `conduit mcp`: the ecdysis command that runs the agent-native MCP server (see docs/design-documents/20260708-mcp-server.md).
conduit/root/pipelines
This file backs `conduit pipelines init --template <name>` (v0.19 Workstream 3, docs/design-documents/20260723-templates-gallery.md) — a small, permanently-maintained, embedded (go:embed) gallery of named, runnable pipeline templates.
This file backs `conduit pipelines init --template <name>` (v0.19 Workstream 3, docs/design-documents/20260723-templates-gallery.md) — a small, permanently-maintained, embedded (go:embed) gallery of named, runnable pipeline templates.
conduit/root/quickstart
Package quickstart implements `conduit quickstart`, the zero-config 5-minute-wow command: it scaffolds an ephemeral demo pipeline (built-in generator -> built-in log) and runs it in-process so a new user sees records flowing within seconds, with nothing written to their working directory.
Package quickstart implements `conduit quickstart`, the zero-config 5-minute-wow command: it scaffolds an ephemeral demo pipeline (built-in generator -> built-in log) and runs it in-process so a new user sees records flowing within seconds, with nothing written to their working directory.
pkg
conduit
Package conduit wires up everything under the hood of a Conduit instance including metrics, telemetry, logging, and server construction.
Package conduit wires up everything under the hood of a Conduit instance including metrics, telemetry, logging, and server construction.
conduit/check
Package check is the neutral diagnostic engine shared by `conduit doctor` and the `connector|processor new` scaffold preflight (see docs/architecture-decision-records/20260707-check-engine-shared-by-doctor-and-scaffold.md).
Package check is the neutral diagnostic engine shared by `conduit doctor` and the `connector|processor new` scaffold preflight (see docs/architecture-decision-records/20260707-check-engine-shared-by-doctor-and-scaffold.md).
conduit/dev
Package dev implements the file watcher behind `conduit run --dev` (see docs/design-documents/20260712-pipeline-dev-hot-reload.md, §4 "Surface").
Package dev implements the file watcher behind `conduit run --dev` (see docs/design-documents/20260712-pipeline-dev-hot-reload.md, §4 "Surface").
conduit/exitcode
Package exitcode computes Conduit's deterministic CLI process exit code from an error.
Package exitcode computes Conduit's deterministic CLI process exit code from an error.
foundation/atomicfile
Package atomicfile writes a whole file's contents durably and atomically: a crash at any point leaves either the previous complete content or the new complete content, never a torn write (Invariant 5 — "state and checkpoint writes are atomic").
Package atomicfile writes a whole file's contents durably and atomically: a crash at any point leaves either the previous complete content or the new complete content, never a torn write (Invariant 5 — "state and checkpoint writes are atomic").
foundation/cerrors
Package cerrors contains functions related to error handling.
Package cerrors contains functions related to error handling.
foundation/cerrors/conduiterr
Package conduiterr defines ConduitError, Conduit's uniform, machine-actionable error type.
Package conduiterr defines ConduitError, Conduit's uniform, machine-actionable error type.
foundation/metrics/noop
Package noop exposes implementations of metrics which do not do anything.
Package noop exposes implementations of metrics which do not do anything.
http/api/mock
Package mock is a generated GoMock package.
Package mock is a generated GoMock package.
lifecycle
Package lifecycle contains the logic to manage the lifecycle of pipelines.
Package lifecycle contains the logic to manage the lifecycle of pipelines.
lifecycle-poc
Package lifecycle contains the logic to manage the lifecycle of pipelines.
Package lifecycle contains the logic to manage the lifecycle of pipelines.
lifecycle/stream
Package stream defines a message and nodes that can be composed into a data pipeline.
Package stream defines a message and nodes that can be composed into a data pipeline.
lifecycle/stream/mock
Package mock is a generated GoMock package.
Package mock is a generated GoMock package.
orchestrator/mock
Package mock is a generated GoMock package.
Package mock is a generated GoMock package.
plugin/connector/mock
Package mock is a generated GoMock package.
Package mock is a generated GoMock package.
plugin/connector/standalone/test/testplugin command
Package main contains a plugin used for testing purposes.
Package main contains a plugin used for testing purposes.
plugin/processor/builtin/impl/avro
Package avro is a generated GoMock package.
Package avro is a generated GoMock package.
plugin/processor/builtin/impl/ollama/mock
Package mock is a generated GoMock package.
Package mock is a generated GoMock package.
plugin/processor/builtin/internal/diff
Package diff computes differences between text files or strings.
Package diff computes differences between text files or strings.
plugin/processor/builtin/internal/diff/difftest
Package difftest supplies a set of tests that will operate on any implementation of a diff algorithm as exposed by "github.com/conduitio/conduit/pkg/plugin/processor/builtin/internal/diff"
Package difftest supplies a set of tests that will operate on any implementation of a diff algorithm as exposed by "github.com/conduitio/conduit/pkg/plugin/processor/builtin/internal/diff"
plugin/processor/builtin/internal/diff/lcs
package lcs contains code to find longest-common-subsequences (and diffs)
package lcs contains code to find longest-common-subsequences (and diffs)
plugin/processor/mock
Package mock is a generated GoMock package.
Package mock is a generated GoMock package.
processor/mock
Package mock is a generated GoMock package.
Package mock is a generated GoMock package.
provisioning
Package provisioning's plan.go implements the preview/diff engine behind `conduit pipelines deploy|apply` (see docs/design-documents/20260708-cli-pipeline-deploy-apply.md).
Package provisioning's plan.go implements the preview/diff engine behind `conduit pipelines deploy|apply` (see docs/design-documents/20260708-cli-pipeline-deploy-apply.md).
provisioning/mock
Package mock is a generated GoMock package.
Package mock is a generated GoMock package.
registry
This file implements the offline/air-gapped install path (plan-v2/step5 §7): Bundle prepares a self-contained tarball on a networked machine, running the full real install-equivalent verification before ever writing the tarball; InstallFromBundle installs from that tarball on a machine with NO network access at all, re-verifying everything against the offline binary's own compiled-in trust anchors — it NEVER skips verification just because the network is down.
This file implements the offline/air-gapped install path (plan-v2/step5 §7): Bundle prepares a self-contained tarball on a networked machine, running the full real install-equivalent verification before ever writing the tarball; InstallFromBundle installs from that tarball on a machine with NO network access at all, re-verifying everything against the offline binary's own compiled-in trust anchors — it NEVER skips verification just because the network is down.
registry/boundedfetch
Package boundedfetch is the shared size-capped fetch primitive behind P0-2 (plan-v2 §2.4 item 1): attacker-controlled bytes (an index, a connector artifact, a signature bundle, a provenance bundle) are always read through an io.LimitReader capped one byte past the caller's limit, so a response that would exceed the cap is distinguished (ErrTooLarge) from one that happens to land exactly on it, and from any other fetch failure.
Package boundedfetch is the shared size-capped fetch primitive behind P0-2 (plan-v2 §2.4 item 1): attacker-controlled bytes (an index, a connector artifact, a signature bundle, a provenance bundle) are always read through an io.LimitReader capped one byte past the caller's limit, so a response that would exceed the cap is distinguished (ErrTooLarge) from one that happens to land exactly on it, and from any other fetch failure.
registry/index
Package index parses, freezes, and (eventually) verifies the signed connector registry index — the security-critical foundation `install`, `audit`, and the publish Action all trust.
Package index parses, freezes, and (eventually) verifies the signed connector registry index — the security-critical foundation `install`, `audit`, and the publish Action all trust.
registry/policy
Package policy is the --allow-unsigned gate (plan-v2 §6, P1-3).
Package policy is the --allow-unsigned gate (plan-v2 §6, P1-3).
registry/trust/adversarial
Package adversarial is the shared fixture corpus (plan-v2 §11, P1-4): a small set of hand-built, LOCALLY-generated (never touching production Sigstore or Conduit registry infrastructure) signed/attested test entities, each paired with the pkg/registry/trust error this build's verification pipeline must produce for it.
Package adversarial is the shared fixture corpus (plan-v2 §11, P1-4): a small set of hand-built, LOCALLY-generated (never touching production Sigstore or Conduit registry infrastructure) signed/attested test entities, each paired with the pkg/registry/trust error this build's verification pipeline must produce for it.
registry/trust/trustroot
Package trustroot embeds a pinned, offline snapshot of Sigstore's own public-good trust root (layer 1: Fulcio CA, Rekor transparency log, CTFE public keys — see pkg/registry/trust's package doc for the two-layer distinction).
Package trustroot embeds a pinned, offline snapshot of Sigstore's own public-good trust root (layer 1: Fulcio CA, Rekor transparency log, CTFE public keys — see pkg/registry/trust's package doc for the two-layer distinction).
scaffold
Package scaffold implements the engine behind `conduit connector new` and `conduit processor new`, per docs/design-documents/20260707-connector-processor-scaffolding.md.
Package scaffold implements the engine behind `conduit connector new` and `conduit processor new`, per docs/design-documents/20260707-connector-processor-scaffolding.md.
scaffold/steps
Package steps implements the post-rewrite pipeline scaffold.Generate runs against a staged, renamed template tree: install the code-gen tool, generate, verify the result builds, and initialize git.
Package steps implements the post-rewrite pipeline scaffold.Generate runs against a staged, renamed template tree: install the code-gen tool, generate, verify the result builds, and initialize git.
scaffold/template
Package template owns the vendored, go:embed'd snapshots of ConduitIO/conduit-connector-template and ConduitIO/conduit-processor-template (Extract), and the Go port of each template's setup.sh rename step (Rewrite).
Package template owns the vendored, go:embed'd snapshots of ConduitIO/conduit-connector-template and ConduitIO/conduit-processor-template (Extract), and the Go port of each template's setup.sh rename step (Rewrite).
web/ui
Package ui embeds and serves conduit-ui, Conduit's built-in web UI (observe + operate).
Package ui embeds and serves conduit-ui, Conduit's built-in web UI (observe + operate).
proto
api/v1
Package apiv1 is a reverse proxy.
Package apiv1 is a reverse proxy.
tests
chaos
Package chaos is the repo's first chaos-testing harness (v0.19 Workstream 7, DBZ-1).
Package chaos is the repo's first chaos-testing harness (v0.19 Workstream 7, DBZ-1).

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL