diff --git a/cmd/conduit/root/pipelines/apply.go b/cmd/conduit/root/pipelines/apply.go index 4a4a592f2..ee32330fb 100644 --- a/cmd/conduit/root/pipelines/apply.go +++ b/cmd/conduit/root/pipelines/apply.go @@ -111,7 +111,7 @@ func (c *ApplyCommand) Args(args []string) error { func (c *ApplyCommand) Config() ecdysis.Config { path := filepath.Dir(c.flags.ConduitCfg.Path) return ecdysis.Config{ - EnvPrefix: "CONDUIT", + EnvPrefix: envPrefix, Parsed: &c.flags.Config, Path: c.flags.ConduitCfg.Path, DefaultValues: conduit.DefaultConfigWithBasePath(path), diff --git a/cmd/conduit/root/pipelines/deploy.go b/cmd/conduit/root/pipelines/deploy.go index cbb09ead5..3eeb860c2 100644 --- a/cmd/conduit/root/pipelines/deploy.go +++ b/cmd/conduit/root/pipelines/deploy.go @@ -124,7 +124,7 @@ func (c *DeployCommand) Args(args []string) error { func (c *DeployCommand) Config() ecdysis.Config { path := filepath.Dir(c.flags.ConduitCfg.Path) return ecdysis.Config{ - EnvPrefix: "CONDUIT", + EnvPrefix: envPrefix, Parsed: &c.flags.Config, Path: c.flags.ConduitCfg.Path, DefaultValues: conduit.DefaultConfigWithBasePath(path), diff --git a/cmd/conduit/root/pipelines/dev.go b/cmd/conduit/root/pipelines/dev.go new file mode 100644 index 000000000..191f58417 --- /dev/null +++ b/cmd/conduit/root/pipelines/dev.go @@ -0,0 +1,168 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package pipelines + +import ( + "context" + "fmt" + "os" + "path/filepath" + + "github.com/conduitio/conduit/pkg/conduit" + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/ecdysis" +) + +var ( + _ ecdysis.CommandWithFlags = (*DevCommand)(nil) + _ ecdysis.CommandWithExecute = (*DevCommand)(nil) + _ ecdysis.CommandWithDocs = (*DevCommand)(nil) + _ ecdysis.CommandWithConfig = (*DevCommand)(nil) + _ ecdysis.CommandWithArgs = (*DevCommand)(nil) +) + +// DevArgs holds DevCommand's optional positional argument. +type DevArgs struct { + // Dir is a GitOps-friendly alias for --pipelines.path, matching `run`'s + // --pipelines flag but as a positional argument (`conduit pipelines dev + // [dir]` reads more naturally than `conduit pipelines dev + // --pipelines.path dir`). + Dir string +} + +// DevFlags holds DevCommand's flags. conduit.Config is embedded for the same +// reason RunFlags embeds it: `pipelines dev` runs a full `conduit run --dev` +// server, so it accepts the same db.*/api.*/pipelines.*/... configuration. +type DevFlags struct { + conduit.Config +} + +// DevCommand implements `conduit pipelines dev [dir]`: the design doc's §4 +// alias, "thin sugar over run --dev ... carrying no logic of its own". It +// does not reimplement any watch/apply behavior (that all lives in +// pkg/conduit/dev, driven from pkg/conduit.Runtime) — it only sets +// Cfg.Dev.Enabled, points Cfg.Pipelines.Path at the optional dir argument, +// applies the "dev defaults" (exit-on-error off, so one bad pipeline while +// iterating doesn't take the whole dev server down), and calls the exact +// same conduit.Entrypoint.Serve RunCommand does. +type DevCommand struct { + args DevArgs + flags DevFlags + Cfg conduit.Config +} + +func (c *DevCommand) Usage() string { return "dev [dir]" } + +func (c *DevCommand) Docs() ecdysis.Docs { + return ecdysis.Docs{ + Short: "Run Conduit with the hot-reload dev watcher (alias for 'conduit run --dev')", + Long: `Starts the Conduit server exactly like 'conduit run --dev', watching pipelines.path (or +[dir], if given) for changes and applying them into the running engine as you save — a processor +edit applies in place (no restart), a source/destination edit applies via a labeled graceful +restart, and a bad edit is reported without ever touching the running pipeline. See +docs/design-documents/20260712-pipeline-dev-hot-reload.md §4 and docs/operations/dev-hot-reload.md. + +This is sugar over 'conduit run --dev [dir]', with exit-on-degraded forced off (dev iterates on +pipelines that may be intentionally broken mid-edit; one degraded pipeline should not take the +whole dev server down) — it carries no watcher logic of its own.`, + Example: "conduit pipelines dev\n" + + "conduit pipelines dev ./pipelines\n" + + "conduit pipelines dev --dev.json", + } +} + +func (c *DevCommand) Args(args []string) error { + if len(args) > 1 { + return cerrors.Errorf("too many arguments") + } + if len(args) == 1 { + c.args.Dir = args[0] + } + return nil +} + +func (c *DevCommand) Execute(ctx context.Context) error { + if c.args.Dir != "" { + c.Cfg.Pipelines.Path = c.args.Dir + } + // `pipelines dev` always runs in dev mode — that is the entire point of + // the alias — and defaults exit-on-degraded off (see the type doc). + c.Cfg.Dev.Enabled = true + c.Cfg.Pipelines.ExitOnDegraded = false + + if !c.Cfg.API.Enabled { + fmt.Print("Warning: API is currently disabled. Most Conduit CLI commands won't work without the API enabled." + + "To enable it, run conduit with `--api.enabled=true` or set `CONDUIT_API_ENABLED=true` in your environment.") + } + + (&conduit.Entrypoint{}).Serve(c.Cfg) + return nil +} + +func (c *DevCommand) Config() ecdysis.Config { + path := filepath.Dir(c.flags.ConduitCfg.Path) + return ecdysis.Config{ + EnvPrefix: envPrefix, + Parsed: &c.Cfg, + Path: c.flags.ConduitCfg.Path, + DefaultValues: conduit.DefaultConfigWithBasePath(path), + } +} + +// Flags mirrors run.RunCommand.Flags's defaulting (storeConfigDefaults +// covers only the offline deploy/apply subset; `dev` runs a full server, so +// it needs the same defaults `run` sets, including the dev.* group so +// --dev.json works without also passing --dev). +func (c *DevCommand) Flags() []ecdysis.Flag { + flags := ecdysis.BuildFlags(&c.flags) + + currentPath, err := os.Getwd() + if err != nil { + panic(cerrors.Errorf("failed to get current working directory: %w", err)) + } + + c.Cfg = conduit.DefaultConfigWithBasePath(currentPath) + flags.SetDefault("config.path", c.Cfg.ConduitCfg.Path) + flags.SetDefault("db.type", c.Cfg.DB.Type) + flags.SetDefault("db.badger.path", c.Cfg.DB.Badger.Path) + flags.SetDefault("db.postgres.connection-string", c.Cfg.DB.Postgres.ConnectionString) + flags.SetDefault("db.postgres.table", c.Cfg.DB.Postgres.Table) + flags.SetDefault("db.sqlite.path", c.Cfg.DB.SQLite.Path) + flags.SetDefault("db.sqlite.table", c.Cfg.DB.SQLite.Table) + flags.SetDefault("api.enabled", c.Cfg.API.Enabled) + flags.SetDefault("api.http.address", c.Cfg.API.HTTP.Address) + flags.SetDefault("api.grpc.address", c.Cfg.API.GRPC.Address) + flags.SetDefault("api.allow-live-restart-apply", c.Cfg.API.AllowLiveRestartApply) + flags.SetDefault("log.level", c.Cfg.Log.Level) + flags.SetDefault("log.format", c.Cfg.Log.Format) + flags.SetDefault("connectors.path", c.Cfg.Connectors.Path) + flags.SetDefault("connectors.max-receive-record-size", c.Cfg.Connectors.MaxReceiveRecordSize) + flags.SetDefault("processors.path", c.Cfg.Processors.Path) + flags.SetDefault("pipelines.path", c.Cfg.Pipelines.Path) + flags.SetDefault("pipelines.error-recovery.min-delay", c.Cfg.Pipelines.ErrorRecovery.MinDelay) + flags.SetDefault("pipelines.error-recovery.max-delay", c.Cfg.Pipelines.ErrorRecovery.MaxDelay) + flags.SetDefault("pipelines.error-recovery.backoff-factor", c.Cfg.Pipelines.ErrorRecovery.BackoffFactor) + flags.SetDefault("pipelines.error-recovery.max-retries", c.Cfg.Pipelines.ErrorRecovery.MaxRetries) + flags.SetDefault("pipelines.error-recovery.max-retries-window", c.Cfg.Pipelines.ErrorRecovery.MaxRetriesWindow) + flags.SetDefault("schema-registry.type", c.Cfg.SchemaRegistry.Type) + flags.SetDefault("schema-registry.confluent.connection-string", c.Cfg.SchemaRegistry.Confluent.ConnectionString) + flags.SetDefault("preview.pipeline-arch-v2", c.Cfg.Preview.PipelineArchV2) + flags.SetDefault("preview.pipeline-arch-v2-disable-metrics", c.Cfg.Preview.PipelineArchV2DisableMetrics) + // pipelines.exit-on-degraded is deliberately NOT defaulted from + // DefaultConfigWithBasePath here: Execute always forces it false for + // `pipelines dev` regardless of flags/env/config file (see the type doc). + + return flags +} diff --git a/cmd/conduit/root/pipelines/pipelines.go b/cmd/conduit/root/pipelines/pipelines.go index 7840a68ae..2fdd10e2d 100644 --- a/cmd/conduit/root/pipelines/pipelines.go +++ b/cmd/conduit/root/pipelines/pipelines.go @@ -18,6 +18,11 @@ import ( "github.com/conduitio/ecdysis" ) +// envPrefix is the environment variable prefix every subcommand's +// ecdysis.Config uses (CONDUIT_DB_TYPE, CONDUIT_PIPELINES_PATH, ...) — a +// shared constant so deploy/apply/dev can't drift on it independently. +const envPrefix = "CONDUIT" + var ( _ ecdysis.CommandWithDocs = (*PipelinesCommand)(nil) _ ecdysis.CommandWithSubCommands = (*PipelinesCommand)(nil) @@ -42,6 +47,7 @@ func (c *PipelinesCommand) SubCommands() []ecdysis.Command { &RepairCommand{}, &StartCommand{}, &StopCommand{}, + &DevCommand{}, } } diff --git a/cmd/conduit/root/run/run.go b/cmd/conduit/root/run/run.go index f9a963e08..65cc60630 100644 --- a/cmd/conduit/root/run/run.go +++ b/cmd/conduit/root/run/run.go @@ -37,6 +37,14 @@ var ( // applied in Execute. const pipelinesFlagName = "pipelines" +// devFlagName is the short --dev alias for --dev.enabled (see +// docs/design-documents/20260712-pipeline-dev-hot-reload.md §4: "conduit run +// --dev [--pipelines.path ]"). Like --pipelines/--pipelines.path, it +// can't be bound directly onto conduit.Config (the name "dev" is already a +// struct — conduit.Config.Dev — not a bool), so it is excluded from the +// viper config binding and applied in Execute. +const devFlagName = "dev" + type RunFlags struct { conduit.Config } @@ -49,13 +57,22 @@ type RunCommand struct { // collides with the Pipelines struct key), so it's excluded from the config // binding (see Config) and applied in Execute. pipelinesAlias string + // devAlias backs the --dev flag — see devFlagName's doc. + devAlias bool } func (c *RunCommand) Execute(ctx context.Context) error { + cmd := ecdysis.CobraCmdFromContext(ctx) + // --pipelines is an alias for --pipelines.path; apply it when explicitly set. - if cmd := ecdysis.CobraCmdFromContext(ctx); cmd != nil && cmd.Flags().Changed(pipelinesFlagName) { + if cmd != nil && cmd.Flags().Changed(pipelinesFlagName) { c.Cfg.Pipelines.Path = c.pipelinesAlias } + // --dev is an alias for --dev.enabled; apply it when explicitly set (never + // clobber an explicit --dev.enabled=false with a stray --dev.enabled default). + if cmd != nil && cmd.Flags().Changed(devFlagName) { + c.Cfg.Dev.Enabled = c.devAlias + } e := &conduit.Entrypoint{} @@ -75,9 +92,10 @@ func (c *RunCommand) Config() ecdysis.Config { EnvPrefix: "CONDUIT", Parsed: &c.Cfg, Path: c.flags.ConduitCfg.Path, - // --pipelines collides with the Pipelines config struct key; exclude it from - // the viper binding and apply it in Execute (see pipelinesAlias). - ExcludedFlags: []string{pipelinesFlagName}, + // --pipelines and --dev collide with existing conduit.Config struct + // keys (Pipelines, Dev); exclude them from the viper binding and + // apply them in Execute (see pipelinesAlias/devAlias). + ExcludedFlags: []string{pipelinesFlagName, devFlagName}, DefaultValues: conduit.DefaultConfigWithBasePath(path), } } @@ -131,12 +149,34 @@ func (c *RunCommand) Flags() []ecdysis.Flag { Ptr: &c.pipelinesAlias, Default: c.Cfg.Pipelines.Path, }) + // --dev is the short alias for --dev.enabled: watch --pipelines.path and + // hot-reload changes into the running engine (see + // docs/design-documents/20260712-pipeline-dev-hot-reload.md §4). Excluded + // from the viper config binding (see Config) because its name collides + // with the Dev struct; Execute copies it into Cfg.Dev.Enabled when set. + flags = append(flags, ecdysis.Flag{ + Long: devFlagName, + Usage: "alias for --dev.enabled: watch pipelines.path and hot-reload changes into the running engine", + Ptr: &c.devAlias, + Default: c.Cfg.Dev.Enabled, + }) return flags } func (c *RunCommand) Docs() ecdysis.Docs { return ecdysis.Docs{ Short: "Run Conduit", - Long: `Starts the Conduit server and runs the configured pipelines.`, + Long: `Starts the Conduit server and runs the configured pipelines. + +With --dev, additionally watches pipelines.path and hot-reloads changes into the running engine as +you save: a processor-only edit applies in place (no restart); a source/destination/topology edit +applies via a labeled graceful restart; a bad edit (parse/validation failure) is reported and never +touches the running pipeline. Every apply --dev drives is authorized by the fact that you ran +--dev and are watching it — it does not set --api.allow-live-restart-apply, which stays +independently gated for the gRPC/HTTP/MCP surface. See +docs/design-documents/20260712-pipeline-dev-hot-reload.md and docs/operations/dev-hot-reload.md.`, + Example: "conduit run\n" + + "conduit run --dev\n" + + "conduit run --dev --pipelines.path ./pipelines --dev.json", } } diff --git a/cmd/conduit/root/run/run_test.go b/cmd/conduit/root/run/run_test.go index 7783e4af9..d8a926405 100644 --- a/cmd/conduit/root/run/run_test.go +++ b/cmd/conduit/root/run/run_test.go @@ -60,6 +60,9 @@ func TestRunCommandFlags(t *testing.T) { {longName: "dev.cpuprofile", usage: "write CPU profile to file"}, {longName: "dev.memprofile", usage: "write memory profile to file"}, {longName: "dev.blockprofile", usage: "write block profile to file"}, + {longName: "dev.enabled", usage: "watch pipelines.path and hot-reload changes into the running engine (see 'conduit run --dev')"}, + {longName: "dev.json", usage: "emit dev-watcher apply events as JSON lines instead of human-readable text"}, + {longName: "dev", usage: "alias for --dev.enabled: watch pipelines.path and hot-reload changes into the running engine"}, } e := ecdysis.New() diff --git a/docs/operations/dev-hot-reload.md b/docs/operations/dev-hot-reload.md new file mode 100644 index 000000000..9fa8c57b0 --- /dev/null +++ b/docs/operations/dev-hot-reload.md @@ -0,0 +1,67 @@ +# Dev-mode hot-reload: `conduit run --dev` + +`conduit run --dev` runs Conduit and watches the pipelines directory, applying config +changes to running pipelines on save — so authoring a pipeline is an edit-see-edit-see loop +instead of stop / edit / restart / wait. It is the developer-facing surface over the live +in-place apply engine (see +[`docs/design-documents/20260712-pipeline-dev-hot-reload.md`](../design-documents/20260712-pipeline-dev-hot-reload.md)). +`conduit pipelines dev [dir]` is a thin alias that runs `conduit run --dev` with dev-tuned +defaults. + +## What it does on each save + +1. Debounces rapid saves (a save-storm coalesces into one apply). +2. Parses every pipeline in the changed file and runs the same enrich + validate the CLI uses. +3. For each pipeline, computes the diff and applies it: + - A **processor config** or pipeline **name/description** change applies **in place** — the + pipeline keeps running, the source never restarts, positions are continuous, no record is + lost or reordered. + - A **connector**, **DLQ**, or **topology** change applies via a **graceful drain-and-restart** + (the same no-loss stop-drain-restart `apply` uses), and the output says so. +4. Ensures the pipeline is running afterward (a brand-new pipeline file, or one left stopped by a + prior failed apply, is started — unless the config declares it `stopped`). + +Each apply prints whether it was in-place or a restart, the diff, the outcome, and timing; +`--json` emits the same as structured events. + +## The authorization model + +`--dev` is the operator authorization. Applying any change to a running pipeline normally +requires the process-level `--api.allow-live-restart-apply` gate (see +[`live-restart-apply.md`](live-restart-apply.md)); `--dev` authorizes the applies **its own +watcher drives from local file edits**, because the operator ran `--dev`, is watching the +terminal, and sees every diff. It does **not** enable the API/MCP `apply` gate — a remote agent +still can't apply to a running pipeline over the network unless that separate flag is set. + +This is the same trust boundary as `--pipelines.path` provisioning at startup: whoever can edit +the watched files and run `--dev` on this box can already restart these pipelines. + +## Failure modes (by design) + +- **Syntax or validation error on save** → the error is printed with its code and path; the + running pipeline is **untouched** and keeps running the last-good config. Fix and save again. +- **A new config fails at open/start** (e.g. an unreachable source) → the pipeline is left stopped + with the error; the next good save recovers it (ensure-running restarts it). +- **A watched file is deleted** → the pipeline is **left running** and a message is logged; + `--dev` never deletes a running pipeline because a file vanished (e.g. a `git stash`). Stop it + explicitly if that was the intent. +- **`Ctrl-C`** → the watcher is cancelled as part of graceful shutdown; in-flight records drain. + +## Operational notes + +- **Not a production deployment mechanism.** `--dev` is a foreground, single-author, local + convenience. For unattended or remote apply, use the API/MCP `apply` path with its own operator + gate. +- **Startup is normal.** Existing pipelines are provisioned and started exactly as `conduit run` + does; the watcher only handles subsequent edits. An empty pipelines directory is fine — creating + the first file live works. +- **In-place is scoped.** Only processor-config and name/description changes apply without a + restart today; connector/DLQ/topology changes restart. See the design doc for why (the engine + swaps a processor node live but not a source/destination's position/connection state). + +## Related + +- [`docs/design-documents/20260712-pipeline-dev-hot-reload.md`](../design-documents/20260712-pipeline-dev-hot-reload.md) + — design, failure-mode analysis, and the engine PR1 built. +- [`live-restart-apply.md`](live-restart-apply.md) — the process-level gate for the API/MCP apply + path, which `--dev` deliberately does not enable. diff --git a/go.mod b/go.mod index 69c885c15..2ed7e58c2 100644 --- a/go.mod +++ b/go.mod @@ -23,6 +23,7 @@ require ( github.com/conduitio/yaml/v3 v3.3.0 github.com/dop251/goja v0.0.0-20240806095544-3491d4a58fbe github.com/dop251/goja_nodejs v0.0.0-20231122114759-e84d9a924c5c + github.com/fsnotify/fsnotify v1.9.0 github.com/gammazero/deque v1.2.1 github.com/goccy/go-json v0.10.6 github.com/google/go-cmp v0.7.0 @@ -134,7 +135,6 @@ require ( github.com/fatih/color v1.19.0 // indirect github.com/fatih/structtag v1.2.0 // indirect github.com/firefart/nonamedreturns v1.0.5 // indirect - github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/fzipp/gocyclo v0.6.0 // indirect github.com/ghostiam/protogetter v0.3.9 // indirect github.com/go-critic/go-critic v0.12.0 // indirect diff --git a/pkg/conduit/config.go b/pkg/conduit/config.go index f9b7f8679..55bebe799 100644 --- a/pkg/conduit/config.go +++ b/pkg/conduit/config.go @@ -157,6 +157,20 @@ type Config struct { CPUProfile string `long:"dev.cpuprofile" usage:"write CPU profile to file"` MemProfile string `long:"dev.memprofile" usage:"write memory profile to file"` BlockProfile string `long:"dev.blockprofile" usage:"write block profile to file"` + + // Enabled turns on the hot-reload dev watcher (see pkg/conduit/dev + // and docs/design-documents/20260712-pipeline-dev-hot-reload.md §4). + // `conduit run --dev` and `conduit pipelines dev` both set this + // (via the --dev alias, see cmd/conduit/root/run.RunCommand) rather + // than exposing "--dev.enabled" as the primary spelling — it is here, + // under the existing Dev config group, for the same reason the flags + // above are: it is a run-time, --pipelines.path-scoped mode, not a + // distinct subsystem toggle. + Enabled bool `long:"dev.enabled" mapstructure:"enabled" usage:"watch pipelines.path and hot-reload changes into the running engine (see 'conduit run --dev')"` + // JSON selects --json structured dev-watcher event lines (one apply + // = one JSON object) over the default human-readable status line. + // Only meaningful when Enabled. + JSON bool `long:"dev.json" mapstructure:"json" usage:"emit dev-watcher apply events as JSON lines instead of human-readable text"` } } diff --git a/pkg/conduit/dev/apply.go b/pkg/conduit/dev/apply.go new file mode 100644 index 000000000..2504627c6 --- /dev/null +++ b/pkg/conduit/dev/apply.go @@ -0,0 +1,315 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev + +import ( + "bytes" + "context" + "os" + "time" + + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/pipeline" + "github.com/conduitio/conduit/pkg/provisioning/config" + "github.com/conduitio/conduit/pkg/provisioning/config/yaml" +) + +// applyFile is invoked once a debounce window settles for path. It never +// panics and never returns an error — every failure is reported through +// w.reporter and logged, because the Watcher must keep running after a bad +// edit (see this package's doc: an invalid edit never touches a running +// pipeline). +// +// The existence check here is what makes atomic-save (rename-over-the +// -file) editors and a genuine file deletion indistinguishable-in-the +// -moment safe to treat identically: by the time the debounce window (300ms +// default) has elapsed with no further fs events, an atomic save has always +// completed (rename(2) is near-instantaneous), so a file that still doesn't +// exist here is, for all practical purposes, actually gone — see debounce.go +// and the design doc's §4 "Debounce/coalesce". +func (w *Watcher) applyFile(ctx context.Context, path string) { + // A dev apply runs arbitrary, rapidly-edited user YAML through the parser, + // enrich, validate, and the provisioner — and it runs on a plain goroutine + // (see debounce.go). An unrecovered panic there would crash the whole server + // and take every running pipeline down with it: the exact opposite of this + // package's invariant that a bad edit never touches a running pipeline. Contain + // it — report the panic as an error event for this file and keep the server + // (and every live pipeline) up. This is the last-resort backstop; the parse/ + // validate paths are expected to return errors, not panic. + defer func() { + if r := recover(); r != nil { + w.reportRawError(ctx, path, "", cerrors.Errorf("dev: recovered from panic while applying %q: %v", path, r)) + } + }() + + info, err := os.Stat(path) + switch { + case err != nil && os.IsNotExist(err): + w.handleDeleted(ctx, path) + return + case err != nil: + w.reportRawError(ctx, path, "", cerrors.Errorf("could not stat %q: %w", path, err)) + return + case !info.Mode().IsRegular(): + // A directory (or other non-regular entry) landed inside the + // watched directory; nothing to parse. + return + } + + pipelines, transient := w.parseFile(ctx, path) + if transient { + w.logger.Debug(ctx).Str("path", path). + Msg("dev: file is empty or unreadable, assuming a transient atomic-save window; waiting for the next event") + return + } + + if len(pipelines) == 0 { + // Either the file failed to parse/validate entirely (already + // reported by parseFile) or it legitimately defines zero pipelines; + // either way there is nothing to apply, and — critically — nothing + // tracked for future deletion reporting is touched: a file that + // never successfully parsed keeps whatever pipeline IDs (if any) it + // was last known to define. + return + } + + ids := make([]string, 0, len(pipelines)) + for _, p := range pipelines { + ids = append(ids, p.ID) + w.applyPipeline(ctx, path, p) + } + + w.mu.Lock() + w.filePipelines[path] = ids + w.mu.Unlock() +} + +// parseFile runs the same parse -> enrich -> validate pipeline +// deploy.ParseSinglePipeline/provisioning.Service.provisionPipeline use, but +// over every pipeline document path defines (deploy.ParseSinglePipeline +// rejects anything but exactly one pipeline; dev must not, since a +// pipelines-directory file is free to define several). Every parse or +// validate failure is reported individually (via w.reportRawError) and its +// pipeline excluded from the result — a single bad document must not drop +// the valid ones in the same file (matching +// provisioning.Service.parsePipelineConfigFile's rule, #2255). +// +// transient=true means path could not be read at all as meaningful content +// (missing mid-check or empty) — the Watcher treats this as an in-flight +// atomic save rather than an error worth printing; see applyFile's doc. +func (w *Watcher) parseFile(ctx context.Context, path string) (pipelines []config.Pipeline, transient bool) { + data, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return nil, true + } + w.reportRawError(ctx, path, "", cerrors.Errorf("could not read %q: %w", path, err)) + return nil, false + } + if len(bytes.TrimSpace(data)) == 0 { + return nil, true + } + + parser := yaml.NewParser(w.logger) + parsed, err := parser.Parse(ctx, bytes.NewReader(data)) + if err != nil { + // parser.Parse returns a raw cerrors.Join of per-document failures + // alongside any documents that DID parse — walk it directly (do not + // wrap it first) so each document's failure is reported as its own + // event, matching cmd/conduit/internal/validate.validateFile's rule. + cerrors.ForEach(err, func(e error) { w.reportRawError(ctx, path, "", e) }) + } + + if len(parsed) == 0 && err == nil { + w.reportRawError(ctx, path, "", cerrors.Errorf("%q defines no pipelines", path)) + return nil, false + } + + out := make([]config.Pipeline, 0, len(parsed)) + for _, p := range parsed { + enriched := config.Enrich(p) + if verr := config.Validate(enriched); verr != nil { + cerrors.ForEach(verr, func(e error) { w.reportRawError(ctx, path, enriched.ID, e) }) + continue + } + out = append(out, enriched) + } + return out, false +} + +// applyPipeline runs the Plan -> ApplyPlanLive -> ensure-running flow for +// one already-parsed-enriched-validated pipeline and emits exactly one +// Event describing the outcome (OutcomeApplied, OutcomeSkipped, or +// OutcomeError). +func (w *Watcher) applyPipeline(ctx context.Context, path string, desired config.Pipeline) { + start := time.Now() + + plan, err := w.provisioner.Plan(ctx, desired) + if err != nil { + w.reportRawError(ctx, path, desired.ID, cerrors.Errorf("could not plan pipeline %q: %w", desired.ID, err)) + return + } + if plan.Empty() { + w.reporter.Emit(Event{Time: start, Path: path, PipelineID: desired.ID, Outcome: OutcomeSkipped}) + return + } + + wasRunning := false + if w.statusFn != nil { + running, serr := w.statusFn(ctx, desired.ID) + if serr != nil { + w.logger.Debug(ctx).Err(serr).Str("pipeline_id", desired.ID). + Msg("dev: could not determine whether pipeline is currently running; labeling the apply conservatively") + } else { + wasRunning = running + } + } + // Pre-apply expectation, used ONLY as a fallback if the engine does not + // report a ground-truth mode below (an idempotent empty apply detected only + // inside ApplyPlanLive's re-Plan). The authoritative label comes from the + // engine's AppliedMode — the plan-derived guess can be wrong, because a + // live-eligible diff falls back to a restart when a processor cannot be + // swapped in place (e.g. it runs parallel). + expectedMode := ModeProvisioned + if wasRunning { + if plan.LiveEligible() { + expectedMode = ModeInPlace + } else { + expectedMode = ModeRestart + } + } + + // The gate = the interactive invocation: allowRestartOnRunning is always + // true here because `--dev`/`conduit pipelines dev` running at all IS the + // operator authorization ApplyPlanLive's gate exists to require — see + // this package's doc. It does not, and must not, set the process-level + // --api.allow-live-restart-apply flag; that is a separate, independently + // -gated surface. + diff, err := w.provisioner.ApplyPlanLive(ctx, desired, plan.Hash, true) + dur := time.Since(start) + if err != nil { + w.reportRawError(ctx, path, desired.ID, cerrors.Errorf("could not apply pipeline %q: %w", desired.ID, err)) + return + } + + // Report the mode the engine actually applied, not the pre-apply guess. + mode := modeFromApplied(diff.AppliedMode, expectedMode) + + started := w.ensureRunning(ctx, desired) + + // Logged in addition to the reporter's --json/human event stream (Out is + // the CLI-facing surface; this is the same structured log every other + // engine component uses, for operators who watch container/journal logs + // rather than dev's own stdout — and it is what lets a test assert "no + // restart log" without capturing os.Stdout). + w.logger.Info(ctx). + Str("pipeline_id", desired.ID). + Str("path", path). + Str("mode", string(mode)). + Bool("started", started). + Int("changes", len(diff.Changes)). + Dur("duration", dur). + Msg("dev: applied") + + w.reporter.Emit(Event{ + Time: start, + Path: path, + PipelineID: desired.ID, + Outcome: OutcomeApplied, + Mode: mode, + Started: started, + DurationMS: dur.Milliseconds(), + Diff: &diff, + }) +} + +// ensureRunning implements the design doc's §4 "Ensure-running": dev mode +// means "keep my pipelines running", so after a successful apply, if desired +// wants the pipeline running, the Watcher starts it — covering both a +// brand-new file (ApplyPlanLive's not-running branch imports without +// starting) and a pipeline left stopped by a prior failed apply. A config +// declaring config.StatusStopped is always left alone: this function is +// never even called with intent to start it (see the desired.Status check +// below), matching "if the config declares it stopped, dev respects that". +// +// Calling Start on an already-running pipeline (by far the common case — +// most edits target a pipeline ApplyPlanLive just left running, whether via +// an in-place swap or a restart) returns pipeline.ErrPipelineRunning, which +// is treated as success-without-action (started=false), not a failure to +// report: ensure-running's job is "make sure it ends up running", and it +// already is. +func (w *Watcher) ensureRunning(ctx context.Context, desired config.Pipeline) bool { + if desired.Status != config.StatusRunning { + return false + } + + err := w.lifecycle.Start(ctx, desired.ID) + switch { + case err == nil: + return true + case cerrors.Is(err, pipeline.ErrPipelineRunning): + return false + default: + w.reportRawError(ctx, "", desired.ID, cerrors.Errorf("ensure-running: could not start pipeline %q: %w", desired.ID, err)) + return false + } +} + +// handleDeleted reports that path — previously known to define one or more +// pipelines — no longer exists. It never touches those pipelines: they are +// left running exactly as they were (see this package's doc). +func (w *Watcher) handleDeleted(ctx context.Context, path string) { + w.mu.Lock() + ids := w.filePipelines[path] + delete(w.filePipelines, path) + w.mu.Unlock() + + if len(ids) == 0 { + w.logger.Debug(ctx).Str("path", path). + Msg("dev: watched file removed (was not a known pipeline config, or was never successfully applied)") + return + } + + now := time.Now() + for _, id := range ids { + w.logger.Warn(ctx).Str("path", path).Str("pipeline_id", id). + Msg("dev: config file removed; pipeline left running") + w.reporter.Emit(Event{Time: now, Path: path, PipelineID: id, Outcome: OutcomeDeleted}) + } +} + +// reportRawError converts err into an OutcomeError Event, logs it (see +// applyPipeline's success-path logging for why: the same structured log +// every other engine component uses, for operators/tests that watch logs +// rather than dev's own Out stream), and emits the Event. pipelineID may be +// empty (a file-level failure, e.g. an unparseable document, is not yet +// attributable to any one pipeline ID). It never touches the pipeline this +// error names, if any — see this package's doc: an invalid edit is reported, +// not applied. +func (w *Watcher) reportRawError(ctx context.Context, path, pipelineID string, err error) { + info := errorInfoFromErr(err) + w.logger.Warn(ctx). + Str("path", path). + Str("pipeline_id", pipelineID). + Str("code", info.Code). + Msg("dev: " + info.Message) + w.reporter.Emit(Event{ + Time: time.Now(), + Path: path, + PipelineID: pipelineID, + Outcome: OutcomeError, + Error: &info, + }) +} diff --git a/pkg/conduit/dev/apply_test.go b/pkg/conduit/dev/apply_test.go new file mode 100644 index 000000000..9acc762e9 --- /dev/null +++ b/pkg/conduit/dev/apply_test.go @@ -0,0 +1,516 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev + +import ( + "bufio" + "bytes" + "context" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/log" + "github.com/conduitio/conduit/pkg/pipeline" + "github.com/conduitio/conduit/pkg/provisioning" + "github.com/conduitio/conduit/pkg/provisioning/config" + json "github.com/goccy/go-json" + "github.com/matryer/is" +) + +const validPipelineYAML = `version: 2.2 +pipelines: + - id: orders + status: running + name: orders + connectors: + - id: src + type: source + plugin: builtin:generator + - id: dst + type: destination + plugin: builtin:log +` + +const twoPipelinesOneInvalidYAML = `version: 2.2 +pipelines: + - id: good + status: running + connectors: + - id: src + type: source + plugin: builtin:generator + - id: dst + type: destination + plugin: builtin:log + - id: "" + status: bogus-status + connectors: + - id: "" + type: bogus-type +` + +// newTestWatcher builds a Watcher wired to fakes, emitting --json events +// into a buffer for assertion via readEvents. dir is used as the (unused by +// these apply-level tests) watch root. +func newTestWatcher(t *testing.T, dir string, prov *fakeProvisioner, lc *fakeLifecycle, statusFn StatusFunc) (*Watcher, *bytes.Buffer) { + t.Helper() + is := is.New(t) + var buf bytes.Buffer + w, err := New(prov, lc, statusFn, Options{ + Path: dir, + Logger: log.Nop(), + Out: &buf, + JSON: true, + }) + is.NoErr(err) + return w, &buf +} + +// readEvents parses every JSON line in buf into an Event. +func readEvents(t *testing.T, buf *bytes.Buffer) []Event { + t.Helper() + is := is.New(t) + var events []Event + sc := bufio.NewScanner(buf) + for sc.Scan() { + line := sc.Bytes() + if len(bytes.TrimSpace(line)) == 0 { + continue + } + var e Event + is.NoErr(json.Unmarshal(line, &e)) + events = append(events, e) + } + is.NoErr(sc.Err()) + return events +} + +func writeFile(t *testing.T, dir, name, content string) string { + t.Helper() + path := filepath.Join(dir, name) + is := is.New(t) + is.NoErr(os.WriteFile(path, []byte(content), 0o600)) + return path +} + +func TestParseFile_Transient_EmptyFile(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := writeFile(t, dir, "p.yaml", "") + + w, buf := newTestWatcher(t, dir, &fakeProvisioner{}, &fakeLifecycle{}, nil) + pipelines, transient := w.parseFile(context.Background(), path) + is.True(transient) + is.Equal(len(pipelines), 0) + is.Equal(buf.Len(), 0) // no error reported for a transient read +} + +func TestParseFile_Transient_WhitespaceOnly(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := writeFile(t, dir, "p.yaml", " \n\t\n") + + w, buf := newTestWatcher(t, dir, &fakeProvisioner{}, &fakeLifecycle{}, nil) + pipelines, transient := w.parseFile(context.Background(), path) + is.True(transient) + is.Equal(len(pipelines), 0) + is.Equal(buf.Len(), 0) +} + +func TestParseFile_SyntaxError_Reported(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := writeFile(t, dir, "p.yaml", "not: [valid: yaml: at: all:\n - -") + + w, buf := newTestWatcher(t, dir, &fakeProvisioner{}, &fakeLifecycle{}, nil) + pipelines, transient := w.parseFile(context.Background(), path) + is.True(!transient) + is.Equal(len(pipelines), 0) + + events := readEvents(t, buf) + is.True(len(events) >= 1) + is.Equal(events[0].Outcome, OutcomeError) +} + +func TestParseFile_ValidSinglePipeline(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := writeFile(t, dir, "p.yaml", validPipelineYAML) + + w, buf := newTestWatcher(t, dir, &fakeProvisioner{}, &fakeLifecycle{}, nil) + pipelines, transient := w.parseFile(context.Background(), path) + is.True(!transient) + is.Equal(len(pipelines), 1) + is.Equal(pipelines[0].ID, "orders") + is.Equal(buf.Len(), 0) // no errors for a valid file +} + +// TestParseFile_OneBadDocumentDoesNotDropTheGoodOne matches +// provisioning.Service.parsePipelineConfigFile's rule (#2255): a single bad +// pipeline document in a multi-pipeline file must not prevent the valid +// ones from being applied. +func TestParseFile_OneBadDocumentDoesNotDropTheGoodOne(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := writeFile(t, dir, "p.yaml", twoPipelinesOneInvalidYAML) + + w, buf := newTestWatcher(t, dir, &fakeProvisioner{}, &fakeLifecycle{}, nil) + pipelines, transient := w.parseFile(context.Background(), path) + is.True(!transient) + is.Equal(len(pipelines), 1) + is.Equal(pipelines[0].ID, "good") + + events := readEvents(t, buf) + is.True(len(events) >= 1) // the invalid document was reported + for _, e := range events { + is.Equal(e.Outcome, OutcomeError) + } +} + +// TestApplyFile_RecoversFromPanic pins the invariant that a bad edit never +// crashes the server: applyFile runs on a plain goroutine (debounce.go), so a +// panic anywhere in the parse/enrich/validate/apply chain must be contained and +// reported as an error event, not propagated to kill the process (and with it +// every running pipeline). +func TestApplyFile_RecoversFromPanic(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := writeFile(t, dir, "p.yaml", validPipelineYAML) + prov := &fakeProvisioner{ + PlanFn: func(context.Context, config.Pipeline) (provisioning.Diff, error) { panic("boom in plan") }, + } + w, buf := newTestWatcher(t, dir, prov, &fakeLifecycle{}, nil) + + // Must return normally (recovered), not propagate the panic. + w.applyFile(context.Background(), path) + + events := readEvents(t, buf) + is.True(len(events) >= 1) + last := events[len(events)-1] + is.Equal(last.Outcome, OutcomeError) + is.True(last.Error != nil) + is.True(strings.Contains(last.Error.Message, "panic")) // reported as a panic, not swallowed +} + +func TestApplyPipeline_EmptyDiff_Skipped(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + prov := &fakeProvisioner{ + PlanFn: func(_ context.Context, desired config.Pipeline) (provisioning.Diff, error) { + return provisioning.Diff{PipelineID: desired.ID}, nil // Empty() == true + }, + } + lc := &fakeLifecycle{} + w, buf := newTestWatcher(t, dir, prov, lc, nil) + + desired := config.Enrich(config.Pipeline{ID: "orders"}) + w.applyPipeline(context.Background(), "p.yaml", desired) + + is.Equal(prov.applyCallCount(), 0) // ApplyPlanLive must never be called for an empty diff + is.Equal(lc.startCallCount(), 0) + + events := readEvents(t, buf) + is.Equal(len(events), 1) + is.Equal(events[0].Outcome, OutcomeSkipped) +} + +func nonEmptyDiff(id string) provisioning.Diff { + return provisioning.Diff{ + PipelineID: id, + Hash: "deadbeef", + Changes: []provisioning.Change{ + {Resource: provisioning.ResourceProcessor, ID: id + ":proc1", Action: provisioning.ChangeActionUpdate, Effect: provisioning.EffectInPlace, LiveSwappable: true}, + }, + } +} + +func TestApplyPipeline_Mode_InPlace_WhenRunningAndLiveEligible(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + diff := nonEmptyDiff("orders") + prov := &fakeProvisioner{PlanFn: func(_ context.Context, desired config.Pipeline) (provisioning.Diff, error) { return diff, nil }} + lc := &fakeLifecycle{} + statusFn := func(context.Context, string) (bool, error) { return true, nil } + w, buf := newTestWatcher(t, dir, prov, lc, statusFn) + + desired := config.Enrich(config.Pipeline{ID: "orders"}) + w.applyPipeline(context.Background(), "p.yaml", desired) + + is.Equal(prov.applyCallCount(), 1) + is.True(prov.applyCalls[0].allowRestartOnRunning) // the gate = the interactive invocation + + events := readEvents(t, buf) + is.Equal(len(events), 1) + is.Equal(events[0].Outcome, OutcomeApplied) + is.Equal(events[0].Mode, ModeInPlace) +} + +func TestApplyPipeline_Mode_Restart_WhenRunningAndNotLiveEligible(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + diff := provisioning.Diff{ + PipelineID: "orders", + Hash: "deadbeef", + Changes: []provisioning.Change{ + {Resource: provisioning.ResourceConnector, ID: "orders:src", Action: provisioning.ChangeActionUpdate, Effect: provisioning.EffectInPlace, LiveSwappable: false}, + }, + } + prov := &fakeProvisioner{PlanFn: func(_ context.Context, desired config.Pipeline) (provisioning.Diff, error) { return diff, nil }} + lc := &fakeLifecycle{} + statusFn := func(context.Context, string) (bool, error) { return true, nil } + w, buf := newTestWatcher(t, dir, prov, lc, statusFn) + + desired := config.Enrich(config.Pipeline{ID: "orders"}) + w.applyPipeline(context.Background(), "p.yaml", desired) + + events := readEvents(t, buf) + is.Equal(len(events), 1) + is.Equal(events[0].Mode, ModeRestart) +} + +func TestApplyPipeline_Mode_Provisioned_WhenNotRunning(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + diff := nonEmptyDiff("orders") + prov := &fakeProvisioner{PlanFn: func(_ context.Context, desired config.Pipeline) (provisioning.Diff, error) { return diff, nil }} + lc := &fakeLifecycle{} + statusFn := func(context.Context, string) (bool, error) { return false, nil } + w, buf := newTestWatcher(t, dir, prov, lc, statusFn) + + desired := config.Enrich(config.Pipeline{ID: "orders"}) // Status defaults to running + w.applyPipeline(context.Background(), "p.yaml", desired) + + events := readEvents(t, buf) + is.Equal(len(events), 1) + is.Equal(events[0].Mode, ModeProvisioned) + is.True(events[0].Started) // ensure-running started it + is.Equal(lc.startCallCount(), 1) +} + +// TestApplyPipeline_Mode_EngineOutcomeOverridesPlanGuess is the regression test +// for the mislabeled-mode bug: a processor edit produces a live-eligible plan, +// so the pre-apply guess is in_place — but the engine fell back to a restart +// (the processor could not be swapped live, e.g. it runs parallel), reported via +// Diff.AppliedMode. The event MUST report the engine's ground truth (restart), +// not the plan-derived guess (in_place). Without honoring AppliedMode, dev would +// tell the operator "applied in place, no restart" while the pipeline actually +// restarted. +func TestApplyPipeline_Mode_EngineOutcomeOverridesPlanGuess(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + planned := nonEmptyDiff("orders") // live-eligible: guess would be in_place + is.True(planned.LiveEligible()) + applied := planned + applied.AppliedMode = provisioning.ApplyModeRestart // engine actually fell back to a restart + prov := &fakeProvisioner{ + PlanFn: func(_ context.Context, _ config.Pipeline) (provisioning.Diff, error) { return planned, nil }, + ApplyFn: func(_ context.Context, _ config.Pipeline, _ string, _ bool) (provisioning.Diff, error) { + return applied, nil + }, + } + lc := &fakeLifecycle{} + statusFn := func(context.Context, string) (bool, error) { return true, nil } // running -> guess=in_place + w, buf := newTestWatcher(t, dir, prov, lc, statusFn) + + desired := config.Enrich(config.Pipeline{ID: "orders"}) + w.applyPipeline(context.Background(), "p.yaml", desired) + + events := readEvents(t, buf) + is.Equal(len(events), 1) + is.Equal(events[0].Mode, ModeRestart) // engine's AppliedMode wins over the in_place guess +} + +// --- ensure-running --- + +func TestEnsureRunning_NewFile_Starts(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + lc := &fakeLifecycle{} + w, _ := newTestWatcher(t, dir, &fakeProvisioner{}, lc, nil) + + desired := config.Enrich(config.Pipeline{ID: "orders"}) // Status: running + started := w.ensureRunning(context.Background(), desired) + + is.True(started) + is.Equal(lc.startCallCount(), 1) + is.Equal(lc.startCalls[0], "orders") +} + +func TestEnsureRunning_PostFailureStopped_Starts(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + lc := &fakeLifecycle{} // Start succeeds — pipeline was left stopped by a prior failed apply + w, _ := newTestWatcher(t, dir, &fakeProvisioner{}, lc, nil) + + desired := config.Enrich(config.Pipeline{ID: "orders"}) + started := w.ensureRunning(context.Background(), desired) + + is.True(started) + is.Equal(lc.startCallCount(), 1) +} + +func TestEnsureRunning_AlreadyRunning_NoErrorNoDoubleStart(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + lc := &fakeLifecycle{StartFn: func(context.Context, string) error { return pipeline.ErrPipelineRunning }} + w, buf := newTestWatcher(t, dir, &fakeProvisioner{}, lc, nil) + + desired := config.Enrich(config.Pipeline{ID: "orders"}) + started := w.ensureRunning(context.Background(), desired) + + is.True(!started) // already running: nothing to report as "started" + is.Equal(buf.Len(), 0) // and definitely not an error +} + +func TestEnsureRunning_StatusStopped_NeverStarts(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + lc := &fakeLifecycle{} + w, _ := newTestWatcher(t, dir, &fakeProvisioner{}, lc, nil) + + desired := config.Enrich(config.Pipeline{ID: "orders", Status: config.StatusStopped}) + started := w.ensureRunning(context.Background(), desired) + + is.True(!started) + is.Equal(lc.startCallCount(), 0) // config says stopped: dev must respect that +} + +func TestEnsureRunning_StartFails_Reported(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + wantErr := cerrors.New("boom") + lc := &fakeLifecycle{StartFn: func(context.Context, string) error { return wantErr }} + w, buf := newTestWatcher(t, dir, &fakeProvisioner{}, lc, nil) + + desired := config.Enrich(config.Pipeline{ID: "orders"}) + started := w.ensureRunning(context.Background(), desired) + + is.True(!started) + events := readEvents(t, buf) + is.Equal(len(events), 1) + is.Equal(events[0].Outcome, OutcomeError) +} + +// --- file deletion --- + +func TestHandleDeleted_KnownFile_ReportsAndLeavesRunning(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + prov := &fakeProvisioner{} + lc := &fakeLifecycle{} + w, buf := newTestWatcher(t, dir, prov, lc, nil) + + w.mu.Lock() + w.filePipelines["orders.yaml"] = []string{"orders"} + w.mu.Unlock() + + w.handleDeleted(context.Background(), "orders.yaml") + + // Never touches the pipeline: no Plan/Apply/Start call as a result of a + // deletion. + is.Equal(prov.applyCallCount(), 0) + is.Equal(lc.startCallCount(), 0) + + events := readEvents(t, buf) + is.Equal(len(events), 1) + is.Equal(events[0].Outcome, OutcomeDeleted) + is.Equal(events[0].PipelineID, "orders") + + // The path is forgotten so a later re-creation starts fresh. + w.mu.Lock() + _, tracked := w.filePipelines["orders.yaml"] + w.mu.Unlock() + is.True(!tracked) +} + +func TestHandleDeleted_UnknownFile_NoEvent(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + w, buf := newTestWatcher(t, dir, &fakeProvisioner{}, &fakeLifecycle{}, nil) + + w.handleDeleted(context.Background(), "never-applied.yaml") + + is.Equal(buf.Len(), 0) +} + +// --- applyFile end to end (in-process, no fsnotify) --- + +func TestApplyFile_TracksPipelineIDsOnSuccess(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := writeFile(t, dir, "orders.yaml", validPipelineYAML) + + diff := nonEmptyDiff("orders") + prov := &fakeProvisioner{PlanFn: func(_ context.Context, desired config.Pipeline) (provisioning.Diff, error) { return diff, nil }} + lc := &fakeLifecycle{} + w, buf := newTestWatcher(t, dir, prov, lc, nil) + + w.applyFile(context.Background(), path) + + events := readEvents(t, buf) + is.Equal(len(events), 1) + is.Equal(events[0].Outcome, OutcomeApplied) + is.Equal(events[0].PipelineID, "orders") + + w.mu.Lock() + ids := w.filePipelines[path] + w.mu.Unlock() + is.Equal(len(ids), 1) + is.Equal(ids[0], "orders") +} + +func TestApplyFile_Deleted_LeavesTrackedPipelineRunning(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := filepath.Join(dir, "orders.yaml") // never created + + prov := &fakeProvisioner{} + lc := &fakeLifecycle{} + w, buf := newTestWatcher(t, dir, prov, lc, nil) + w.mu.Lock() + w.filePipelines[path] = []string{"orders"} + w.mu.Unlock() + + w.applyFile(context.Background(), path) + + is.Equal(prov.applyCallCount(), 0) + is.Equal(lc.startCallCount(), 0) + + events := readEvents(t, buf) + is.Equal(len(events), 1) + is.Equal(events[0].Outcome, OutcomeDeleted) +} + +func TestApplyFile_ParseError_NeverCallsApply(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := writeFile(t, dir, "orders.yaml", "not: [valid: yaml:\n -") + + prov := &fakeProvisioner{} + lc := &fakeLifecycle{} + w, buf := newTestWatcher(t, dir, prov, lc, nil) + + w.applyFile(context.Background(), path) + + is.Equal(prov.applyCallCount(), 0) + is.Equal(lc.startCallCount(), 0) + + events := readEvents(t, buf) + is.True(len(events) >= 1) + is.Equal(events[0].Outcome, OutcomeError) +} diff --git a/pkg/conduit/dev/debounce.go b/pkg/conduit/dev/debounce.go new file mode 100644 index 000000000..4d9443ff3 --- /dev/null +++ b/pkg/conduit/dev/debounce.go @@ -0,0 +1,151 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev + +import ( + "context" + "time" +) + +// DefaultDebounce is the debounce window used when Options.Debounce is zero +// (design doc §4: "a 300ms debounce collapses save-storms"). +const DefaultDebounce = 300 * time.Millisecond + +// debouncer coalesces a burst of trigger() calls for one file path into a +// single call to apply, run after delay has elapsed with no further +// triggers (a classic trailing-edge debounce), and ensures at most one +// apply runs at a time for that path — see run's doc for the two rules +// this implements. +// +// One debouncer exists per watched file path (see Watcher.consume, which +// creates them lazily) and owns exactly one goroutine (run), so state is +// unsynchronized by design: trigger only ever sends on a channel, and run is +// the only goroutine that reads it or mutates the debouncer's fields. +type debouncer struct { + clock Clock + delay time.Duration + apply func(ctx context.Context) + + // triggerCh is buffered to 1 so trigger() never blocks the caller (the + // single-goroutine event loop in Watcher.consume): a burst of N events + // collapses to at most one buffered signal, which is exactly the + // coalescing debounce wants. + triggerCh chan struct{} +} + +func newDebouncer(clock Clock, delay time.Duration, apply func(ctx context.Context)) *debouncer { + return &debouncer{ + clock: clock, + delay: delay, + apply: apply, + triggerCh: make(chan struct{}, 1), + } +} + +// trigger records that a relevant fs event happened for this debouncer's +// path. It never blocks: if a trigger is already pending (the buffered +// channel is full), this call is a no-op — one pending trigger is exactly as +// informative as several, since run's debounce window restarts on every +// trigger it observes regardless of count. +func (d *debouncer) trigger() { + select { + case d.triggerCh <- struct{}{}: + default: + } +} + +// run is the debouncer's only goroutine. It implements two rules from the +// design doc's §4 "Debounce/coalesce": +// +// 1. A burst of triggers collapses to one apply, run only after delay has +// passed with no further triggers (each new trigger resets the window — +// "quiet for delay" is the condition, not "delay after the first +// trigger"). +// 2. A trigger that arrives while an apply is already in flight for this +// path is coalesced into at most one queued follow-up, run (after +// another debounce window) once the in-flight apply completes — never a +// pile of queued applies, and never a second apply running concurrently +// with the first for the same path. +// +// run returns when ctx is cancelled (invariant 7: tied to the serve context). +// Crucially, on cancellation it does NOT return while an apply it started is +// still in flight: an apply mutates engine + DB state (a pipeline restart or an +// in-place processor swap), so it must complete before run returns and, above +// it, Watcher.consume's wg.Wait unblocks and Watcher.Run returns — otherwise +// the runtime would proceed to tear the engine and database down underneath a +// still-running apply. The apply goroutine always signals done exactly once +// when d.apply returns (done is buffered to 1 so that send never blocks), and +// run drains it exactly once per apply — either via the normal <-done case or, +// on shutdown, via the explicit wait below. That balance guarantees no leak and +// no deadlock. +func (d *debouncer) run(ctx context.Context) { + var timerC <-chan time.Time + applying := false + queued := false + done := make(chan struct{}, 1) + + for { + select { + case <-ctx.Done(): + // Invariant 7 (graceful shutdown): wait out the in-flight apply so + // it never races the runtime's engine + DB teardown. ctx is already + // cancelled, so d.apply(ctx) returns promptly; we just don't return + // before it does. + if applying { + <-done + } + return + + case <-d.triggerCh: + if applying { + // Rule 2: coalesce into a single queued follow-up. + queued = true + continue + } + // Rule 1: (re)start the debounce window. Replacing timerC here is + // intentional even if a previous window was already pending — a + // new event means "not quiet yet", so the window must restart. + timerC = d.clock.After(d.delay) + + case <-timerC: + timerC = nil + if ctx.Err() != nil { + // Shutting down: both ctx.Done and this timer can be ready at + // once and select may pick the timer. Don't kick off a fresh + // (possibly restart-class) apply during teardown — loop back so + // the ctx.Done case returns. (An aborted apply is recoverable + // from checkpoint on next start; not starting one is cleaner.) + continue + } + applying = true + go func() { + d.apply(ctx) + // Never blocks: done is buffered to 1 and run drains exactly one + // signal per apply it starts (normal case or shutdown wait). + done <- struct{}{} + }() + + case <-done: + applying = false + if queued { + queued = false + // One more debounce window, not an immediate re-apply: a + // save that lands the instant the in-flight apply finishes + // deserves the same coalescing window as any other burst. + timerC = d.clock.After(d.delay) + } + } + } +} diff --git a/pkg/conduit/dev/debounce_test.go b/pkg/conduit/dev/debounce_test.go new file mode 100644 index 000000000..f4c44b080 --- /dev/null +++ b/pkg/conduit/dev/debounce_test.go @@ -0,0 +1,220 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev + +import ( + "context" + "testing" + "time" + + "github.com/matryer/is" +) + +// awaitApply blocks until applied receives a value, failing the test after a +// generous timeout instead of hanging forever if the debouncer has a bug. +func awaitApply(t *testing.T, applied <-chan struct{}) { + t.Helper() + select { + case <-applied: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for apply to run") + } +} + +// assertNoApply asserts apply is NOT called within a short window — used to +// prove coalescing actually suppressed extra applies, not just that they +// haven't happened yet. +func assertNoApply(t *testing.T, applied <-chan struct{}) { + t.Helper() + select { + case <-applied: + t.Fatal("apply ran, expected it to be coalesced") + case <-time.After(50 * time.Millisecond): + } +} + +func TestDebouncer_SingleTrigger_AppliesOnce(t *testing.T) { + is := is.New(t) + clock := newFakeClock() + applied := make(chan struct{}, 8) + + d := newDebouncer(clock, time.Second, func(context.Context) { applied <- struct{}{} }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go d.run(ctx) + + d.trigger() + clock.awaitCall() + clock.fireLatest() + + awaitApply(t, applied) + assertNoApply(t, applied) // exactly one apply, not more + is.True(true) +} + +func TestDebouncer_Burst_CollapsesToOneApply(t *testing.T) { + clock := newFakeClock() + applied := make(chan struct{}, 8) + + d := newDebouncer(clock, time.Second, func(context.Context) { applied <- struct{}{} }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go d.run(ctx) + + // A burst of 5 triggers, each resetting the debounce window (as a real + // save-storm would): every trigger causes a fresh After call. + const burst = 5 + for i := 0; i < burst; i++ { + d.trigger() + clock.awaitCall() + } + // Only the LAST window should ever fire in practice (a real clock would + // have superseded the earlier ones); fire it and confirm exactly one + // apply happens. + clock.fireLatest() + + awaitApply(t, applied) + assertNoApply(t, applied) +} + +func TestDebouncer_TriggerDuringApply_QueuesExactlyOneFollowUp(t *testing.T) { + is := is.New(t) + clock := newFakeClock() + applied := make(chan struct{}, 8) + release := make(chan struct{}) + + callCount := 0 + d := newDebouncer(clock, time.Second, func(context.Context) { + callCount++ + applied <- struct{}{} + <-release // block "in flight" until the test lets it finish + }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go d.run(ctx) + + // First apply starts. + d.trigger() + clock.awaitCall() + clock.fireLatest() + awaitApply(t, applied) + + // While the first apply is still in flight (blocked on release), fire a + // storm of triggers — design doc: "an in-flight apply queues at most one + // follow-up". + d.trigger() + d.trigger() + d.trigger() + + // Let the first apply finish. + close(release) + + // Exactly one more debounce window should be requested (the queued + // follow-up), then exactly one more apply. + clock.awaitCall() + clock.fireLatest() + awaitApply(t, applied) + assertNoApply(t, applied) + + is.Equal(callCount, 2) +} + +func TestDebouncer_NoTriggerDuringApply_NoFollowUp(t *testing.T) { + clock := newFakeClock() + applied := make(chan struct{}, 8) + + d := newDebouncer(clock, time.Second, func(context.Context) { applied <- struct{}{} }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go d.run(ctx) + + d.trigger() + clock.awaitCall() + clock.fireLatest() + awaitApply(t, applied) + + // No trigger arrived during (or after) the apply; nothing more should + // ever run. + assertNoApply(t, applied) +} + +func TestDebouncer_StopsOnContextCancel(t *testing.T) { + clock := newFakeClock() + applied := make(chan struct{}, 8) + done := make(chan struct{}) + + d := newDebouncer(clock, time.Second, func(context.Context) { applied <- struct{}{} }) + ctx, cancel := context.WithCancel(context.Background()) + go func() { + d.run(ctx) + close(done) + }() + + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("debouncer.run did not return after context cancellation") + } +} + +// TestDebouncer_ContextCancel_WaitsForInFlightApply is the regression test for +// the shutdown race: run must NOT return while an apply it started is still +// executing, because that apply mutates engine + DB state and Watcher.consume's +// wg.Wait (which only tracks run, not the apply goroutine run spawns) would +// otherwise unblock, Watcher.Run would return, and the runtime would tear the +// engine and database down underneath the still-running apply. run owns the +// apply goroutine's lifetime: on cancel it blocks until the apply returns. +func TestDebouncer_ContextCancel_WaitsForInFlightApply(t *testing.T) { + clock := newFakeClock() + started := make(chan struct{}) + release := make(chan struct{}) + finished := make(chan struct{}) + + d := newDebouncer(clock, time.Second, func(context.Context) { + close(started) + <-release // hold the apply "in flight" until the test releases it + close(finished) + }) + ctx, cancel := context.WithCancel(context.Background()) + runReturned := make(chan struct{}) + go func() { + d.run(ctx) + close(runReturned) + }() + + // Get an apply in flight, then cancel while it is still blocked. + d.trigger() + clock.awaitCall() + clock.fireLatest() + <-started + cancel() + + // run must still be blocked on the in-flight apply — it has NOT returned. + select { + case <-runReturned: + t.Fatal("run returned while an apply was still in flight — shutdown race") + case <-time.After(100 * time.Millisecond): + } + + // Let the apply finish; only now may run return. + close(release) + <-finished + select { + case <-runReturned: + case <-time.After(5 * time.Second): + t.Fatal("run did not return after the in-flight apply completed") + } +} diff --git a/pkg/conduit/dev/doc.go b/pkg/conduit/dev/doc.go new file mode 100644 index 000000000..d85b15447 --- /dev/null +++ b/pkg/conduit/dev/doc.go @@ -0,0 +1,79 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package dev implements the file watcher behind `conduit run --dev` (see +// docs/design-documents/20260712-pipeline-dev-hot-reload.md, §4 "Surface"). +// It is PR2 of that design: PR1 (merged) added the engine capability this +// package drives — pkg/provisioning.Service.Plan/ApplyPlanLive, which now +// applies a live-eligible diff to a running pipeline in place (a processor +// swap, no restart) and drain-restarts everything else. This package adds no +// new engine behavior; it is purely a debounced directory watcher that turns +// "a file changed on disk" into "call Plan, then ApplyPlanLive" and reports +// the outcome. +// +// # Where this lives +// +// The Watcher is constructed and started by pkg/conduit.Runtime.Run, after +// provisioning.Service.Init has provisioned and started whatever pipelines +// exist at startup — the watcher only ever reacts to *subsequent* edits, and +// is tied to the same serve context startup uses, so Ctrl-C/SIGTERM cancels +// it along with everything else (invariant 7). It is not CLI-layer code: +// cmd/conduit/root/run.RunCommand only flips conduit.Config.Dev.Enabled; +// every byte of watch/debounce/apply logic lives here so the CLI and the +// `conduit pipelines dev` alias can never drift from what the runtime +// actually does. See the design doc's "Three-faces rule": dev is +// legitimately CLI-only — an agent automates the same effect by calling +// ApplyPipeline directly, not by asking a Conduit instance to watch files. +// +// # The operator-authorization gate +// +// ApplyPlanLive refuses to touch a running pipeline unless its caller passes +// allowRestartOnRunning=true (see pkg/provisioning/plan.go's doc on +// CodeLiveApplyUnauthorized) — a Tier-1 data-path gate that exists so an +// unattended/remote caller can never silently restart a live pipeline. The +// Watcher always passes true for the applies IT drives, and only those: the +// human explicitly ran `conduit run --dev` (or `conduit pipelines dev`) and +// is watching each diff scroll by as they save — that interactive act *is* +// the authorization the gate exists to require. This is deliberately +// independent of, and never sets, the process-level +// --api.allow-live-restart-apply flag: the gRPC/HTTP/MCP surface stays gated +// exactly as it was before dev mode existed. +// +// # Ensure-running +// +// ApplyPlanLive's not-currently-running branch imports a pipeline's new +// config without starting it (the correct behavior for provisioning at +// startup, and for a one-shot `conduit pipelines apply`). Dev mode means +// "keep my pipelines running while I iterate", so after every successful +// apply the Watcher calls Start if, and only if, the just-applied config +// wants the pipeline running (config.StatusRunning) — covering a +// brand-new file, and a pipeline left stopped by a prior failed apply. If +// the config declares config.StatusStopped, the Watcher never starts it. +// Start is idempotent from the Watcher's point of view: calling it on an +// already-running pipeline (the common case — most edits target a pipeline +// that ApplyPlanLive already left running, whether via an in-place swap or a +// restart) returns pipeline.ErrPipelineRunning, which the Watcher treats as +// success, not a failure to report. +// +// # Invariants +// +// - Invariant 7 (graceful shutdown): every long-running goroutine this +// package starts is derived from, and exits on cancellation of, the +// context passed to Watcher.Run. +// - An invalid edit never touches a running pipeline: a file that fails to +// parse, or a pipeline document that fails config.Enrich/config.Validate, +// is reported and skipped — Plan/ApplyPlanLive are never called for it. +// - A deleted watched file never deletes the pipeline it described: the +// Watcher only logs/reports it and leaves the pipeline exactly as it was. +package dev diff --git a/pkg/conduit/dev/events.go b/pkg/conduit/dev/events.go new file mode 100644 index 000000000..1b2787483 --- /dev/null +++ b/pkg/conduit/dev/events.go @@ -0,0 +1,239 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev + +import ( + "fmt" + "io" + "strings" + "sync" + "time" + + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" + "github.com/conduitio/conduit/pkg/provisioning" + json "github.com/goccy/go-json" +) + +// Outcome classifies what happened for one file/pipeline the Watcher looked +// at after a debounced fs event. +type Outcome string + +const ( + // OutcomeApplied means Plan computed a non-empty diff and ApplyPlanLive + // applied it successfully (Mode says how). + OutcomeApplied Outcome = "applied" + // OutcomeSkipped means Plan computed an empty diff — the file already + // matches the running pipeline's state; nothing was applied. + OutcomeSkipped Outcome = "skipped" + // OutcomeError means a parse, validation, Plan, or ApplyPlanLive error + // occurred; the pipeline this event names (if any) was left untouched. + OutcomeError Outcome = "error" + // OutcomeDeleted means the watched file was removed; the pipeline it + // last described is left running, never auto-deleted. + OutcomeDeleted Outcome = "deleted" +) + +// Mode classifies *how* an OutcomeApplied event was applied. It is taken from +// the engine's ground-truth provisioning.Diff.AppliedMode (the path +// ApplyPlanLive actually took), falling back to the pre-apply running status +// (StatusFunc) + Diff.LiveEligible() guess only when the engine reports no mode +// — see modeFromApplied and pkg/provisioning/plan.go's ApplyMode. +type Mode string + +const ( + // ModeInPlace: the pipeline was running and every change in the diff was + // live-swappable — applied via a processor swap / metadata update, no + // restart. + ModeInPlace Mode = "in_place" + // ModeRestart: the pipeline was running and at least one change was not + // live-swappable — applied via ApplyPlanLive's graceful drain-and + // -restart. + ModeRestart Mode = "restart" + // ModeProvisioned: the pipeline was not running before this apply (a + // brand-new pipeline, or one left stopped by config/a prior failed + // apply) — its config was imported without disrupting anything, and + // ensure-running may have started it (see Event.Started). + ModeProvisioned Mode = "provisioned" +) + +// modeFromApplied maps the engine's ground-truth provisioning.ApplyMode to the +// dev watcher's reported Mode. The engine's value is authoritative — it reflects +// the path ApplyPlanLive actually took, including an in-place->restart fallback +// the pre-apply plan cannot foresee. fallback is used only when the engine +// reports ApplyModeUnknown (it ran no mutating apply, e.g. an idempotent empty +// diff detected only inside ApplyPlanLive's re-Plan), where the caller's own +// pre-apply expectation is the best label available. +func modeFromApplied(applied provisioning.ApplyMode, fallback Mode) Mode { + switch applied { + case provisioning.ApplyModeInPlace: + return ModeInPlace + case provisioning.ApplyModeRestart: + return ModeRestart + case provisioning.ApplyModeProvisioned: + return ModeProvisioned + case provisioning.ApplyModeUnknown: + return fallback + default: // an unrecognized future mode: fall back rather than guess wrong + return fallback + } +} + +// ErrorInfo is the --json rendering of an error the Watcher reported — +// mirroring the *conduiterr.ConduitError fields the rest of the CLI already +// exposes (see cmd/conduit/internal/validate.Finding), so a --json consumer +// gets the same stable code/configPath/suggestion shape everywhere. +type ErrorInfo struct { + Code string `json:"code,omitempty"` + Message string `json:"message"` + ConfigPath string `json:"configPath,omitempty"` + Suggestion string `json:"suggestion,omitempty"` +} + +// errorInfoFromErr converts err into an ErrorInfo, preserving code/ +// configPath/suggestion when err is (or wraps) a *conduiterr.ConduitError, +// and falling back to its plain message otherwise. Mirrors +// cmd/conduit/internal/validate.findingFromError's error-shaping rule, but +// that function lives under cmd/conduit/internal and can't be imported from +// pkg (internal package boundary), so it is duplicated here at a fraction of +// the size (this package doesn't need Findings, just this one conversion). +func errorInfoFromErr(err error) ErrorInfo { + if ce, ok := conduiterr.Get(err); ok { + return ErrorInfo{ + Code: ce.Code.Reason(), + Message: ce.Message, + ConfigPath: ce.ConfigPath, + Suggestion: ce.Suggestion, + } + } + return ErrorInfo{Message: err.Error()} +} + +// Event is one line of the Watcher's output — either a --json line or a +// human-readable status line, chosen by Reporter's json flag. Exactly one +// Event is emitted per debounced (file, pipeline) apply attempt, per file +// deletion, and per parse/validate failure. +type Event struct { + Time time.Time `json:"time"` + Path string `json:"path"` + PipelineID string `json:"pipelineID,omitempty"` + Outcome Outcome `json:"outcome"` + Mode Mode `json:"mode,omitempty"` + Started bool `json:"started,omitempty"` + DurationMS int64 `json:"durationMs,omitempty"` + Diff *provisioning.Diff `json:"diff,omitempty"` + Error *ErrorInfo `json:"error,omitempty"` +} + +// Reporter serializes Event output to Out, either as JSON lines (one +// compact JSON object per line) or as human-readable text — concurrency +// -safe (Writer is guarded by mu) because applies for different files can +// complete concurrently (each file's debouncer runs its apply in its own +// goroutine; see debounce.go), and unsynchronized concurrent writes to Out +// would interleave partial lines. +type Reporter struct { + mu sync.Mutex + out io.Writer + json bool +} + +func newReporter(out io.Writer, jsonLines bool) *Reporter { + return &Reporter{out: out, json: jsonLines} +} + +// Emit writes one Event to the reporter's Out, in whichever format the +// reporter was configured for. Marshal/format failures are never fatal to +// the Watcher (dev's job is to keep the pipeline running, not to guarantee +// every status line lands) — they are swallowed here after a best-effort +// fallback write. +func (r *Reporter) Emit(e Event) { + r.mu.Lock() + defer r.mu.Unlock() + + if r.json { + b, err := json.Marshal(e) + if err != nil { + // Event is a plain struct of strings/bools/ints plus an + // additive *provisioning.Diff and *ErrorInfo, both themselves + // plain JSON-tagged structs — Marshal cannot fail for it in + // practice. Fall back to a minimal, always-marshalable line + // rather than drop the event silently. + fmt.Fprintf(r.out, `{"time":%q,"outcome":"error","error":{"message":"could not marshal dev event"}}`+"\n", e.Time.Format(time.RFC3339)) + return + } + fmt.Fprintln(r.out, string(b)) + return + } + + fmt.Fprintln(r.out, renderHuman(e)) +} + +// renderHuman renders e as one human-readable status line. +func renderHuman(e Event) string { + var b strings.Builder + fmt.Fprint(&b, "[dev] ") + if e.PipelineID != "" { + fmt.Fprintf(&b, "%s: ", e.PipelineID) + } + + switch e.Outcome { + case OutcomeApplied: + fmt.Fprint(&b, applyVerb(e.Mode)) + if e.Started { + fmt.Fprint(&b, ", started") + } + if e.Diff != nil { + fmt.Fprintf(&b, " (%d change(s))", len(e.Diff.Changes)) + } + fmt.Fprintf(&b, " — %s", e.Path) + if e.DurationMS > 0 { + fmt.Fprintf(&b, " (%dms)", e.DurationMS) + } + case OutcomeSkipped: + fmt.Fprintf(&b, "no changes — %s", e.Path) + case OutcomeDeleted: + fmt.Fprintf(&b, "config file removed (%s); pipeline left running", e.Path) + case OutcomeError: + b.Reset() + fmt.Fprintf(&b, "[dev] ERROR %s", e.Path) + if e.PipelineID != "" { + fmt.Fprintf(&b, " (%s)", e.PipelineID) + } + if e.Error != nil { + fmt.Fprintf(&b, ": %s", e.Error.Message) + if e.Error.Code != "" { + fmt.Fprintf(&b, " [%s]", e.Error.Code) + } + if e.Error.Suggestion != "" { + fmt.Fprintf(&b, " — %s", e.Error.Suggestion) + } + } + default: + fmt.Fprintf(&b, "%s — %s", e.Outcome, e.Path) + } + return b.String() +} + +func applyVerb(m Mode) string { + switch m { + case ModeInPlace: + return "applied in place" + case ModeRestart: + return "applied via restart" + case ModeProvisioned: + return "provisioned" + default: + return "applied" + } +} diff --git a/pkg/conduit/dev/events_test.go b/pkg/conduit/dev/events_test.go new file mode 100644 index 000000000..e9dadec59 --- /dev/null +++ b/pkg/conduit/dev/events_test.go @@ -0,0 +1,138 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev + +import ( + "bytes" + "strings" + "testing" + + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" + "github.com/conduitio/conduit/pkg/provisioning" + json "github.com/goccy/go-json" + "github.com/matryer/is" +) + +func TestReporter_Human_AppliedInPlace(t *testing.T) { + is := is.New(t) + var buf bytes.Buffer + r := newReporter(&buf, false) + + r.Emit(Event{ + Path: "orders.yaml", + PipelineID: "orders", + Outcome: OutcomeApplied, + Mode: ModeInPlace, + DurationMS: 12, + Diff: &provisioning.Diff{Changes: []provisioning.Change{{}}}, + }) + + out := buf.String() + is.True(strings.Contains(out, "orders")) + is.True(strings.Contains(out, "applied in place")) + is.True(strings.Contains(out, "orders.yaml")) + is.True(strings.Contains(out, "12ms")) +} + +func TestReporter_Human_Restart(t *testing.T) { + is := is.New(t) + var buf bytes.Buffer + r := newReporter(&buf, false) + + r.Emit(Event{PipelineID: "orders", Path: "orders.yaml", Outcome: OutcomeApplied, Mode: ModeRestart}) + + is.True(strings.Contains(buf.String(), "applied via restart")) +} + +func TestReporter_Human_Error(t *testing.T) { + is := is.New(t) + var buf bytes.Buffer + r := newReporter(&buf, false) + + info := ErrorInfo{Code: "config.field_required", Message: `"id" is mandatory`, Suggestion: "set an id"} + r.Emit(Event{Path: "orders.yaml", Outcome: OutcomeError, Error: &info}) + + out := buf.String() + is.True(strings.Contains(out, "ERROR")) + is.True(strings.Contains(out, "orders.yaml")) + is.True(strings.Contains(out, `"id" is mandatory`)) + is.True(strings.Contains(out, "config.field_required")) + is.True(strings.Contains(out, "set an id")) +} + +func TestReporter_Human_Deleted(t *testing.T) { + is := is.New(t) + var buf bytes.Buffer + r := newReporter(&buf, false) + + r.Emit(Event{Path: "orders.yaml", PipelineID: "orders", Outcome: OutcomeDeleted}) + + out := buf.String() + is.True(strings.Contains(out, "removed")) + is.True(strings.Contains(out, "left running")) +} + +func TestReporter_JSON_RoundTrips(t *testing.T) { + is := is.New(t) + var buf bytes.Buffer + r := newReporter(&buf, true) + + want := Event{ + Path: "orders.yaml", + PipelineID: "orders", + Outcome: OutcomeApplied, + Mode: ModeInPlace, + Started: true, + DurationMS: 42, + } + r.Emit(want) + + var got Event + is.NoErr(json.Unmarshal(buf.Bytes(), &got)) + is.Equal(got.Path, want.Path) + is.Equal(got.PipelineID, want.PipelineID) + is.Equal(got.Outcome, want.Outcome) + is.Equal(got.Mode, want.Mode) + is.Equal(got.Started, want.Started) + is.Equal(got.DurationMS, want.DurationMS) + + // Exactly one line — a --json consumer parses one event per line. + is.Equal(strings.Count(buf.String(), "\n"), 1) +} + +func TestErrorInfoFromErr_ConduitError_PreservesFields(t *testing.T) { + is := is.New(t) + code := conduiterr.Register("dev_test.example", 0) + ce := conduiterr.New(code, "boom") + ce.ConfigPath = "/pipelines/0/id" + ce.Suggestion = "fix it" + + info := errorInfoFromErr(ce) + is.Equal(info.Code, "dev_test.example") + is.Equal(info.Message, "boom") + is.Equal(info.ConfigPath, "/pipelines/0/id") + is.Equal(info.Suggestion, "fix it") +} + +func TestErrorInfoFromErr_PlainError_MessageOnly(t *testing.T) { + is := is.New(t) + info := errorInfoFromErr(plainTestError("boom")) + is.Equal(info.Message, "boom") + is.Equal(info.Code, "") +} + +type plainTestError string + +func (e plainTestError) Error() string { return string(e) } diff --git a/pkg/conduit/dev/fakes_test.go b/pkg/conduit/dev/fakes_test.go new file mode 100644 index 000000000..18bb0f7e6 --- /dev/null +++ b/pkg/conduit/dev/fakes_test.go @@ -0,0 +1,146 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev + +import ( + "context" + "sync" + "time" + + "github.com/conduitio/conduit/pkg/provisioning" + "github.com/conduitio/conduit/pkg/provisioning/config" +) + +// fakeProvisioner is a test double for Provisioner: PlanFn/ApplyFn are +// called if set, otherwise Plan returns an empty Diff and ApplyPlanLive +// returns planResult unchanged. Every call is recorded for assertions. +type fakeProvisioner struct { + mu sync.Mutex + + PlanFn func(ctx context.Context, desired config.Pipeline) (provisioning.Diff, error) + ApplyFn func(ctx context.Context, desired config.Pipeline, hash string, allowRestartOnRunning bool) (provisioning.Diff, error) + + planCalls []config.Pipeline + applyCalls []applyCall +} + +type applyCall struct { + desired config.Pipeline + hash string + allowRestartOnRunning bool +} + +func (f *fakeProvisioner) Plan(ctx context.Context, desired config.Pipeline) (provisioning.Diff, error) { + f.mu.Lock() + f.planCalls = append(f.planCalls, desired) + f.mu.Unlock() + if f.PlanFn != nil { + return f.PlanFn(ctx, desired) + } + return provisioning.Diff{PipelineID: desired.ID}, nil +} + +func (f *fakeProvisioner) ApplyPlanLive(ctx context.Context, desired config.Pipeline, hash string, allowRestartOnRunning bool) (provisioning.Diff, error) { + f.mu.Lock() + f.applyCalls = append(f.applyCalls, applyCall{desired: desired, hash: hash, allowRestartOnRunning: allowRestartOnRunning}) + f.mu.Unlock() + if f.ApplyFn != nil { + return f.ApplyFn(ctx, desired, hash, allowRestartOnRunning) + } + return provisioning.Diff{PipelineID: desired.ID, Hash: hash}, nil +} + +func (f *fakeProvisioner) applyCallCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.applyCalls) +} + +// fakeLifecycle is a test double for LifecycleStarter. +type fakeLifecycle struct { + mu sync.Mutex + + StartFn func(ctx context.Context, pipelineID string) error + + startCalls []string +} + +func (f *fakeLifecycle) Start(ctx context.Context, pipelineID string) error { + f.mu.Lock() + f.startCalls = append(f.startCalls, pipelineID) + f.mu.Unlock() + if f.StartFn != nil { + return f.StartFn(ctx, pipelineID) + } + return nil +} + +func (f *fakeLifecycle) startCallCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.startCalls) +} + +// fakeClock is a deterministic, manually-driven Clock for testing the +// debounce engine (debounce.go) without a real sleep. Every After call is +// recorded (in order) and reported on calls, so a test can synchronize +// ("wait until the debouncer actually asked for a timer") before firing one. +type fakeClock struct { + mu sync.Mutex + pending []chan time.Time + calls chan time.Duration +} + +func newFakeClock() *fakeClock { + return &fakeClock{calls: make(chan time.Duration, 64)} +} + +func (c *fakeClock) After(d time.Duration) <-chan time.Time { + ch := make(chan time.Time, 1) + c.mu.Lock() + c.pending = append(c.pending, ch) + c.mu.Unlock() + c.calls <- d + return ch +} + +// awaitCall blocks until the next After call is recorded, or panics after 5s +// (a debouncer that never calls After is a bug the test must surface loudly, +// not hang forever). +func (c *fakeClock) awaitCall() time.Duration { + select { + case d := <-c.calls: + return d + case <-time.After(5 * time.Second): + panic("fakeClock: timed out waiting for After to be called") + } +} + +// fireLatest fires the most recently created pending timer — the one a +// debouncer's single timerC variable currently references, since every new +// trigger replaces that reference with a fresh After call (see debounce.go). +// Any earlier, now-unreferenced timer left in pending is never observed by +// anything and firing it would be a no-op from the debouncer's perspective; +// fireLatest always targets the live one. +func (c *fakeClock) fireLatest() { + c.mu.Lock() + defer c.mu.Unlock() + if len(c.pending) == 0 { + return + } + ch := c.pending[len(c.pending)-1] + c.pending = c.pending[:len(c.pending)-1] + ch <- time.Now() +} diff --git a/pkg/conduit/dev/interfaces.go b/pkg/conduit/dev/interfaces.go new file mode 100644 index 000000000..750ef53f2 --- /dev/null +++ b/pkg/conduit/dev/interfaces.go @@ -0,0 +1,71 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev + +import ( + "context" + "time" + + "github.com/conduitio/conduit/pkg/provisioning" + "github.com/conduitio/conduit/pkg/provisioning/config" +) + +// Provisioner is the subset of *pkg/provisioning.Service the Watcher needs. +// *provisioning.Service satisfies it directly; declaring it here (rather +// than depending on the concrete type) lets the apply flow be unit-tested +// against a fake, without a database or plugin runtime — the same pattern +// pkg/http/api.Provisioner and cmd/conduit/internal/deploy.PlanApplier use +// for the same method pair. +type Provisioner interface { + // Plan computes the Diff needed to reconcile the currently stored + // pipeline with desired, without applying anything. + Plan(ctx context.Context, desired config.Pipeline) (provisioning.Diff, error) + // ApplyPlanLive applies a previously-planned diff. allowRestartOnRunning + // is the Tier-1 operator-authorization gate (see plan.go's doc on + // CodeLiveApplyUnauthorized); the Watcher always passes true — see this + // package's doc on why that is safe here specifically. + ApplyPlanLive(ctx context.Context, desired config.Pipeline, hash string, allowRestartOnRunning bool) (provisioning.Diff, error) +} + +// LifecycleStarter is the subset of the lifecycle service the Watcher needs +// for ensure-running (see this package's doc). *pkg/lifecycle.Service and +// *pkg/lifecycle-poc.Service both satisfy it. +type LifecycleStarter interface { + Start(ctx context.Context, pipelineID string) error +} + +// StatusFunc reports whether pipelineID currently has live, in-process work +// — used only to label an apply accurately (see Mode's doc): a diff that +// is not live-eligible only means "this apply restarts the pipeline" if a +// pipeline was actually running to restart. A brand-new or previously +// -stopped pipeline is merely provisioned (and possibly started by +// ensure-running), never disrupted, and should not be reported as a +// restart. A nil StatusFunc is treated as "assume not running", which is +// always a safe (if less informative) label to fall back to — it never +// changes what the Watcher actually does, only how it is reported. +type StatusFunc func(ctx context.Context, pipelineID string) (running bool, err error) + +// Clock abstracts time.After so the debounce/coalesce engine (debounce.go) +// can be driven deterministically in tests, without a real sleep. Production +// code uses realClock; tests inject a fake that controls exactly when a +// debounce window elapses. +type Clock interface { + After(d time.Duration) <-chan time.Time +} + +// realClock is the production Clock, backed by the standard library. +type realClock struct{} + +func (realClock) After(d time.Duration) <-chan time.Time { return time.After(d) } diff --git a/pkg/conduit/dev/watcher.go b/pkg/conduit/dev/watcher.go new file mode 100644 index 000000000..6c9376f3b --- /dev/null +++ b/pkg/conduit/dev/watcher.go @@ -0,0 +1,265 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev + +import ( + "context" + "io" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/log" + "github.com/fsnotify/fsnotify" +) + +// Options configures a Watcher. +type Options struct { + // Path is the pipelines path to watch — a single .yml/.yaml file or a + // directory of them (conduit.Config.Pipelines.Path; --pipelines.path). + // Required. + Path string + + // Debounce is the coalescing window (see debounce.go). Defaults to + // DefaultDebounce (300ms) when zero. + Debounce time.Duration + + // Clock abstracts time for the debounce engine. Defaults to the real + // clock; tests inject a fake. + Clock Clock + + // Logger receives diagnostic (debug/warn/error) logging. Required: the + // zero value embeds a zero-value zerolog.Logger with a nil writer, which + // panics on first use — pass log.Nop() explicitly to disable logging. + Logger log.CtxLogger + + // Out is where Event output is written. Defaults to os.Stdout. + Out io.Writer + + // JSON selects --json structured event lines over human-readable text. + JSON bool +} + +// Watcher watches Options.Path for pipeline config changes and drives +// Plan/ApplyPlanLive against them — see this package's doc for the full +// design. +type Watcher struct { + provisioner Provisioner + lifecycle LifecycleStarter + statusFn StatusFunc + + watchDir string + match func(name string) bool + + debounce time.Duration + clock Clock + logger log.CtxLogger + reporter *Reporter + + mu sync.Mutex + filePipelines map[string][]string // path -> pipeline IDs last successfully parsed from it +} + +// New constructs a Watcher. provisioner and lifecycle are required; +// statusFn may be nil (see StatusFunc's doc for the effect). It resolves +// opts.Path immediately (stat-ing it, and its parent directory if it does +// not exist yet) so a bad --pipelines.path fails fast at startup rather than +// silently watching nothing. +func New(provisioner Provisioner, lifecycle LifecycleStarter, statusFn StatusFunc, opts Options) (*Watcher, error) { + if provisioner == nil { + return nil, cerrors.New("dev: a Provisioner is required") + } + if lifecycle == nil { + return nil, cerrors.New("dev: a LifecycleStarter is required") + } + + dir, match, err := resolveWatchTarget(opts.Path) + if err != nil { + return nil, err + } + + debounce := opts.Debounce + if debounce <= 0 { + debounce = DefaultDebounce + } + clock := opts.Clock + if clock == nil { + clock = realClock{} + } + out := opts.Out + if out == nil { + out = os.Stdout + } + + return &Watcher{ + provisioner: provisioner, + lifecycle: lifecycle, + statusFn: statusFn, + watchDir: dir, + match: match, + debounce: debounce, + clock: clock, + logger: opts.Logger.WithComponent("conduit.dev.Watcher"), + reporter: newReporter(out, opts.JSON), + filePipelines: map[string][]string{}, + }, nil +} + +// Run starts watching and blocks until ctx is cancelled, at which point it +// returns ctx.Err() (after every in-flight apply and per-path debouncer +// goroutine this call started has exited — see consume's doc). Invariant 7: +// callers (pkg/conduit.Runtime) derive ctx from the same serve context +// `conduit run`'s Ctrl-C/SIGTERM handling cancels. +func (w *Watcher) Run(ctx context.Context) error { + fsw, err := fsnotify.NewWatcher() + if err != nil { + return cerrors.Errorf("dev: could not start file watcher: %w", err) + } + defer fsw.Close() // best-effort cleanup on the way out + + if err := fsw.Add(w.watchDir); err != nil { + return cerrors.Errorf("dev: could not watch %q: %w", w.watchDir, err) + } + + w.logger.Info(ctx). + Str("dir", w.watchDir). + Dur("debounce", w.debounce). + Msg("dev: watching for pipeline config changes") + + return w.consume(ctx, fsw.Events, fsw.Errors) +} + +// consume is Run's core loop, split out so it can be unit-tested with +// synthetic events instead of a real fsnotify.Watcher (see watcher_test.go). +// It creates one debouncer per distinct file path the first time a relevant +// event names it, and waits for every debouncer goroutine it started to +// exit before returning — so Run never returns while an apply could still +// be in flight, and no goroutine leaks past a cancelled ctx. +func (w *Watcher) consume(ctx context.Context, events <-chan fsnotify.Event, fsErrors <-chan error) error { + // Each debouncer goroutine (d.run) only exits when its context is cancelled. + // Derive a child context and cancel it before wg.Wait so the debouncers are + // always torn down no matter WHY consume returns — a parent-ctx cancel OR the + // events/fsErrors channels closing (e.g. fsnotify.Close). Without this, a + // channel-close return would leave d.run goroutines blocked forever and + // wg.Wait would hang, stalling shutdown. Defer order matters: cancel() is + // declared last so it runs first (LIFO), unblocking d.run before wg.Wait. + ctx, cancel := context.WithCancel(ctx) + debouncers := map[string]*debouncer{} + var wg sync.WaitGroup + defer wg.Wait() + defer cancel() + + for { + select { + case <-ctx.Done(): + return ctx.Err() + + case ev, ok := <-events: + if !ok { + return nil + } + if !w.relevant(ev) { + continue + } + d, exists := debouncers[ev.Name] + if !exists { + path := ev.Name + d = newDebouncer(w.clock, w.debounce, func(applyCtx context.Context) { + w.applyFile(applyCtx, path) + }) + debouncers[ev.Name] = d + wg.Add(1) + go func() { + defer wg.Done() + d.run(ctx) + }() + } + d.trigger() + + case err, ok := <-fsErrors: + if !ok { + // The error channel is closed (the watcher is going away); stop + // consuming rather than busy-spinning on a permanently-ready recv. + return nil + } + w.logger.Err(ctx, err).Msg("dev: file watcher error") + } + } +} + +// relevant reports whether ev is an fs event the Watcher should react to: +// one of the ops that can change a file's content or existence (Chmod-only +// events are noise — a permission change never changes what a pipeline +// config says), for a name this Watcher was configured to care about. +func (w *Watcher) relevant(ev fsnotify.Event) bool { + const contentOps = fsnotify.Write | fsnotify.Create | fsnotify.Remove | fsnotify.Rename + if ev.Op&contentOps == 0 { + return false + } + return w.match(filepath.Base(ev.Name)) +} + +// resolveWatchTarget turns a conduit.Config.Pipelines.Path value (a single +// file or a directory) into the directory to hand fsnotify and a matcher +// for which file names within it are relevant. +// +// A single file is watched via its *parent directory* (not the file +// directly): watching an individual path is fragile across atomic-save +// editors, which replace the inode at that path via rename(2) — some +// platforms' fsnotify backends stop reporting events for a watch on the old, +// now-unlinked inode. Watching the directory and filtering by name (as the +// design doc's §4 "Debounce/coalesce" specifies) sidesteps this entirely and +// is what makes atomic-save tolerance possible in the first place. +func resolveWatchTarget(path string) (dir string, match func(name string) bool, err error) { + if path == "" { + return "", nil, cerrors.New("dev: pipelines path cannot be empty") + } + + info, statErr := os.Stat(path) + switch { + case statErr == nil && info.IsDir(): + return path, hasYAMLExt, nil + case statErr == nil: + base := filepath.Base(path) + return filepath.Dir(path), func(name string) bool { return name == base }, nil + case os.IsNotExist(statErr): + // path does not exist yet — most likely --pipelines.path naming a + // not-yet-created file. Watch its parent directory (which must + // exist) and match on this exact basename, so dev picks up the file + // the moment it is created ("an empty dir is valid — dev still + // watches so the first new file works", design doc §4). + parent := filepath.Dir(path) + if _, derr := os.Stat(parent); derr != nil { + return "", nil, cerrors.Errorf("dev: pipelines path %q does not exist, and its parent directory %q is not accessible: %w", path, parent, derr) + } + base := filepath.Base(path) + return parent, func(name string) bool { return name == base }, nil + default: + return "", nil, cerrors.Errorf("dev: could not stat pipelines path %q: %w", path, statErr) + } +} + +// hasYAMLExt reports whether name (a bare file name, not a full path) ends +// in .yml or .yaml, case-insensitively — the same rule +// pkg/provisioning/config.IsYAMLFile uses, minus the os.Stat call that +// function makes (which would always fail for a just-deleted file; this +// package needs to match a Remove event's name too). +func hasYAMLExt(name string) bool { + ext := strings.ToLower(filepath.Ext(name)) + return ext == ".yml" || ext == ".yaml" +} diff --git a/pkg/conduit/dev/watcher_test.go b/pkg/conduit/dev/watcher_test.go new file mode 100644 index 000000000..c530d3e4a --- /dev/null +++ b/pkg/conduit/dev/watcher_test.go @@ -0,0 +1,262 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/conduitio/conduit/pkg/provisioning" + "github.com/conduitio/conduit/pkg/provisioning/config" + "github.com/fsnotify/fsnotify" + "github.com/matryer/is" +) + +func TestResolveWatchTarget_Directory(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + + watchDir, match, err := resolveWatchTarget(dir) + is.NoErr(err) + is.Equal(watchDir, dir) + is.True(match("a.yaml")) + is.True(match("a.YML")) + is.True(!match("a.txt")) + is.True(!match("a.yaml.bak")) +} + +func TestResolveWatchTarget_SingleExistingFile(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := writeFile(t, dir, "pipelines.yaml", validPipelineYAML) + + watchDir, match, err := resolveWatchTarget(path) + is.NoErr(err) + is.Equal(watchDir, dir) + is.True(match("pipelines.yaml")) + is.True(!match("other.yaml")) // single-file mode only matches that one name +} + +func TestResolveWatchTarget_NotYetCreatedFile(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := filepath.Join(dir, "brand-new.yaml") + + watchDir, match, err := resolveWatchTarget(path) + is.NoErr(err) + is.Equal(watchDir, dir) + is.True(match("brand-new.yaml")) +} + +func TestResolveWatchTarget_NeitherPathNorParentExists(t *testing.T) { + is := is.New(t) + _, _, err := resolveWatchTarget(filepath.Join(t.TempDir(), "missing-dir", "p.yaml")) + is.True(err != nil) +} + +func TestResolveWatchTarget_EmptyPath(t *testing.T) { + is := is.New(t) + _, _, err := resolveWatchTarget("") + is.True(err != nil) +} + +func TestWatcher_Relevant(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + w, err := New(&fakeProvisioner{}, &fakeLifecycle{}, nil, Options{Path: dir}) + is.NoErr(err) + + is.True(w.relevant(fsnotify.Event{Name: filepath.Join(dir, "a.yaml"), Op: fsnotify.Write})) + is.True(w.relevant(fsnotify.Event{Name: filepath.Join(dir, "a.yaml"), Op: fsnotify.Create})) + is.True(w.relevant(fsnotify.Event{Name: filepath.Join(dir, "a.yaml"), Op: fsnotify.Remove})) + is.True(w.relevant(fsnotify.Event{Name: filepath.Join(dir, "a.yaml"), Op: fsnotify.Rename})) + is.True(!w.relevant(fsnotify.Event{Name: filepath.Join(dir, "a.yaml"), Op: fsnotify.Chmod})) + is.True(!w.relevant(fsnotify.Event{Name: filepath.Join(dir, "a.txt"), Op: fsnotify.Write})) +} + +// TestConsume_EventToFileToPipeline drives the Watcher's core loop +// (consume) with synthetic fsnotify events over a real temp directory, +// proving the event->file->pipeline mapping end to end without needing a +// live fsnotify.Watcher: a Write event for a valid pipeline file results in +// exactly one ApplyPlanLive call for the pipeline ID that file defines. +func TestConsume_EventToFileToPipeline(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := writeFile(t, dir, "orders.yaml", validPipelineYAML) + + applied := make(chan config.Pipeline, 1) + prov := &fakeProvisioner{ + PlanFn: func(_ context.Context, desired config.Pipeline) (provisioning.Diff, error) { + return nonEmptyDiff(desired.ID), nil + }, + ApplyFn: func(_ context.Context, desired config.Pipeline, hash string, _ bool) (provisioning.Diff, error) { + applied <- desired + return provisioning.Diff{PipelineID: desired.ID, Hash: hash}, nil + }, + } + lc := &fakeLifecycle{} + w, err := New(prov, lc, nil, Options{ + Path: dir, + Debounce: 10 * time.Millisecond, + }) + is.NoErr(err) + w.reporter = newReporter(&nopWriter{}, false) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + events := make(chan fsnotify.Event, 1) + errs := make(chan error) + done := make(chan error, 1) + go func() { done <- w.consume(ctx, events, errs) }() + + events <- fsnotify.Event{Name: path, Op: fsnotify.Write} + + select { + case desired := <-applied: + is.Equal(desired.ID, "orders") + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the pipeline to be applied") + } + + cancel() + select { + case err := <-done: + is.True(err != nil) // consume returns ctx.Err() (context.Canceled) on shutdown + case <-time.After(5 * time.Second): + t.Fatal("consume did not return after context cancellation") + } +} + +// TestConsume_ChannelClose_TearsDownAndReturns proves consume returns (and does +// not hang) when the fsnotify events channel closes while its context is still +// alive. The per-path debouncer goroutines only exit on their context, so +// consume must cancel a derived child context on the way out; without that, +// wg.Wait would block forever on a still-running d.run and Watcher.Run would +// never return, stalling shutdown. +func TestConsume_ChannelClose_TearsDownAndReturns(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := writeFile(t, dir, "orders.yaml", validPipelineYAML) + + prov := &fakeProvisioner{ + PlanFn: func(_ context.Context, desired config.Pipeline) (provisioning.Diff, error) { + return nonEmptyDiff(desired.ID), nil + }, + } + w, err := New(prov, &fakeLifecycle{}, nil, Options{Path: dir, Debounce: 10 * time.Millisecond}) + is.NoErr(err) + w.reporter = newReporter(&nopWriter{}, false) + + ctx := context.Background() // never cancelled: teardown must come from the channel close, not ctx + events := make(chan fsnotify.Event, 1) + errs := make(chan error) + done := make(chan error, 1) + go func() { done <- w.consume(ctx, events, errs) }() + + // Spawn a debouncer whose run goroutine only exits on the (child) context. + events <- fsnotify.Event{Name: path, Op: fsnotify.Write} + time.Sleep(50 * time.Millisecond) + + // Closing events must tear the debouncer down and let consume return, even + // though the parent ctx is still alive. + close(events) + select { + case err := <-done: + is.NoErr(err) // a channel-close return is nil, not ctx.Err() + case <-time.After(5 * time.Second): + t.Fatal("consume did not return after events channel close — debouncer teardown hung") + } +} + +// TestConsume_IrrelevantEvent_NeverApplies proves a non-matching file (wrong +// extension) never triggers a debouncer/apply at all. +func TestConsume_IrrelevantEvent_NeverApplies(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + path := writeFile(t, dir, "notes.txt", "hello") + + prov := &fakeProvisioner{} + w, err := New(prov, &fakeLifecycle{}, nil, Options{Path: dir, Debounce: 5 * time.Millisecond}) + is.NoErr(err) + w.reporter = newReporter(&nopWriter{}, false) + + ctx, cancel := context.WithCancel(context.Background()) + events := make(chan fsnotify.Event, 1) + errs := make(chan error) + go func() { _ = w.consume(ctx, events, errs) }() + + events <- fsnotify.Event{Name: path, Op: fsnotify.Write} + time.Sleep(50 * time.Millisecond) + cancel() + + is.Equal(prov.applyCallCount(), 0) +} + +// TestWatcher_Run_RealFsnotify exercises the real fsnotify.Watcher wiring +// (Run, not just consume) over a real temp directory: writing a valid +// pipeline file results in a real apply. +func TestWatcher_Run_RealFsnotify(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + + applied := make(chan config.Pipeline, 1) + prov := &fakeProvisioner{ + PlanFn: func(_ context.Context, desired config.Pipeline) (provisioning.Diff, error) { + return nonEmptyDiff(desired.ID), nil + }, + ApplyFn: func(_ context.Context, desired config.Pipeline, hash string, _ bool) (provisioning.Diff, error) { + applied <- desired + return provisioning.Diff{PipelineID: desired.ID, Hash: hash}, nil + }, + } + w, err := New(prov, &fakeLifecycle{}, nil, Options{ + Path: dir, + Debounce: 20 * time.Millisecond, + Out: &nopWriter{}, + }) + is.NoErr(err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + runDone := make(chan error, 1) + go func() { runDone <- w.Run(ctx) }() + + // Give the watcher a moment to register with the OS before writing — + // otherwise the write could race the Add() call. + time.Sleep(50 * time.Millisecond) + is.NoErr(os.WriteFile(filepath.Join(dir, "orders.yaml"), []byte(validPipelineYAML), 0o600)) + + select { + case desired := <-applied: + is.Equal(desired.ID, "orders") + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for a real fsnotify event to drive an apply") + } + + cancel() + select { + case <-runDone: + case <-time.After(5 * time.Second): + t.Fatal("Run did not return after context cancellation") + } +} + +type nopWriter struct{} + +func (*nopWriter) Write(p []byte) (int, error) { return len(p), nil } diff --git a/pkg/conduit/dev_integration_test.go b/pkg/conduit/dev_integration_test.go new file mode 100644 index 000000000..054cf968e --- /dev/null +++ b/pkg/conduit/dev_integration_test.go @@ -0,0 +1,252 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package conduit_test + +import ( + "bufio" + "context" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/conduitio/conduit/pkg/conduit" + "github.com/conduitio/conduit/pkg/foundation/log" + "github.com/matryer/is" + "github.com/rs/zerolog" +) + +// TestRunDev_HotReload_EndToEnd is the design doc's PR2 integration AC (see +// docs/design-documents/20260712-pipeline-dev-hot-reload.md §4 "Testing"): +// a real generator -> processor -> file pipeline, run under `--dev` (i.e. +// Config.Dev.Enabled, exactly what cmd/conduit/root/run.RunCommand sets), +// watched by a REAL fsnotify.Watcher over a real temp directory (no fakes +// anywhere in this test — pkg/conduit/dev's own package tests cover the +// fake-driven unit cases). +// +// It exercises, in order: +// 1. Editing the processor's config -> applied in place (no restart log, +// output reflects the new value). +// 2. Editing the source connector's setting -> applied via a labeled +// restart (a restart log appears). +// 3. Saving a syntax error -> the pipeline keeps running (no further +// restart, output keeps growing) and the error is logged. +// +// "No restart" is verified the way the design doc's AC literally states it: +// by the absence of a second "pipeline started" log line for this pipeline +// between edits — pkg/lifecycle.Service.Start (the only thing that emits +// that line) is called once at startup and, again, only by +// ApplyPlanLive's restart path. This is the same signal an operator would +// grep for in production, not a synthetic testing hook. +func TestRunDev_HotReload_EndToEnd(t *testing.T) { + if testing.Short() { + t.Skip("skipping full-engine dev hot-reload integration test in -short mode") + } + is := is.New(t) + + tmp := t.TempDir() + pipelinesDir := filepath.Join(tmp, "pipelines") + is.NoErr(os.MkdirAll(pipelinesDir, 0o755)) + pipelinePath := filepath.Join(pipelinesDir, "orders.yaml") + outputPath := filepath.Join(tmp, "output.ndjson") + + is.NoErr(os.WriteFile(pipelinePath, []byte(pipelineYAML(outputPath, "50", "before")), 0o600)) + + logs := &safeBuffer{} + cfg := conduit.DefaultConfig() + cfg.DB.Badger.Path = filepath.Join(tmp, "conduit.db") + cfg.API.Enabled = false // not needed for this test; keeps startup fast and simple + cfg.Pipelines.Path = pipelinesDir + cfg.Dev.Enabled = true + cfg.Log.Level = "debug" + cfg.Log.NewLogger = func(level, _ string) log.CtxLogger { + l, _ := zerolog.ParseLevel(level) + zl := zerolog.New(logs).With().Timestamp().Logger().Level(l) + return log.New(zl) + } + + r, err := conduit.NewRuntime(cfg) + is.NoErr(err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + runDone := make(chan error, 1) + go func() { runDone <- r.Run(ctx) }() + + select { + case <-r.Ready: + case err := <-runDone: + t.Fatalf("runtime exited before becoming ready: %v", err) + case <-time.After(20 * time.Second): + t.Fatal("runtime did not become ready in time") + } + + // The pipeline was provisioned+started by normal startup provisioning, + // not by the dev watcher (design doc: "startup provisioning runs + // normally; the watcher handles subsequent edits"). + is.NoErr(waitForLines(outputPath, 3, 10*time.Second)) + is.Equal(startedCount(logs), 1) + + // --- 1. processor-only edit: applies in place --- + linesBefore := countLines(t, outputPath) + is.NoErr(os.WriteFile(pipelinePath, []byte(pipelineYAML(outputPath, "50", "after")), 0o600)) + + // The AC: a processor-only edit applies IN PLACE — the new tag shows up + // in the output with no pipeline restart. waitForTag only succeeds once + // records carrying the new transform reach the destination, which can + // only happen if the live swap actually took (ReconfigureProcessor ran + // against the real processor.Service, not a mock). + if err := waitForTag(outputPath, "after", 15*time.Second); err != nil { + t.Fatalf("in-place apply did not land the new tag: %v\n\n=== LOGS ===\n%s", err, logs.String()) + } + is.Equal(startedCount(logs), 1) // no restart: still exactly the one startup Start + is.True(strings.Contains(logs.String(), `"mode":"in_place"`)) // engine reported an in-place apply + is.NoErr(waitForLines(outputPath, linesBefore+1, 10*time.Second)) // kept producing output + + // --- 2. source-setting edit: applies via a labeled restart --- + // Changing the generator's rate is a connector setting, not a processor + // setting, so it is restart-class: the engine tears the pipeline down and + // rebuilds it rather than swapping a node in place. + is.NoErr(os.WriteFile(pipelinePath, []byte(pipelineYAML(outputPath, "20", "after")), 0o600)) + + deadline := time.Now().Add(15 * time.Second) + for startedCount(logs) < 2 && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + is.Equal(startedCount(logs), 2) // restarted: a second "pipeline started" + is.True(strings.Contains(logs.String(), `"mode":"restart"`)) + + linesAfterRestart := countLines(t, outputPath) + is.NoErr(waitForLines(outputPath, linesAfterRestart+1, 10*time.Second)) // resumed producing output + is.NoErr(waitForTag(outputPath, "after", 10*time.Second)) // the tag change landed one way or another + + // --- 3. syntax error: pipeline keeps running, untouched --- + is.NoErr(os.WriteFile(pipelinePath, []byte("not: [valid: yaml:\n -"), 0o600)) + + // Give dev's debounce + a bad-apply attempt time to (not) happen. + time.Sleep(500 * time.Millisecond) + is.Equal(startedCount(logs), 2) // no further restart attempt + is.True(strings.Contains(logs.String(), "dev: ")) + + linesBeforeBadEdit := countLines(t, outputPath) + is.NoErr(waitForLines(outputPath, linesBeforeBadEdit+1, 10*time.Second)) // still running, still producing + + cancel() + select { + case <-runDone: + case <-time.After(20 * time.Second): + t.Fatal("runtime did not shut down after context cancellation") + } +} + +// pipelineYAML renders a generator -> field.set -> file pipeline writing to +// outputPath. rate controls the generator's records/second (a source +// connector setting — changing it is restart-class); tagValue controls the +// field.set processor's static value (a processor setting — changing it is +// live-swappable). +func pipelineYAML(outputPath, rate, tagValue string) string { + return fmt.Sprintf(`version: 2.2 +pipelines: + - id: dev-e2e + status: running + name: dev-e2e + connectors: + - id: src + type: source + plugin: builtin:generator + settings: + rate: %q + format.type: structured + format.options.id: int + operations: create + - id: dst + type: destination + plugin: builtin:file + settings: + path: %s + processors: + - id: tag + plugin: builtin:field.set + settings: + field: .Payload.After.tag + value: %q +`, rate, outputPath, tagValue) +} + +func startedCount(logs *safeBuffer) int { + return strings.Count(logs.String(), `"message":"pipeline started"`) +} + +func countLines(t *testing.T, path string) int { + t.Helper() + f, err := os.Open(path) + if os.IsNotExist(err) { + return 0 + } + if err != nil { + t.Fatalf("could not open %q: %v", path, err) + } + defer f.Close() + n := 0 + sc := bufio.NewScanner(f) + for sc.Scan() { + if strings.TrimSpace(sc.Text()) != "" { + n++ + } + } + return n +} + +func waitForLines(path string, minLines int, timeout time.Duration) error { + deadline := time.Now().Add(timeout) + for { + f, err := os.Open(path) + if err == nil { + n := 0 + sc := bufio.NewScanner(f) + for sc.Scan() { + if strings.TrimSpace(sc.Text()) != "" { + n++ + } + } + f.Close() + if n >= minLines { + return nil + } + } + if time.Now().After(deadline) { + return fmt.Errorf("timed out waiting for %q to have at least %d line(s)", path, minLines) + } + time.Sleep(20 * time.Millisecond) + } +} + +func waitForTag(path, tag string, timeout time.Duration) error { + deadline := time.Now().Add(timeout) + needle := `"tag":"` + tag + `"` + for { + data, err := os.ReadFile(path) + if err == nil && strings.Contains(string(data), needle) { + return nil + } + if time.Now().After(deadline) { + return fmt.Errorf("timed out waiting for %q to contain %q", path, needle) + } + time.Sleep(20 * time.Millisecond) + } +} diff --git a/pkg/conduit/runtime.go b/pkg/conduit/runtime.go index d04540dcd..1481d3a63 100644 --- a/pkg/conduit/runtime.go +++ b/pkg/conduit/runtime.go @@ -37,6 +37,7 @@ import ( pconnutils "github.com/conduitio/conduit-connector-protocol/pconnutils/v1/server" connutilsv1 "github.com/conduitio/conduit-connector-protocol/proto/connutils/v1" conduitschemaregistry "github.com/conduitio/conduit-schema-registry" + "github.com/conduitio/conduit/pkg/conduit/dev" "github.com/conduitio/conduit/pkg/connector" "github.com/conduitio/conduit/pkg/foundation/cerrors" "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" @@ -1027,5 +1028,70 @@ func (r *Runtime) initServices(ctx context.Context, t *tomb.Tomb) error { }) } + if r.Config.Dev.Enabled { + if err := r.startDevWatcher(ctx, t); err != nil { + return cerrors.Errorf("failed to start dev watcher: %w", err) + } + } + return nil } + +// startDevWatcher starts the `conduit run --dev` hot-reload file watcher +// (pkg/conduit/dev) as a tomb-managed goroutine, once startup provisioning +// (ProvisionService.Init above) and pipeline auto-resume +// (lifecycleService.Init above) have both already run — the watcher only +// ever reacts to *subsequent* edits, per +// docs/design-documents/20260712-pipeline-dev-hot-reload.md §4. +// +// Invariant 7: ctx here is the tomb-derived context Run constructed at its +// top (`t, ctx := tomb.WithContext(ctx)`), so Ctrl-C/SIGTERM cancelling it +// cancels the watcher the same way it cancels every other service — t.Go +// makes dev.Watcher.Run's return value part of the tomb's shutdown +// accounting, and a normal cancellation (context.Canceled) is translated to +// nil so it is never mistaken for a watcher failure. +func (r *Runtime) startDevWatcher(ctx context.Context, t *tomb.Tomb) error { + w, err := dev.New(r.ProvisionService, r.lifecycleService, r.devPipelineStatus, dev.Options{ + Path: r.Config.Pipelines.Path, + Logger: r.logger, + Out: os.Stdout, + JSON: r.Config.Dev.JSON, + }) + if err != nil { + return err + } + + t.Go(func() error { + err := w.Run(ctx) + if err != nil && cerrors.Is(err, context.Canceled) { + return nil + } + return err + }) + return nil +} + +// devPipelineStatus is the dev.StatusFunc the watcher uses purely to label +// an apply accurately (see dev.StatusFunc's doc) — it reports whether +// pipelineID currently has live, in-process work, mirroring +// provisioning.isRunningStatus's classification (pkg/provisioning/plan.go), +// which is unexported and cannot be called from here. This duplicates a +// three-case predicate, not engine behavior: keep it in sync with +// provisioning's own definition if that classification ever changes. +func (r *Runtime) devPipelineStatus(ctx context.Context, pipelineID string) (bool, error) { + inst, err := r.Orchestrator.Pipelines.Get(ctx, pipelineID) + if err != nil { + if cerrors.Is(err, pipeline.ErrInstanceNotFound) { + return false, nil + } + return false, err + } + switch inst.GetStatus() { + case pipeline.StatusRunning, pipeline.StatusRecovering, pipeline.StatusDegraded: + return true, nil + case pipeline.StatusSystemStopped, pipeline.StatusUserStopped: + return false, nil + default: + return false, nil + } +}