Return structured gRPC errors on source table missing (#4194)

Helps to unify the handling across connectors and consume on the caller
without string matching.

Also clean up unused api error and move string constants to pkg. Would
be nicer to have everything in protos but string constants are not
supported there and not all our constants are uppercase to be enums.

Author: ilidemi

flow/cmd/api_error.go

@@ -2,14 +2,17 @@ package cmd
 
 import (
 	"errors"
+	"fmt"
 	"log/slog"
 
 	"github.com/gogo/googleapis/google/rpc"
+	"google.golang.org/genproto/googleapis/rpc/errdetails"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/protoadapt"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/grpc_handler"
+	"github.com/PeerDB-io/peerdb/flow/pkg/common"
 )
 
 // APIError is a strongly-typed error that must be a gRPC status error.
@@ -24,31 +27,31 @@ func newAPIError(s *status.Status) *apiError {
 	return &apiError{status: s}
 }
 
-func NewInvalidArgumentApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+func NewInvalidArgumentApiError(err error, details ...protoadapt.MessageV1) *apiError {
 	return newAPIError(convertToStatus(codes.InvalidArgument, err, details...))
 }
 
-func NewFailedPreconditionApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+func NewFailedPreconditionApiError(err error, details ...protoadapt.MessageV1) *apiError {
 	return newAPIError(convertToStatus(codes.FailedPrecondition, err, details...))
 }
 
-func NewInternalApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+func NewInternalApiError(err error, details ...protoadapt.MessageV1) *apiError {
 	return newAPIError(convertToStatus(codes.Internal, err, details...))
 }
 
-func NewUnavailableApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+func NewUnavailableApiError(err error, details ...protoadapt.MessageV1) *apiError {
 	return newAPIError(convertToStatus(codes.Unavailable, err, details...))
 }
 
-func NewUnimplementedApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+func NewUnimplementedApiError(err error, details ...protoadapt.MessageV1) *apiError {
 	return newAPIError(convertToStatus(codes.Unimplemented, err, details...))
 }
 
-func NewAlreadyExistsApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+func NewAlreadyExistsApiError(err error, details ...protoadapt.MessageV1) *apiError {
 	return newAPIError(convertToStatus(codes.AlreadyExists, err, details...))
 }
 
-func NewNotFoundApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+func NewNotFoundApiError(err error, details ...protoadapt.MessageV1) *apiError {
 	return newAPIError(convertToStatus(codes.NotFound, err, details...))
 }
 
@@ -70,18 +73,14 @@ func (e *apiError) Code() codes.Code {
 	return e.status.Code()
 }
 
-func convertToStatus(code codes.Code, err error, details ...*rpc.ErrorInfo) *status.Status {
+func convertToStatus(code codes.Code, err error, details ...protoadapt.MessageV1) *status.Status {
 	errorStatus := status.New(code, err.Error())
 	if len(details) == 0 {
 		return errorStatus
 	}
-	convertedDetails := make([]protoadapt.MessageV1, len(details))
-	for i, detail := range details {
-		convertedDetails[i] = detail
-	}
-	richStatus, err := errorStatus.WithDetails(convertedDetails...)
+	richStatus, err := errorStatus.WithDetails(details...)
 	if err != nil {
-		// This cannot happen because we control all calls to convertToStatus and only pass code != OK and allow only rpc.ErrorInfo in details
+		// This cannot happen because we control all calls to convertToStatus and only pass code != OK and valid proto details
 		slog.Error("Failed to convert to grpc proto error", slog.Any("error", err)) //nolint:sloglint // No context in conversion helper
 		return errorStatus
 	}
@@ -107,34 +106,57 @@ func AsAPIError(err error) APIError {
 	return NewInternalApiError(err)
 }
 
-const (
-	ErrorInfoReasonClickHousePeer = "CLICKHOUSE_PEER"
-	ErrorInfoReasonMirror         = "MIRROR"
-)
+func NewMirrorErrorInfo(metadata map[string]string) *rpc.ErrorInfo {
+	return &rpc.ErrorInfo{
+		Reason:   common.ErrorInfoReasonMirror,
+		Domain:   common.ErrorInfoDomain,
+		Metadata: metadata,
+	}
+}
 
-const (
-	ErrorInfoDomain = "peerdb.io"
-)
+func NewSourceTableMissingErrorInfo() *rpc.ErrorInfo {
+	return &rpc.ErrorInfo{
+		Reason: common.ErrorInfoReasonSourceTableMissing,
+		Domain: common.ErrorInfoDomain,
+	}
+}
 
-const (
-	ErrorMetadataDownstreamErrorCode = "downstreamErrorCode"
-	ErrorMetadataOffendingField      = "offendingField"
-)
+// NewSourceTableMissingPreconditionFailure builds the per-table breakdown detail
+// that accompanies the SOURCE_TABLE_MISSING ErrorInfo on FailedPrecondition.
+func NewSourceTableMissingPreconditionFailure(tables []common.QualifiedTable) protoadapt.MessageV1 {
+	violations := make([]*errdetails.PreconditionFailure_Violation, len(tables))
+	for i, t := range tables {
+		subject := fmt.Sprintf("%s.%s", t.Namespace, t.Table)
+		violations[i] = &errdetails.PreconditionFailure_Violation{
+			Type:        common.ErrorInfoReasonSourceTableMissing,
+			Subject:     subject,
+			Description: fmt.Sprintf("source table %s does not exist", subject),
+		}
+	}
+	return protoadapt.MessageV1Of(&errdetails.PreconditionFailure{Violations: violations})
+}
 
-func NewClickHousePeerErrorInfo(metadata map[string]string) *rpc.ErrorInfo {
+func NewTablesNotInPublicationErrorInfo(publication string) *rpc.ErrorInfo {
 	return &rpc.ErrorInfo{
-		Reason:   ErrorInfoReasonClickHousePeer,
-		Domain:   ErrorInfoDomain,
-		Metadata: metadata,
+		Reason:   common.ErrorInfoReasonTablesNotInPublication,
+		Domain:   common.ErrorInfoDomain,
+		Metadata: map[string]string{common.ErrorMetadataPublication: publication},
 	}
 }
 
-func NewMirrorErrorInfo(metadata map[string]string) *rpc.ErrorInfo {
-	return &rpc.ErrorInfo{
-		Reason:   ErrorInfoReasonMirror,
-		Domain:   ErrorInfoDomain,
-		Metadata: metadata,
+// NewTablesNotInPublicationPreconditionFailure builds the per-table breakdown detail
+// that accompanies the TABLES_NOT_IN_PUBLICATION ErrorInfo on FailedPrecondition.
+func NewTablesNotInPublicationPreconditionFailure(publication string, tables []common.QualifiedTable) protoadapt.MessageV1 {
+	violations := make([]*errdetails.PreconditionFailure_Violation, len(tables))
+	for i, t := range tables {
+		subject := fmt.Sprintf("%s.%s", t.Namespace, t.Table)
+		violations[i] = &errdetails.PreconditionFailure_Violation{
+			Type:        common.ErrorInfoReasonTablesNotInPublication,
+			Subject:     subject,
+			Description: fmt.Sprintf("table %s is not in publication %q", subject, publication),
+		}
 	}
+	return protoadapt.MessageV1Of(&errdetails.PreconditionFailure{Violations: violations})
 }
 
 var ErrUnderMaintenance = errors.New("PeerDB is under maintenance. Please retry in a few minutes")

flow/cmd/handler.go

@@ -509,7 +509,7 @@ func (h *FlowRequestHandler) FlowStateChange(
 				if _, err := h.ValidateCDCMirror(ctx, &protos.CreateCDCFlowRequest{
 					ConnectionConfigs: config,
 				}); err != nil {
-					return nil, NewFailedPreconditionApiError(fmt.Errorf("invalid mirror: %w", err))
+					return nil, err
 				}
 				changeErr = model.FlowSignalStateChange.SignalClientWorkflow(ctx, h.temporalClient, workflowID, "", req)
 				if changeErr == nil {
@@ -540,6 +540,10 @@ func (h *FlowRequestHandler) FlowStateChange(
 		}
 		if changeErr != nil {
 			slog.ErrorContext(ctx, "unable to signal workflow", logs, slog.Any("error", changeErr))
+			// Preserve typed API errors
+			if apiErr, ok := changeErr.(APIError); ok {
+				return nil, apiErr
+			}
 			return nil, NewInternalApiError(fmt.Errorf("unable to signal workflow: %w", changeErr))
 		}
 	}

flow/cmd/validate_mirror.go

@@ -12,6 +12,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/generated/proto_conversions"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
+	"github.com/PeerDB-io/peerdb/flow/pkg/common"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
@@ -65,7 +66,7 @@ func (h *FlowRequestHandler) validateCDCMirrorImpl(
 			return nil, NewAlreadyExistsApiError(
 				fmt.Errorf("mirror with name %s already exists", connectionConfigs.FlowJobName),
 				NewMirrorErrorInfo(map[string]string{
-					ErrorMetadataOffendingField: "flow_job_name",
+					common.ErrorMetadataOffendingField: "flow_job_name",
 				}))
 		}
 	}
@@ -106,6 +107,18 @@ func (h *FlowRequestHandler) validateCDCMirrorImpl(
 	defer srcClose(ctx)
 
 	if err := srcConn.ValidateMirrorSource(ctx, connectionConfigs); err != nil {
+		if missing, ok := errors.AsType[*common.SourceTablesMissingError](err); ok {
+			return nil, NewFailedPreconditionApiError(
+				missing,
+				NewSourceTableMissingErrorInfo(),
+				NewSourceTableMissingPreconditionFailure(missing.Tables))
+		}
+		if notInPub, ok := errors.AsType[*common.TablesNotInPublicationError](err); ok {
+			return nil, NewFailedPreconditionApiError(
+				notInPub,
+				NewTablesNotInPublicationErrorInfo(notInPub.Publication),
+				NewTablesNotInPublicationPreconditionFailure(notInPub.Publication, notInPub.Tables))
+		}
 		return nil, NewFailedPreconditionApiError(
 			fmt.Errorf("failed to validate source connector %s: %w", connectionConfigs.SourceName, err))
 	}

flow/connectors/bigquery/source.go

@@ -4,18 +4,21 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
 
 	"cloud.google.com/go/storage"
 	"google.golang.org/api/iterator"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/pkg/common"
 )
 
 func (c *BigQueryConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.FlowConnectionConfigsCore) error {
 	if !cfg.InitialSnapshotOnly || !cfg.DoInitialSnapshot {
 		return fmt.Errorf("BigQuery source connector only supports initial snapshot flows. CDC is not supported")
 	}
 
+	var missingTables []common.QualifiedTable
 	for _, tableMapping := range cfg.TableMappings {
 		dstDatasetTable, err := c.convertToDatasetTable(tableMapping.SourceTableIdentifier)
 		if err != nil {
@@ -25,9 +28,19 @@ func (c *BigQueryConnector) ValidateMirrorSource(ctx context.Context, cfg *proto
 		table := c.client.DatasetInProject(c.projectID, dstDatasetTable.dataset).Table(dstDatasetTable.table)
 
 		if _, err := table.Metadata(ctx); err != nil {
+			if c.isApiErrorWithStatusCode(err, http.StatusNotFound) {
+				missingTables = append(missingTables, common.QualifiedTable{
+					Namespace: dstDatasetTable.dataset,
+					Table:     dstDatasetTable.table,
+				})
+				continue
+			}
 			return fmt.Errorf("failed to get metadata for table %s: %w", tableMapping.DestinationTableIdentifier, err)
 		}
 	}
+	if len(missingTables) > 0 {
+		return common.NewSourceTablesMissingError(missingTables)
+	}
 
 	if cfg.SnapshotStagingPath == "" {
 		return fmt.Errorf("snapshot bucket is required for BigQuery source connector")

flow/connectors/core.go

@@ -44,6 +44,8 @@ type ValidationConnector interface {
 type MirrorSourceValidationConnector interface {
 	GetTableSchemaConnector
 
+	// ValidateMirrorSource checks that the source is ready to replicate the configured tables.
+	// MUST return *common.SourceTablesMissingError when a mapped source table is absent.
 	ValidateMirrorSource(context.Context, *protos.FlowConnectionConfigsCore) error
 }
 

flow/connectors/mongo/validate.go

@@ -21,14 +21,6 @@ func (c *MongoConnector) ValidateCheck(ctx context.Context) error {
 }
 
 func (c *MongoConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.FlowConnectionConfigsCore) error {
-	if cfg.DoInitialSnapshot && cfg.InitialSnapshotOnly {
-		return nil
-	}
-
-	if err := shared_mongo.ValidateOplogRetention(ctx, c.client); err != nil {
-		return err
-	}
-
 	tables := make([]*common.QualifiedTable, 0, len(cfg.TableMappings))
 	for _, tm := range cfg.TableMappings {
 		t, err := common.ParseTableIdentifier(tm.SourceTableIdentifier)
@@ -41,5 +33,10 @@ func (c *MongoConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.F
 		return err
 	}
 
-	return nil
+	// no need to check oplog retention for initial-snapshot-only mirrors
+	if cfg.DoInitialSnapshot && cfg.InitialSnapshotOnly {
+		return nil
+	}
+
+	return shared_mongo.ValidateOplogRetention(ctx, c.client)
 }

flow/connectors/mysql/validate.go

@@ -15,11 +15,19 @@ import (
 )
 
 func (c *MySqlConnector) CheckSourceTables(ctx context.Context, tableNames []*common.QualifiedTable) error {
+	var missingTables []common.QualifiedTable
 	for _, parsedTable := range tableNames {
 		if _, err := c.Execute(ctx, fmt.Sprintf("SELECT * FROM %s LIMIT 0", parsedTable.MySQL())); err != nil {
+			if mErr, ok := errors.AsType[*mysql.MyError](err); ok && mErr.Code == mysql.ER_NO_SUCH_TABLE {
+				missingTables = append(missingTables, *parsedTable)
+				continue
+			}
 			return fmt.Errorf("error checking table %s: %w", parsedTable.MySQL(), err)
 		}
 	}
+	if len(missingTables) > 0 {
+		return common.NewSourceTablesMissingError(missingTables)
+	}
 	return nil
 }
 

flow/connectors/postgres/validate.go

@@ -7,7 +7,9 @@ import (
 	"slices"
 	"strings"
 
+	"github.com/jackc/pgerrcode"
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
@@ -41,6 +43,7 @@ func (c *PostgresConnector) CheckSourceTables(
 
 	// Check that we can select from all tables
 	tableArr := make([]string, 0, len(tableNames))
+	var missingTables []common.QualifiedTable
 	for idx, parsedTable := range tableNames {
 		var row pgx.Row
 		tableArr = append(tableArr, fmt.Sprintf(`(%s::text,%s::text)`,
@@ -62,9 +65,16 @@ func (c *PostgresConnector) CheckSourceTables(
 		if err := c.conn.QueryRow(ctx,
 			fmt.Sprintf("SELECT %s FROM %s LIMIT 0", selectedColumnsStr, parsedTable),
 		).Scan(&row); err != nil && !errors.Is(err, pgx.ErrNoRows) {
+			if pgErr, ok := errors.AsType[*pgconn.PgError](err); ok && pgErr.Code == pgerrcode.UndefinedTable {
+				missingTables = append(missingTables, *parsedTable)
+				continue
+			}
 			return fmt.Errorf("failed to select from table %s: %w", parsedTable, err)
 		}
 	}
+	if len(missingTables) > 0 {
+		return common.NewSourceTablesMissingError(missingTables)
+	}
 
 	if pubName != "" && !noCDC {
 		// Check if publication exists
@@ -93,20 +103,19 @@ func (c *PostgresConnector) CheckSourceTables(
 			if err != nil {
 				return err
 			}
-			missing, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (string, error) {
-				var schema string
-				var table string
-				if err := row.Scan(&schema, &table); err != nil {
-					return "", err
+			missing, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (common.QualifiedTable, error) {
+				var qt common.QualifiedTable
+				if err := row.Scan(&qt.Namespace, &qt.Table); err != nil {
+					return common.QualifiedTable{}, err
 				}
-				return fmt.Sprintf("%s.%s", common.QuoteIdentifier(schema), common.QuoteIdentifier(table)), nil
+				return qt, nil
 			})
 			if err != nil {
 				return err
 			}
 
 			if len(missing) != 0 {
-				return fmt.Errorf("some tables missing from publication: %s", strings.Join(missing, ", "))
+				return common.NewTablesNotInPublicationError(pubName, missing)
 			}
 		}
 	}
@@ -241,6 +250,11 @@ func (c *PostgresConnector) ValidateMirrorSource(ctx context.Context, cfg *proto
 	}
 
 	pubName := cfg.PublicationName
+	// Check source tables before the publication, for better errors
+	if err := c.CheckSourceTables(ctx, sourceTables, cfg.TableMappings, pubName, noCDC); err != nil {
+		return fmt.Errorf("provided source tables invalidated: %w", err)
+	}
+
 	if pubName == "" && !noCDC {
 		srcTableNames := make([]string, 0, len(sourceTables))
 		for _, srcTable := range sourceTables {
@@ -252,10 +266,6 @@ func (c *PostgresConnector) ValidateMirrorSource(ctx context.Context, cfg *proto
 		}
 	}
 
-	if err := c.CheckSourceTables(ctx, sourceTables, cfg.TableMappings, pubName, noCDC); err != nil {
-		return fmt.Errorf("provided source tables invalidated: %w", err)
-	}
-
 	return nil
 }