From 1e06fa7ca3d0b2e5efee300b2b5d139438eb3701 Mon Sep 17 00:00:00 2001 From: Devaris Date: Sun, 5 Jul 2026 14:35:50 -0700 Subject: [PATCH 1/2] feat(api): wire ConduitError into the API error boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 2 of the structured-error rollout: the ConduitError foundation now reaches the API. When an error returned to a gRPC/HTTP handler carries a *ConduitError (anywhere in its chain), status.go emits its gRPC category plus an additive google.rpc.ErrorInfo detail carrying the stable reason, configPath, suggestion, and fix — so an API/MCP/UI consumer gets a machine-actionable error. This is a single integration point (conduitErrorStatus, checked at the top of PipelineError/ConnectorError/ProcessorError/PluginError). It is purely additive: any error that is NOT a ConduitError falls through to the existing sentinel->code mapping, byte-for-byte unchanged (existing tests still pass). Boundaries get migrated to emit ConduitError codes incrementally from here. Tests: a wrapped ConduitError yields the right code + ErrorInfo detail (reason, configPath, suggestion); a sentinel error maps exactly as before. Design: docs/design-documents/20260705-conduit-error-and-structured-output.md Co-Authored-By: Claude Opus 4.8 (1M context) --- pkg/http/api/status/status.go | 29 +++++++++++++++++++++ pkg/http/api/status/status_test.go | 41 ++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/pkg/http/api/status/status.go b/pkg/http/api/status/status.go index 3cbde0377..354eb1ba4 100644 --- a/pkg/http/api/status/status.go +++ b/pkg/http/api/status/status.go @@ -17,6 +17,7 @@ package status import ( "github.com/conduitio/conduit/pkg/connector" "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" "github.com/conduitio/conduit/pkg/orchestrator" "github.com/conduitio/conduit/pkg/pipeline" conn_plugin "github.com/conduitio/conduit/pkg/plugin/connector" @@ -25,7 +26,24 @@ import ( grpcstatus "google.golang.org/grpc/status" ) +// conduitErrorStatus returns the structured gRPC status for a ConduitError in err's +// chain — its gRPC category plus a google.rpc.ErrorInfo detail carrying the stable +// reason and any configPath/suggestion/fix — and true if one was found. This is the +// single point where the ConduitError model reaches the API; boundaries that have +// not yet been migrated fall through to the sentinel-based code mapping below, +// unchanged. +func conduitErrorStatus(err error) (error, bool) { + if ce, ok := conduiterr.Get(err); ok { + return conduiterr.ToStatus(ce).Err(), true + } + return nil, false +} + func PipelineError(err error) error { + if s, ok := conduitErrorStatus(err); ok { + return s + } + var code codes.Code switch { @@ -41,6 +59,10 @@ func PipelineError(err error) error { } func ConnectorError(err error) error { + if s, ok := conduitErrorStatus(err); ok { + return s + } + var code codes.Code switch { @@ -56,6 +78,10 @@ func ConnectorError(err error) error { } func ProcessorError(err error) error { + if s, ok := conduitErrorStatus(err); ok { + return s + } + var code codes.Code switch { @@ -71,6 +97,9 @@ func ProcessorError(err error) error { } func PluginError(err error) error { + if s, ok := conduitErrorStatus(err); ok { + return s + } return grpcstatus.Error(codeFromError(err), err.Error()) } diff --git a/pkg/http/api/status/status_test.go b/pkg/http/api/status/status_test.go index 9e1ce0631..f8840cefe 100644 --- a/pkg/http/api/status/status_test.go +++ b/pkg/http/api/status/status_test.go @@ -19,10 +19,12 @@ import ( "github.com/conduitio/conduit/pkg/connector" "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" "github.com/conduitio/conduit/pkg/orchestrator" "github.com/conduitio/conduit/pkg/pipeline" "github.com/conduitio/conduit/pkg/processor" "github.com/matryer/is" + "google.golang.org/genproto/googleapis/rpc/errdetails" "google.golang.org/grpc/codes" grpcstatus "google.golang.org/grpc/status" ) @@ -126,6 +128,45 @@ func TestProcessorError(t *testing.T) { } } +// TestConduitErrorFlowsThroughBoundary proves the ConduitError foundation reaches +// the API: a ConduitError (even wrapped) yields a gRPC status with the code's +// category and an additive google.rpc.ErrorInfo detail carrying the stable reason, +// configPath, and suggestion. +func TestConduitErrorFlowsThroughBoundary(t *testing.T) { + is := is.New(t) + + ce := conduiterr.New(conduiterr.CodeConnectorPluginNotFound, "connector plugin not found") + ce.ConfigPath = "/connectors/0/plugin" + ce.Suggestion = "run `conduit connectors install `" + + // wrapped, to prove errors.As finds it through the chain + got := PipelineError(cerrors.Errorf("provisioning failed: %w", ce)) + + st, ok := grpcstatus.FromError(got) + is.True(ok) + is.Equal(st.Code(), codes.NotFound) // the code's gRPC category + + var info *errdetails.ErrorInfo + for _, d := range st.Details() { + if ei, ok := d.(*errdetails.ErrorInfo); ok { + info = ei + } + } + is.True(info != nil) // the structured detail is present + is.Equal(info.GetReason(), conduiterr.CodeConnectorPluginNotFound.Reason()) + is.Equal(info.GetDomain(), "conduit") + is.Equal(info.GetMetadata()["configPath"], "/connectors/0/plugin") + is.Equal(info.GetMetadata()["suggestion"], "run `conduit connectors install `") +} + +// TestNonConduitErrorUnchanged confirms the wiring is additive: a plain sentinel +// error still maps exactly as before, with no detail. +func TestNonConduitErrorUnchanged(t *testing.T) { + is := is.New(t) + got := PipelineError(pipeline.ErrInstanceNotFound) + is.Equal(grpcstatus.Error(codes.NotFound, pipeline.ErrInstanceNotFound.Error()), got) +} + func TestCodeFromError(t *testing.T) { type args struct { err error From 38ac8d0484847c0390ac0b2a6c1eb207b40995cd Mon Sep 17 00:00:00 2001 From: Devaris Date: Sun, 5 Jul 2026 14:56:19 -0700 Subject: [PATCH 2/2] feat(plugin): emit connector.plugin_not_found ConduitError code MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 3 of the structured-error rollout — the first real error source migrated to a ConduitError code, exercising the API boundary wired in the previous step. When the builtin connector registry can't find a plugin (unknown name, or a known name with no matching version), it now returns a ConduitError with code `connector.plugin_not_found` and a suggested fix, instead of a bare sentinel / generic wrapped error. The `plugin.ErrPluginNotFound` sentinel remains in the chain, so existing `errors.Is` checks are unaffected (invariant noted at the site; existing tests unchanged). End to end: registry -> ConduitError(code, suggestion) -> orchestrator -> API (status.ConnectorError/PluginError) -> ToStatus -> a google.rpc.ErrorInfo detail carrying reason=connector.plugin_not_found + the suggestion, on the wire. Test augments TestRegistry_NewDispenser_PluginNotFound to assert both the sentinel Is-check AND the new code + suggestion. Design: docs/design-documents/20260705-conduit-error-and-structured-output.md Co-Authored-By: Claude Opus 4.8 (1M context) --- pkg/plugin/connector/builtin/registry.go | 20 +++++++++++++++++-- pkg/plugin/connector/builtin/registry_test.go | 7 ++++++- 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/pkg/plugin/connector/builtin/registry.go b/pkg/plugin/connector/builtin/registry.go index ad75ad8fe..1f4312e81 100644 --- a/pkg/plugin/connector/builtin/registry.go +++ b/pkg/plugin/connector/builtin/registry.go @@ -16,6 +16,7 @@ package builtin import ( "context" + "fmt" "runtime/debug" file "github.com/conduitio/conduit-connector-file" @@ -28,6 +29,7 @@ import ( sdk "github.com/conduitio/conduit-connector-sdk" "github.com/conduitio/conduit-connector-sdk/schema" "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" "github.com/conduitio/conduit/pkg/foundation/log" "github.com/conduitio/conduit/pkg/plugin" "github.com/conduitio/conduit/pkg/plugin/connector" @@ -190,7 +192,15 @@ func newFullName(pluginName, pluginVersion string) plugin.FullName { func (r *Registry) NewDispenser(logger log.CtxLogger, fullName plugin.FullName, cfg pconnector.PluginConfig) (connector.Dispenser, error) { versionMap, ok := r.plugins[fullName.PluginName()] if !ok { - return nil, plugin.ErrPluginNotFound + // Invariant: errors.Is(err, plugin.ErrPluginNotFound) still holds — the + // sentinel is wrapped, and the ConduitError adds the machine-actionable code. + err := conduiterr.Wrap( + conduiterr.CodeConnectorPluginNotFound, + fmt.Sprintf("builtin connector plugin %q not found", fullName.PluginName()), + plugin.ErrPluginNotFound, + ) + err.Suggestion = "run `conduit connectors list` to see available built-in connectors" + return nil, err } b, ok := versionMap[fullName.PluginVersion()] if !ok { @@ -198,7 +208,13 @@ func (r *Registry) NewDispenser(logger log.CtxLogger, fullName plugin.FullName, for k := range versionMap { availableVersions = append(availableVersions, k) } - return nil, cerrors.Errorf("could not find builtin plugin %q, only found versions %v: %w", fullName, availableVersions, plugin.ErrPluginNotFound) + err := conduiterr.Wrap( + conduiterr.CodeConnectorPluginNotFound, + fmt.Sprintf("builtin connector plugin %q not found; available versions: %v", fullName, availableVersions), + plugin.ErrPluginNotFound, + ) + err.Suggestion = "check the plugin version, or omit it to use the latest" + return nil, err } return b.dispenserFactory(fullName, cfg, logger), nil diff --git a/pkg/plugin/connector/builtin/registry_test.go b/pkg/plugin/connector/builtin/registry_test.go index df6869cb4..43d27da40 100644 --- a/pkg/plugin/connector/builtin/registry_test.go +++ b/pkg/plugin/connector/builtin/registry_test.go @@ -20,6 +20,7 @@ import ( "github.com/conduitio/conduit-connector-protocol/pconnector" "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" "github.com/conduitio/conduit/pkg/foundation/log" "github.com/conduitio/conduit/pkg/plugin" "github.com/google/go-cmp/cmp" @@ -98,7 +99,11 @@ func TestRegistry_NewDispenser_PluginNotFound(t *testing.T) { dispenser, err := underTest.NewDispenser(log.Nop(), plugin.FullName(tc.pluginName), pconnector.PluginConfig{}) if tc.wantErr { - is.True(cerrors.Is(err, plugin.ErrPluginNotFound)) + is.True(cerrors.Is(err, plugin.ErrPluginNotFound)) // sentinel still in the chain + ce, ok := conduiterr.Get(err) + is.True(ok) // now also carries a machine-actionable ConduitError code + is.Equal(ce.Code.Reason(), conduiterr.CodeConnectorPluginNotFound.Reason()) + is.True(ce.Suggestion != "") // with a suggested fix is.True(dispenser == nil) return }