diff --git a/pkg/foundation/cerrors/conduiterr/conduiterr.go b/pkg/foundation/cerrors/conduiterr/conduiterr.go index 59e720bdc..3b1dd2a68 100644 --- a/pkg/foundation/cerrors/conduiterr/conduiterr.go +++ b/pkg/foundation/cerrors/conduiterr/conduiterr.go @@ -104,6 +104,9 @@ var ( // CodeConnectorPluginNotFound is raised when a referenced connector plugin // cannot be located. CodeConnectorPluginNotFound = Register("connector.plugin_not_found", codes.NotFound) + // CodeProcessorPluginNotFound is raised when a referenced processor plugin + // cannot be located. + CodeProcessorPluginNotFound = Register("processor.plugin_not_found", codes.NotFound) ) // Fix is a structured, machine-appliable change that resolves an error. The same diff --git a/pkg/plugin/connector/standalone/registry.go b/pkg/plugin/connector/standalone/registry.go index 3968a271d..4b93a6844 100644 --- a/pkg/plugin/connector/standalone/registry.go +++ b/pkg/plugin/connector/standalone/registry.go @@ -16,6 +16,7 @@ package standalone import ( "context" + "fmt" "os" "path" "path/filepath" @@ -26,6 +27,7 @@ import ( "github.com/conduitio/conduit-connector-protocol/pconnector/client" "github.com/conduitio/conduit-connector-protocol/pconnutils" "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" @@ -195,7 +197,15 @@ func (r *Registry) NewDispenser(logger log.CtxLogger, fullName plugin.FullName, 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("standalone connector plugin %q not found", fullName.PluginName()), + plugin.ErrPluginNotFound, + ) + err.Suggestion = "check the plugin name and version" + return nil, err } bp, ok := versionMap[fullName.PluginVersion()] if !ok { @@ -203,7 +213,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 standalone connector plugin, only found versions %v: %w", availableVersions, plugin.ErrPluginNotFound) + err := conduiterr.Wrap( + conduiterr.CodeConnectorPluginNotFound, + fmt.Sprintf("standalone connector plugin %q not found; available versions: %v", fullName.PluginName(), availableVersions), + plugin.ErrPluginNotFound, + ) + err.Suggestion = "check the plugin version, or omit it to use the latest" + return nil, err } logger = logger.WithComponent("plugin.standalone") diff --git a/pkg/plugin/processor/builtin/registry.go b/pkg/plugin/processor/builtin/registry.go index 422753989..216ea90ec 100644 --- a/pkg/plugin/processor/builtin/registry.go +++ b/pkg/plugin/processor/builtin/registry.go @@ -16,12 +16,14 @@ package builtin import ( "context" + "fmt" "runtime/debug" sdk "github.com/conduitio/conduit-processor-sdk" "github.com/conduitio/conduit-processor-sdk/pprocutils" "github.com/conduitio/conduit-processor-sdk/schema" "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" "github.com/conduitio/conduit/pkg/foundation/ctxutil" "github.com/conduitio/conduit/pkg/foundation/log" "github.com/conduitio/conduit/pkg/plugin" @@ -190,7 +192,15 @@ func newFullName(pluginName, pluginVersion string) plugin.FullName { func (r *Registry) NewProcessor(_ context.Context, fullName plugin.FullName, id string) (sdk.Processor, 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.CodeProcessorPluginNotFound, + fmt.Sprintf("builtin processor plugin %q not found", fullName.PluginName()), + plugin.ErrPluginNotFound, + ) + err.Suggestion = "check the plugin name and version" + return nil, err } b, ok := versionMap[fullName.PluginVersion()] if !ok { @@ -198,7 +208,13 @@ func (r *Registry) NewProcessor(_ context.Context, fullName plugin.FullName, id 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.CodeProcessorPluginNotFound, + fmt.Sprintf("builtin processor 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 } p := b.constructor(r.logger) diff --git a/pkg/plugin/processor/standalone/registry.go b/pkg/plugin/processor/standalone/registry.go index 46148f927..903d96358 100644 --- a/pkg/plugin/processor/standalone/registry.go +++ b/pkg/plugin/processor/standalone/registry.go @@ -25,6 +25,7 @@ import ( sdk "github.com/conduitio/conduit-processor-sdk" "github.com/conduitio/conduit-processor-sdk/pprocutils" "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/stealthrocket/wazergo" @@ -125,7 +126,15 @@ func (r *Registry) NewProcessor(ctx context.Context, fullName plugin.FullName, i versions, 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.CodeProcessorPluginNotFound, + fmt.Sprintf("standalone processor plugin %q not found", fullName.PluginName()), + plugin.ErrPluginNotFound, + ) + err.Suggestion = "check the plugin name and version" + return nil, err } bp, ok := versions[fullName.PluginVersion()] if !ok { @@ -133,7 +142,13 @@ func (r *Registry) NewProcessor(ctx context.Context, fullName plugin.FullName, i for k := range versions { availableVersions = append(availableVersions, k) } - return nil, cerrors.Errorf("could not find standalone processor plugin, only found versions %v: %w", availableVersions, plugin.ErrPluginNotFound) + err := conduiterr.Wrap( + conduiterr.CodeProcessorPluginNotFound, + fmt.Sprintf("standalone processor plugin %q not found; available versions: %v", fullName.PluginName(), availableVersions), + plugin.ErrPluginNotFound, + ) + err.Suggestion = "check the plugin version, or omit it to use the latest" + return nil, err } p, err := newWASMProcessor(ctx, r.runtime, bp.module, r.hostModule, r.schemaService, id, r.logger) diff --git a/pkg/processor/codes.go b/pkg/processor/codes.go new file mode 100644 index 000000000..ce4cc9ca7 --- /dev/null +++ b/pkg/processor/codes.go @@ -0,0 +1,29 @@ +// 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 processor + +import ( + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" + "google.golang.org/grpc/codes" +) + +// Processor error codes. Every error carries one of these codes plus a +// suggested fix, so an API, MCP, or UI consumer knows what happened without +// parsing message text. +var ( + // CodeProcessorNotFound is raised when a referenced processor instance + // cannot be located. + CodeProcessorNotFound = conduiterr.Register("processor.instance_not_found", codes.NotFound) +) diff --git a/pkg/processor/service.go b/pkg/processor/service.go index 5533f0abf..c0c48aa57 100644 --- a/pkg/processor/service.go +++ b/pkg/processor/service.go @@ -18,11 +18,13 @@ package processor import ( "context" + "fmt" "time" "github.com/conduitio/conduit-commons/database" sdk "github.com/conduitio/conduit-processor-sdk" "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/foundation/metrics/measure" ) @@ -86,7 +88,15 @@ func (s *Service) List(_ context.Context) map[string]*Instance { func (s *Service) Get(_ context.Context, id string) (*Instance, error) { ins, ok := s.instances[id] if !ok { - return nil, cerrors.Errorf("%w (ID: %s)", ErrInstanceNotFound, id) + // Invariant: errors.Is(err, ErrInstanceNotFound) still holds — the sentinel + // is wrapped, and the ConduitError adds the machine-actionable code. + err := conduiterr.Wrap( + CodeProcessorNotFound, + fmt.Sprintf("processor %q not found", id), + ErrInstanceNotFound, + ) + err.Suggestion = "check the processor ID in your pipeline configuration" + return nil, err } return ins, nil } diff --git a/pkg/processor/service_test.go b/pkg/processor/service_test.go index 1c803f048..7ff54224f 100644 --- a/pkg/processor/service_test.go +++ b/pkg/processor/service_test.go @@ -23,6 +23,7 @@ import ( dbmock "github.com/conduitio/conduit-commons/database/mock" sdk "github.com/conduitio/conduit-processor-sdk" "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" proc_plugin "github.com/conduitio/conduit/pkg/plugin/processor" @@ -328,7 +329,11 @@ func TestService_Get_Fail(t *testing.T) { service := NewService(log.Nop(), db, &proc_plugin.PluginService{}) got, err := service.Get(ctx, "non-existent processor") - is.True(cerrors.Is(err, ErrInstanceNotFound)) // expected instance not found error + is.True(cerrors.Is(err, ErrInstanceNotFound)) // 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(), CodeProcessorNotFound.Reason()) + is.True(ce.Suggestion != "") // with a suggested fix is.Equal(got, nil) }