diff --git a/flow/cmd/api_error.go b/flow/cmd/api_error.go index 1dfaa8429..65d371101 100644 --- a/flow/cmd/api_error.go +++ b/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") diff --git a/flow/cmd/handler.go b/flow/cmd/handler.go index 83511bace..4a37d77df 100644 --- a/flow/cmd/handler.go +++ b/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)) } } diff --git a/flow/cmd/validate_mirror.go b/flow/cmd/validate_mirror.go index 1b3fe2390..d6882a1de 100644 --- a/flow/cmd/validate_mirror.go +++ b/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)) } diff --git a/flow/connectors/bigquery/source.go b/flow/connectors/bigquery/source.go index 0ead1dd6d..13eebee50 100644 --- a/flow/connectors/bigquery/source.go +++ b/flow/connectors/bigquery/source.go @@ -4,11 +4,13 @@ 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 { @@ -16,6 +18,7 @@ func (c *BigQueryConnector) ValidateMirrorSource(ctx context.Context, cfg *proto 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") diff --git a/flow/connectors/core.go b/flow/connectors/core.go index d8275a4c5..2179ba7e1 100644 --- a/flow/connectors/core.go +++ b/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 } diff --git a/flow/connectors/mongo/validate.go b/flow/connectors/mongo/validate.go index 5f6a1bbec..faa6cea07 100644 --- a/flow/connectors/mongo/validate.go +++ b/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) } diff --git a/flow/connectors/mysql/validate.go b/flow/connectors/mysql/validate.go index 3fe604c84..0ef83e4cc 100644 --- a/flow/connectors/mysql/validate.go +++ b/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 } diff --git a/flow/connectors/postgres/validate.go b/flow/connectors/postgres/validate.go index 099a91a95..4d942747a 100644 --- a/flow/connectors/postgres/validate.go +++ b/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 } diff --git a/flow/e2e/api_test.go b/flow/e2e/api_test.go index 171886095..4903bcabe 100644 --- a/flow/e2e/api_test.go +++ b/flow/e2e/api_test.go @@ -20,6 +20,7 @@ import ( "go.mongodb.org/mongo-driver/v2/mongo/options" "go.temporal.io/api/enums/v1" "go.temporal.io/api/workflowservice/v1" + "google.golang.org/genproto/googleapis/rpc/errdetails" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" @@ -1370,6 +1371,227 @@ func (s APITestSuite) TestResyncCompleted() { require.Equal(s.t, "test", tagMap[common.PipeNameTag]) } +func (s APITestSuite) TestResyncSourceTableMissing() { + tableNames := []string{"missing_src_a", "missing_src_b"} + qualifiedSourceTables := []string{AttachSchema(s, tableNames[0]), AttachSchema(s, tableNames[1])} + var cols string + switch s.source.(type) { + case *PostgresSource, *MySqlSource: + for _, qt := range qualifiedSourceTables { + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("CREATE TABLE %s(id int primary key, val text)", qt))) + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("INSERT INTO %s(id, val) values (1,'first')", qt))) + } + cols = "id,val" + case *MongoSource: + for _, tn := range tableNames { + res, err := s.Source().(*MongoSource).AdminClient(). + Database(Schema(s)).Collection(tn). + InsertOne(s.t.Context(), bson.D{bson.E{Key: "id", Value: 1}, bson.E{Key: "val", Value: "first"}}, options.InsertOne()) + require.NoError(s.t, err) + require.True(s.t, res.Acknowledged) + } + cols = fmt.Sprintf("%s,%s", connmongo.DefaultDocumentKeyColumnName, connmongo.DefaultFullDocumentColumnName) + default: + require.Fail(s.t, fmt.Sprintf("unknown source type %T", s.source)) + } + + connectionGen := FlowConnectionGenerationConfig{ + FlowJobName: "resync_missing_" + s.suffix, + TableNameMapping: map[string]string{ + qualifiedSourceTables[0]: tableNames[0], + qualifiedSourceTables[1]: tableNames[1], + }, + Destination: s.ch.Peer().Name, + } + flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) + flowConnConfig.DoInitialSnapshot = true + flowConnConfig.InitialSnapshotOnly = true + + response, err := s.CreateCDCFlow(s.t.Context(), &protos.CreateCDCFlowRequest{ConnectionConfigs: flowConnConfig}) + require.NoError(s.t, err) + require.NotNil(s.t, response) + + tc := NewTemporalClient(s.t) + env, err := GetPeerflow(s.t.Context(), s.catalog, tc, flowConnConfig.FlowJobName) + require.NoError(s.t, err) + SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) + EnvWaitForFinished(s.t, env, 3*time.Minute) + for _, tn := range tableNames { + RequireEqualTables(s.ch, tn, cols) + } + + switch src := s.source.(type) { + case *PostgresSource, *MySqlSource: + for _, qt := range qualifiedSourceTables { + require.NoError(s.t, s.source.Exec(s.t.Context(), "DROP TABLE "+qt)) + } + case *MongoSource: + for _, tn := range tableNames { + require.NoError(s.t, src.AdminClient().Database(Schema(s)).Collection(tn).Drop(s.t.Context())) + } + default: + require.Fail(s.t, fmt.Sprintf("unknown source type %T", s.source)) + } + + _, err = s.FlowStateChange(s.t.Context(), &protos.FlowStateChangeRequest{ + FlowJobName: flowConnConfig.FlowJobName, + RequestedFlowState: protos.FlowStatus_STATUS_RESYNC, + }) + require.Error(s.t, err) + + // Shape of the error on the wire (see AIP-193, https://google.aip.dev/193): + // status.Code = FailedPrecondition + // status.Details = [ + // google.rpc.ErrorInfo{ + // Domain: "peerdb.io", Reason: "SOURCE_TABLE_MISSING", + // }, + // google.rpc.PreconditionFailure{ + // Violations: [ + // {Type: "SOURCE_TABLE_MISSING", Subject: ".", Description: "..."}, + // {Type: "SOURCE_TABLE_MISSING", Subject: ".", Description: "..."}, + // ], + // }, + // ] + st, ok := status.FromError(err) + require.True(s.t, ok, "expected gRPC status error, got %T: %v", err, err) + require.Equal(s.t, codes.FailedPrecondition, st.Code(), "expected FailedPrecondition, got %s", st.Code()) + + var hasSourceTableMissing bool + var violations []*errdetails.PreconditionFailure_Violation + for _, d := range st.Details() { + switch detail := d.(type) { + case *errdetails.ErrorInfo: + if detail.Domain == common.ErrorInfoDomain && detail.Reason == common.ErrorInfoReasonSourceTableMissing { + hasSourceTableMissing = true + } + case *errdetails.PreconditionFailure: + violations = append(violations, detail.Violations...) + } + } + require.True(s.t, hasSourceTableMissing, "expected SourceTableMissing ErrorInfo detail in status") + require.Len(s.t, violations, len(tableNames), "expected one PreconditionFailure violation per dropped table") + gotSubjects := make([]string, len(violations)) + for i, v := range violations { + require.Equal(s.t, common.ErrorInfoReasonSourceTableMissing, v.Type) + gotSubjects[i] = v.Subject + } + wantSubjects := make([]string, len(tableNames)) + for i, tn := range tableNames { + wantSubjects[i] = fmt.Sprintf("%s.%s", Schema(s), tn) + } + require.ElementsMatch(s.t, wantSubjects, gotSubjects) +} + +func (s APITestSuite) TestResyncTablesNotInPublication() { + if _, ok := s.source.(*PostgresSource); !ok { + s.t.Skip("only for PostgreSQL (publications are PG-specific)") + } + + tableNames := []string{"resync_pub_a", "resync_pub_b"} + qualifiedSourceTables := []string{AttachSchema(s, tableNames[0]), AttachSchema(s, tableNames[1])} + for _, qt := range qualifiedSourceTables { + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("CREATE TABLE %s(id int primary key, val text)", qt))) + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("INSERT INTO %s(id, val) values (1,'first')", qt))) + } + + pubName := "pub_resync_" + s.suffix + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("CREATE PUBLICATION %s FOR TABLE %s, %s", pubName, qualifiedSourceTables[0], qualifiedSourceTables[1]))) + s.t.Cleanup(func() { + _ = s.source.Exec(context.Background(), "DROP PUBLICATION IF EXISTS "+pubName) + }) + + connectionGen := FlowConnectionGenerationConfig{ + FlowJobName: "resync_not_in_pub_" + s.suffix, + TableNameMapping: map[string]string{ + qualifiedSourceTables[0]: tableNames[0], + qualifiedSourceTables[1]: tableNames[1], + }, + Destination: s.ch.Peer().Name, + } + flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) + flowConnConfig.DoInitialSnapshot = true + flowConnConfig.PublicationName = pubName + + response, err := s.CreateCDCFlow(s.t.Context(), &protos.CreateCDCFlowRequest{ConnectionConfigs: flowConnConfig}) + require.NoError(s.t, err) + require.NotNil(s.t, response) + + tc := NewTemporalClient(s.t) + env, err := GetPeerflow(s.t.Context(), s.catalog, tc, flowConnConfig.FlowJobName) + require.NoError(s.t, err) + SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) + for _, tn := range tableNames { + EnvWaitForCount(env, s.ch, "initial snapshot", tn, "id,val", 1) + } + EnvWaitFor(s.t, env, 3*time.Minute, "flow running", func() bool { + return env.GetFlowStatus(s.t) == protos.FlowStatus_STATUS_RUNNING + }) + + for _, qt := range qualifiedSourceTables { + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("ALTER PUBLICATION %s DROP TABLE %s", pubName, qt))) + } + + _, err = s.FlowStateChange(s.t.Context(), &protos.FlowStateChangeRequest{ + FlowJobName: flowConnConfig.FlowJobName, + RequestedFlowState: protos.FlowStatus_STATUS_RESYNC, + }) + require.Error(s.t, err) + + // Shape of the error on the wire: + // status.Code = FailedPrecondition + // status.Details = [ + // google.rpc.ErrorInfo{ + // Domain: "peerdb.io", Reason: "TABLES_NOT_IN_PUBLICATION", + // Metadata: {"publication": ""}, + // }, + // google.rpc.PreconditionFailure{ + // Violations: [ + // {Type: "TABLES_NOT_IN_PUBLICATION", Subject: ".", Description: "..."}, + // {Type: "TABLES_NOT_IN_PUBLICATION", Subject: ".", Description: "..."}, + // ], + // }, + // ] + st, ok := status.FromError(err) + require.True(s.t, ok, "expected gRPC status error, got %T: %v", err, err) + require.Equal(s.t, codes.FailedPrecondition, st.Code(), "expected FailedPrecondition, got %s", st.Code()) + + var gotReason, gotPublication string + var violations []*errdetails.PreconditionFailure_Violation + for _, d := range st.Details() { + switch detail := d.(type) { + case *errdetails.ErrorInfo: + if detail.Domain == common.ErrorInfoDomain { + gotReason = detail.Reason + gotPublication = detail.Metadata[common.ErrorMetadataPublication] + } + case *errdetails.PreconditionFailure: + violations = append(violations, detail.Violations...) + } + } + require.Equal(s.t, common.ErrorInfoReasonTablesNotInPublication, gotReason) + require.Equal(s.t, pubName, gotPublication) + require.Len(s.t, violations, len(tableNames)) + gotSubjects := make([]string, len(violations)) + for i, v := range violations { + require.Equal(s.t, common.ErrorInfoReasonTablesNotInPublication, v.Type) + gotSubjects[i] = v.Subject + } + wantSubjects := make([]string, len(tableNames)) + for i, tn := range tableNames { + wantSubjects[i] = fmt.Sprintf("%s.%s", Schema(s), tn) + } + require.ElementsMatch(s.t, wantSubjects, gotSubjects) + + env.Cancel(s.t.Context()) + RequireEnvCanceled(s.t, env) +} + func (s APITestSuite) TestResyncFailed() { pgSource, ok := s.source.(*PostgresSource) if !ok { diff --git a/flow/e2e/bigquery_source_test.go b/flow/e2e/bigquery_source_test.go index 9e727dc51..f020d77d9 100644 --- a/flow/e2e/bigquery_source_test.go +++ b/flow/e2e/bigquery_source_test.go @@ -19,6 +19,7 @@ import ( "github.com/PeerDB-io/peerdb/flow/e2eshared" "github.com/PeerDB-io/peerdb/flow/generated/protos" "github.com/PeerDB-io/peerdb/flow/model" + "github.com/PeerDB-io/peerdb/flow/pkg/common" "github.com/PeerDB-io/peerdb/flow/shared" "github.com/PeerDB-io/peerdb/flow/shared/types" ) @@ -302,7 +303,11 @@ func (s BigQueryClickhouseSuite) Test_BigQuery_Source_Invalid_Table_Mappings() { err := bqConn.ValidateMirrorSource(ctx, flowConfig) require.Error(t, err, "should fail with non-existent table") - require.Contains(t, err.Error(), "failed to get metadata for table", "error should mention metadata failure") + missing, ok := errors.AsType[*common.SourceTablesMissingError](err) + require.True(t, ok, "expected SourceTableMissingError, got %T: %v", err, err) + require.Len(t, missing.Tables, 1) + require.Equal(t, source.config.DatasetId, missing.Tables[0].Namespace) + require.Equal(t, "nonexistent_table", missing.Tables[0].Table) } func (s BigQueryClickhouseSuite) Test_BigQuery_Source_ValidateMirrorSource_Success() { diff --git a/flow/pkg/common/errors.go b/flow/pkg/common/errors.go new file mode 100644 index 000000000..68f695527 --- /dev/null +++ b/flow/pkg/common/errors.go @@ -0,0 +1,51 @@ +package common + +import ( + "fmt" + "strings" +) + +// gRPC ErrorInfo constants +const ( + ErrorInfoDomain = "peerdb.io" + + ErrorInfoReasonMirror = "MIRROR" + ErrorInfoReasonSourceTableMissing = "SOURCE_TABLE_MISSING" + ErrorInfoReasonTablesNotInPublication = "TABLES_NOT_IN_PUBLICATION" + + ErrorMetadataOffendingField = "offendingField" + ErrorMetadataPublication = "publication" +) + +type SourceTablesMissingError struct { + Tables []QualifiedTable +} + +func NewSourceTablesMissingError(tables []QualifiedTable) *SourceTablesMissingError { + return &SourceTablesMissingError{Tables: tables} +} + +func (e *SourceTablesMissingError) Error() string { + parts := make([]string, len(e.Tables)) + for i, t := range e.Tables { + parts[i] = fmt.Sprintf("%s.%s", t.Namespace, t.Table) + } + return "source tables do not exist: " + strings.Join(parts, ", ") +} + +type TablesNotInPublicationError struct { + Tables []QualifiedTable + Publication string +} + +func NewTablesNotInPublicationError(publication string, tables []QualifiedTable) *TablesNotInPublicationError { + return &TablesNotInPublicationError{Publication: publication, Tables: tables} +} + +func (e *TablesNotInPublicationError) Error() string { + parts := make([]string, len(e.Tables)) + for i, t := range e.Tables { + parts[i] = fmt.Sprintf("%s.%s", t.Namespace, t.Table) + } + return fmt.Sprintf("tables not in publication %q: %s", e.Publication, strings.Join(parts, ", ")) +} diff --git a/flow/pkg/mongo/validation.go b/flow/pkg/mongo/validation.go index 44653662a..0c8852f03 100644 --- a/flow/pkg/mongo/validation.go +++ b/flow/pkg/mongo/validation.go @@ -116,6 +116,7 @@ func ValidateCollections(ctx context.Context, client *mongo.Client, tables []*co databaseCollectionsMapping[t.Namespace] = append(databaseCollectionsMapping[t.Namespace], t.Table) } + var missingTables []common.QualifiedTable for database, collections := range databaseCollectionsMapping { allCollections, err := GetCollectionNames(ctx, client, database) if err != nil { @@ -123,10 +124,13 @@ func ValidateCollections(ctx context.Context, client *mongo.Client, tables []*co } for _, col := range collections { if !slices.Contains(allCollections, col) { - return fmt.Errorf("collection %s.%s does not exist", database, col) + missingTables = append(missingTables, common.QualifiedTable{Namespace: database, Table: col}) } } } + if len(missingTables) > 0 { + return common.NewSourceTablesMissingError(missingTables) + } return nil }