From bcf6c66fa4511889bad2df82c949b16bf421b1b2 Mon Sep 17 00:00:00 2001 From: Devaris Date: Sun, 5 Jul 2026 16:49:31 -0700 Subject: [PATCH] feat: emit pipeline/connector instance_not_found ConduitError codes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 4 of the structured-error rollout: two more error sources now emit ConduitError codes instead of bare sentinels, following the pattern from connector.plugin_not_found (#2528). - pipeline.ErrInstanceNotFound, raised in pipeline.Service.Get (the sole choke point — Update/Delete/AddConnector/etc. all call Get internally), now wraps with a new code pipeline.instance_not_found (codes.NotFound) and a suggestion to run `conduit pipelines list`. - connector.ErrInstanceNotFound, raised in connector.Service.Get (same choke-point property), wraps with connector.instance_not_found (codes.NotFound) and a suggestion to run `conduit connectors list`. Both sentinels remain in the error chain via conduiterr.Wrap, so existing errors.Is(err, ErrInstanceNotFound) checks throughout pkg/provisioning, pkg/orchestrator, and pkg/http/api/status are unaffected — verified by running those suites plus pkg/lifecycle(-poc). Tests augment TestService_GetInstanceNotFound in both packages to assert the sentinel Is-check AND the new code + non-empty Suggestion via conduiterr.Get. processor.ErrInstanceNotFound is a separate, un-migrated error source (pkg/processor/service.go) — left out of scope for this step to keep the blast radius tight; it's a natural next candidate. Design: docs/design-documents/20260705-conduit-error-and-structured-output.md Co-Authored-By: Claude Opus 4.8 (1M context) --- pkg/connector/codes.go | 29 +++++++++++++++++++++++++++++ pkg/connector/service.go | 12 +++++++++++- pkg/connector/service_test.go | 7 ++++++- pkg/pipeline/codes.go | 29 +++++++++++++++++++++++++++++ pkg/pipeline/service.go | 12 +++++++++++- pkg/pipeline/service_test.go | 7 ++++++- 6 files changed, 92 insertions(+), 4 deletions(-) create mode 100644 pkg/connector/codes.go create mode 100644 pkg/pipeline/codes.go diff --git a/pkg/connector/codes.go b/pkg/connector/codes.go new file mode 100644 index 000000000..ec642dda6 --- /dev/null +++ b/pkg/connector/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 connector + +import ( + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" + "google.golang.org/grpc/codes" +) + +// Connector 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 ( + // CodeConnectorNotFound is raised when a referenced connector instance + // cannot be located. + CodeConnectorNotFound = conduiterr.Register("connector.instance_not_found", codes.NotFound) +) diff --git a/pkg/connector/service.go b/pkg/connector/service.go index a5ddfcc8d..8f8ea546d 100644 --- a/pkg/connector/service.go +++ b/pkg/connector/service.go @@ -16,12 +16,14 @@ package connector import ( "context" + "fmt" "regexp" "strings" "time" "github.com/conduitio/conduit-commons/database" "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" ) @@ -104,7 +106,15 @@ func (s *Service) List(context.Context) map[string]*Instance { func (s *Service) Get(_ context.Context, id string) (*Instance, error) { ins, ok := s.connectors[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( + CodeConnectorNotFound, + fmt.Sprintf("connector %q not found", id), + ErrInstanceNotFound, + ) + err.Suggestion = "run `conduit connectors list` to see existing connectors" + return nil, err } return ins, nil } diff --git a/pkg/connector/service_test.go b/pkg/connector/service_test.go index 94b9a1794..ffd16b1a4 100644 --- a/pkg/connector/service_test.go +++ b/pkg/connector/service_test.go @@ -25,6 +25,7 @@ import ( "github.com/conduitio/conduit-commons/database/mock" "github.com/conduitio/conduit-commons/opencdc" "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" "github.com/conduitio/conduit/pkg/foundation/log" pmock "github.com/conduitio/conduit/pkg/plugin/connector/mock" "github.com/google/uuid" @@ -423,7 +424,11 @@ func TestService_GetInstanceNotFound(t *testing.T) { // get connector that does not exist got, err := service.Get(ctx, uuid.NewString()) is.True(err != nil) - is.True(cerrors.Is(err, ErrInstanceNotFound)) + 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(), CodeConnectorNotFound.Reason()) + is.True(ce.Suggestion != "") // with a suggested fix is.Equal(got, nil) } diff --git a/pkg/pipeline/codes.go b/pkg/pipeline/codes.go new file mode 100644 index 000000000..2614b4630 --- /dev/null +++ b/pkg/pipeline/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 pipeline + +import ( + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" + "google.golang.org/grpc/codes" +) + +// Pipeline 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 ( + // CodePipelineNotFound is raised when a referenced pipeline instance + // cannot be located. + CodePipelineNotFound = conduiterr.Register("pipeline.instance_not_found", codes.NotFound) +) diff --git a/pkg/pipeline/service.go b/pkg/pipeline/service.go index a58f57ab9..842e46fff 100644 --- a/pkg/pipeline/service.go +++ b/pkg/pipeline/service.go @@ -16,12 +16,14 @@ package pipeline import ( "context" + "fmt" "regexp" "strings" "time" "github.com/conduitio/conduit-commons/database" "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" ) @@ -104,7 +106,15 @@ func (s *Service) List(context.Context) map[string]*Instance { func (s *Service) Get(_ context.Context, id string) (*Instance, error) { p, 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( + CodePipelineNotFound, + fmt.Sprintf("pipeline %q not found", id), + ErrInstanceNotFound, + ) + err.Suggestion = "run `conduit pipelines list` to see existing pipelines" + return nil, err } return p, nil } diff --git a/pkg/pipeline/service_test.go b/pkg/pipeline/service_test.go index a5a80502a..5417b7da2 100644 --- a/pkg/pipeline/service_test.go +++ b/pkg/pipeline/service_test.go @@ -23,6 +23,7 @@ import ( "github.com/conduitio/conduit-commons/database/inmemory" "github.com/conduitio/conduit-commons/database/mock" "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" "github.com/conduitio/conduit/pkg/foundation/log" "github.com/google/uuid" "github.com/matryer/is" @@ -312,7 +313,11 @@ func TestService_GetInstanceNotFound(t *testing.T) { // get pipeline instance that does not exist got, err := service.Get(ctx, uuid.NewString()) is.True(err != nil) - is.True(cerrors.Is(err, ErrInstanceNotFound)) + 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(), CodePipelineNotFound.Reason()) + is.True(ce.Suggestion != "") // with a suggested fix is.Equal(got, nil) }