Return structured gRPC errors on source table missing

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

@@ -505,7 +505,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 {
@@ -534,6 +534,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
 }