Gate BigQuery classification on an explicit error wrapper (#4309)

The previous classifier change matched bare *googleapi.Error and
labelled every match as bigquery. But *googleapi.Error is the shared Go
error type for any HTTP-level Google API failure — GCS-outside-BQ, KMS,
Pub/Sub, etc. all surface it. A stray googleapi error from anywhere in
the codebase would be misattributed to BigQuery, polluting per-source
telemetry.

Introduce exceptions.BigQueryError as an explicit wrapper that callers
in the BigQuery connector use to mark "this error came from a BQ code
path", and gate BigQuery classification on its presence.

Inside the gate, both SDK error types are inspected because they come
from different paths and never co-occur:
- *googleapi.Error (HTTP-level, integer Code) — from API call failures
like Run, Wait, Metadata, Create, Delete, GCS iterator Next. 401/403/
404 map to connectivity.
- *bigquery.Error (job-level, string Reason like "invalidQuery" or
"accessDenied") — produced by JobStatus.Err() when a job is accepted by
the API but fails during execution. The Reason is surfaced as the code
for diagnostic clarity; we don't classify individual Reasons yet, so the
class falls through to ErrorOther with BQ source.

Tests updated to wrap inputs in BigQueryError; new negative test asserts
that an unwrapped *googleapi.Error from outside the BQ connector now
falls through to ErrorSourceOther rather than being mislabelled.

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Jeremyyang920

flow/alerting/classifier.go

@@ -13,6 +13,7 @@ import (
 	"strings"
 	"syscall"
 
+	"cloud.google.com/go/bigquery"
 	chproto "github.com/ClickHouse/ch-go/proto"
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/go-mysql-org/go-mysql/mysql"
@@ -874,16 +875,21 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		return ErrorRetryRecoverable, mongoErrorInfo
 	}
 
-	if apiErr, ok := errors.AsType[*googleapi.Error](err); ok {
+	if _, ok := errors.AsType[*exceptions.BigQueryError](err); ok {
 		bqErrorInfo := ErrorInfo{
 			Source: ErrorSourceBigQuery,
-			Code:   strconv.Itoa(apiErr.Code),
+			Code:   "UNKNOWN",
 		}
-		switch apiErr.Code {
-		case 401, // Unauthorized
-			403, // Forbidden
-			404: // Not Found (e.g. missing dataset/table/staging bucket)
-			return ErrorNotifyConnectivity, bqErrorInfo
+		if apiErr, ok := errors.AsType[*googleapi.Error](err); ok {
+			bqErrorInfo.Code = strconv.Itoa(apiErr.Code)
+			switch apiErr.Code {
+			case 401, // Unauthorized
+				403, // Forbidden
+				404: // Not Found (e.g. missing dataset/table/staging bucket)
+				return ErrorNotifyConnectivity, bqErrorInfo
+			}
+		} else if bqErr, ok := errors.AsType[*bigquery.Error](err); ok && bqErr.Reason != "" {
+			bqErrorInfo.Code = bqErr.Reason
 		}
 		return ErrorOther, bqErrorInfo
 	}

flow/alerting/classifier_test.go

@@ -10,6 +10,7 @@ import (
 	"syscall"
 	"testing"
 
+	"cloud.google.com/go/bigquery"
 	"cloud.google.com/go/storage"
 	chproto "github.com/ClickHouse/ch-go/proto"
 	"github.com/ClickHouse/clickhouse-go/v2"
@@ -1018,15 +1019,16 @@ func TestMySQLGeometryParseErrorUnknownShouldFallThrough(t *testing.T) {
 	}, errInfo)
 }
 
-func TestGoogleAPIAuthAndAccessErrorsShouldBeConnectivity(t *testing.T) {
+func TestBigQueryAuthAndAccessErrorsShouldBeConnectivity(t *testing.T) {
 	t.Parallel()
 
 	for _, code := range []int{401, 403, 404} {
 		t.Run(strconv.Itoa(code), func(t *testing.T) {
 			t.Parallel()
 
 			apiErr := &googleapi.Error{Code: code, Message: "boom"}
-			errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("failed to access staging bucket: %w", apiErr))
+			err := exceptions.NewBigQueryError(apiErr)
+			errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("failed to access staging bucket: %w", err))
 			assert.Equal(t, ErrorNotifyConnectivity, errorClass)
 			assert.Equal(t, ErrorInfo{
 				Source: ErrorSourceBigQuery,
@@ -1041,28 +1043,73 @@ func TestGoogleAPIAuthAndAccessErrorsShouldBeConnectivity(t *testing.T) {
 //	return fmt.Errorf("%w: %w", ErrBucketNotExist, err)
 //
 // `it.Next()` on a missing staging bucket surfaces this exact shape at
-// connectors/bigquery/source.go:59.
-func TestGCSBucketNotExistShouldBeConnectivity(t *testing.T) {
+// connectors/bigquery/source.go:59 (then wrapped in BigQueryError there).
+func TestBigQueryGCSBucketNotExistShouldBeConnectivity(t *testing.T) {
 	t.Parallel()
 
 	apiErr := &googleapi.Error{Code: 404, Message: "Not Found"}
-	wrapped := fmt.Errorf("%w: %w", storage.ErrBucketNotExist, apiErr)
-	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("failed to access staging bucket: %w", wrapped))
+	storageWrapped := fmt.Errorf("%w: %w", storage.ErrBucketNotExist, apiErr)
+	err := exceptions.NewBigQueryError(storageWrapped)
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("failed to access staging bucket: %w", err))
 	assert.Equal(t, ErrorNotifyConnectivity, errorClass)
 	assert.Equal(t, ErrorInfo{
 		Source: ErrorSourceBigQuery,
 		Code:   "404",
 	}, errInfo)
 }
 
-func TestGoogleAPIUnclassifiedCodeShouldBeOther(t *testing.T) {
+func TestBigQueryUnclassifiedCodeShouldBeOther(t *testing.T) {
 	t.Parallel()
 
 	apiErr := &googleapi.Error{Code: 500, Message: "internal"}
-	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("export job failed: %w", apiErr))
+	err := exceptions.NewBigQueryError(apiErr)
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("export job failed: %w", err))
 	assert.Equal(t, ErrorOther, errorClass)
 	assert.Equal(t, ErrorInfo{
 		Source: ErrorSourceBigQuery,
 		Code:   "500",
 	}, errInfo)
 }
+
+func TestBigQueryErrorWithoutGoogleAPIShouldBeOther(t *testing.T) {
+	t.Parallel()
+
+	err := exceptions.NewBigQueryError(fmt.Errorf("opaque failure"))
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("bq op failed: %w", err))
+	assert.Equal(t, ErrorOther, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceBigQuery,
+		Code:   "UNKNOWN",
+	}, errInfo)
+}
+
+// bigquery.Error (with a Reason like "invalidQuery", "accessDenied") is what
+// surfaces from job-level failures e.g. status.Err() on Job.Wait. We surface
+// the Reason as the code for diagnostic clarity even though we don't classify
+// individual Reasons yet.
+func TestBigQueryJobErrorWithReasonShouldUseReasonAsCode(t *testing.T) {
+	t.Parallel()
+
+	jobErr := &bigquery.Error{Reason: "invalidQuery", Message: "bad SQL", Location: "query"}
+	err := exceptions.NewBigQueryError(jobErr)
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("export job completed with error: %w", err))
+	assert.Equal(t, ErrorOther, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceBigQuery,
+		Code:   "invalidQuery",
+	}, errInfo)
+}
+
+// Unwrapped *googleapi.Error from outside the BigQuery connector must NOT be
+// attributed to BigQuery — only errors explicitly wrapped in BigQueryError are.
+func TestUnwrappedGoogleAPIErrorShouldNotBeBigQuery(t *testing.T) {
+	t.Parallel()
+
+	apiErr := &googleapi.Error{Code: 403, Message: "forbidden"}
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("some other gcp call: %w", apiErr))
+	assert.Equal(t, ErrorOther, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceOther,
+		Code:   "UNKNOWN",
+	}, errInfo)
+}

flow/connectors/bigquery/source.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/pkg/common"
+	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 )
 
 func (c *BigQueryConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.FlowConnectionConfigsCore) error {
@@ -56,7 +57,7 @@ func (c *BigQueryConnector) ValidateMirrorSource(ctx context.Context, cfg *proto
 	it := bucket.Objects(ctx, &storage.Query{Prefix: stagingPath.QueryPrefix()})
 	_, err = it.Next()
 	if err != nil && !errors.Is(err, iterator.Done) {
-		return fmt.Errorf("failed to access staging bucket: %w", err)
+		return fmt.Errorf("failed to access staging bucket: %w", exceptions.NewBigQueryError(err))
 	}
 
 	return nil

flow/shared/exceptions/bigquery.go

@@ -0,0 +1,17 @@
+package exceptions
+
+type BigQueryError struct {
+	error
+}
+
+func NewBigQueryError(err error) *BigQueryError {
+	return &BigQueryError{err}
+}
+
+func (e *BigQueryError) Error() string {
+	return "BigQuery Error: " + e.error.Error()
+}
+
+func (e *BigQueryError) Unwrap() error {
+	return e.error
+}