Merge remote-tracking branch 'upstream/main' into DBI-640/local-dev-env/e2e/toxy+split_pgcon

Author: pfcoperez

.tiltignore

@@ -0,0 +1,2 @@
+# When launching tests via VS Code, it builds binaries with this prefix in the test folder
+**/__debug_bin*

buf.gen.yaml

@@ -9,7 +9,7 @@ plugins:
   - plugin: buf.build/protocolbuffers/go:v1.36.11
     out: flow/generated/protos
     opt: paths=source_relative
-  - plugin: buf.build/grpc/go:v1.5.1
+  - plugin: buf.build/grpc/go:v1.6.1
     out: flow/generated/protos
     opt:
       - paths=source_relative
@@ -24,12 +24,12 @@ plugins:
     out: nexus/pt/src/gen
     opt:
       - ignore_unknown_fields=true
-  - plugin: buf.build/community/stephenh-ts-proto:v2.11.2
+  - plugin: buf.build/community/stephenh-ts-proto:v2.11.6
     out: ui/grpc_generated
     opt:
       - esModuleInterop=true
       - outputServices=none
-  - plugin: buf.build/grpc-ecosystem/gateway:v2.27.7
+  - plugin: buf.build/grpc-ecosystem/gateway:v2.28.0
     out: flow/generated/protos
     opt:
       - paths=source_relative

flow/activities/flowable.go

@@ -827,21 +827,24 @@ func (a *FlowableActivity) DropFlowSource(ctx context.Context, req *protos.DropF
 	ctx = context.WithValue(ctx, shared.FlowNameKey, req.FlowJobName)
 	srcConn, srcClose, err := connectors.GetByNameAs[connectors.CDCPullConnectorCore](ctx, nil, a.CatalogPool, req.PeerName)
 	if err != nil {
-		var notFound *exceptions.NotFoundError
-		if errors.As(err, &notFound) {
+		if _, ok := errors.AsType[*exceptions.NotFoundError](err); ok {
 			logger := internal.LoggerFromCtx(ctx)
 			logger.Warn("peer missing, skipping", slog.String("peer", req.PeerName))
 			return nil
 		}
+		if _, ok := errors.AsType[*exceptions.AuthError](err); ok {
+			logger := internal.LoggerFromCtx(ctx)
+			logger.Warn("auth error, skipping to avoid triggering security tools", slog.String("peer", req.PeerName))
+			return nil
+		}
 		return a.Alerter.LogFlowError(ctx, req.FlowJobName,
 			exceptions.NewDropFlowError(fmt.Errorf("[DropFlowSource] failed to get source connector: %w", err)),
 		)
 	}
 	defer srcClose(ctx)
 
 	if err := srcConn.PullFlowCleanup(ctx, req.FlowJobName); err != nil {
-		var dnsErr *net.DNSError
-		if errors.As(err, &dnsErr) && dnsErr.IsNotFound {
+		if dnsErr, ok := errors.AsType[*net.DNSError](err); ok && dnsErr.IsNotFound {
 			a.Alerter.LogFlowWarning(ctx, req.FlowJobName, fmt.Errorf("[DropFlowSource] hostname not found, skipping: %w", err))
 			return nil
 		} else {

flow/activities/flowable_core.go

@@ -728,58 +728,64 @@ func (a *FlowableActivity) normalizeLoop(
 	defer normalizeWaiting.Store(false)
 
 	for {
-		normalizeWaiting.Store(true)
-		ch := normalizeRequests.Wait()
-		if ch == nil {
-			logger.Info("[normalize-loop] lastChan closed")
-			return
-		}
-		select {
-		case <-syncDone:
-			logger.Info("[normalize-loop] syncDone closed")
-			return
-		case <-ctx.Done():
-			logger.Info("[normalize-loop] context closed")
-			return
-		case <-ch:
-			reqBatchID := normalizeRequests.Load()
-			lastNormalizedBatchID := normalizingBatchID.Load()
-			if reqBatchID <= lastNormalizedBatchID {
+		// Check for pending work before waiting on the channel.
+		// This avoids a race where Update() replaces the channel while we're
+		// processing, causing Wait() to return the new (unclosed) channel
+		// and missing the signal entirely until the next Update.
+		reqBatchID := normalizeRequests.Load()
+		lastNormalizedBatchID := normalizingBatchID.Load()
+		if reqBatchID <= lastNormalizedBatchID {
+			normalizeWaiting.Store(true)
+			ch := normalizeRequests.Wait()
+			if ch == nil {
+				logger.Info("[normalize-loop] lastChan closed")
+				return
+			}
+			select {
+			case <-syncDone:
+				logger.Info("[normalize-loop] syncDone closed")
+				return
+			case <-ctx.Done():
+				logger.Info("[normalize-loop] context closed")
+				return
+			case <-ch:
 				continue
 			}
-			retryInterval := time.Minute
-		retryLoop:
-			for {
-				normalizingBatchID.Store(reqBatchID)
-				if err := a.startNormalize(ctx, config, reqBatchID, normalizeResponses); err != nil {
-					_ = a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
-					for {
-						// update req to latest normalize request & retry
-						select {
-						case <-syncDone:
-							logger.Info("[normalize-loop] syncDone closed before retry")
-							return
-						case <-ctx.Done():
-							logger.Info("[normalize-loop] context closed before retry")
-							return
-						default:
-							time.Sleep(retryInterval)
-							retryInterval = min(retryInterval*2, 5*time.Minute)
-							// record the last normalized batch ID even if retry fails to populate metrics consistently
-							a.OtelManager.Metrics.LastNormalizedBatchIdGauge.Record(
-								ctx, lastNormalizedBatchID, metric.WithAttributeSet(attribute.NewSet(
-									attribute.String(otel_metrics.FlowNameKey, config.FlowJobName),
-								)))
-							reqBatchID = normalizeRequests.Load()
-							continue retryLoop
-						}
+		}
+		normalizeWaiting.Store(false)
+
+		retryInterval := time.Minute
+	retryLoop:
+		for {
+			normalizingBatchID.Store(reqBatchID)
+			if err := a.startNormalize(ctx, config, reqBatchID, normalizeResponses); err != nil {
+				_ = a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
+				for {
+					// update req to latest normalize request & retry
+					select {
+					case <-syncDone:
+						logger.Info("[normalize-loop] syncDone closed before retry")
+						return
+					case <-ctx.Done():
+						logger.Info("[normalize-loop] context closed before retry")
+						return
+					default:
+						time.Sleep(retryInterval)
+						retryInterval = min(retryInterval*2, 5*time.Minute)
+						// record the last normalized batch ID even if retry fails to populate metrics consistently
+						a.OtelManager.Metrics.LastNormalizedBatchIdGauge.Record(
+							ctx, lastNormalizedBatchID, metric.WithAttributeSet(attribute.NewSet(
+								attribute.String(otel_metrics.FlowNameKey, config.FlowJobName),
+							)))
+						reqBatchID = normalizeRequests.Load()
+						continue retryLoop
 					}
 				}
-				a.OtelManager.Metrics.LastNormalizedBatchIdGauge.Record(ctx, reqBatchID, metric.WithAttributeSet(attribute.NewSet(
-					attribute.String(otel_metrics.FlowNameKey, config.FlowJobName),
-				)))
-				break
 			}
+			a.OtelManager.Metrics.LastNormalizedBatchIdGauge.Record(ctx, reqBatchID, metric.WithAttributeSet(attribute.NewSet(
+				attribute.String(otel_metrics.FlowNameKey, config.FlowJobName),
+			)))
+			break
 		}
 	}
 }

flow/alerting/classifier.go

@@ -52,8 +52,11 @@ var (
 		`Cannot insert Avro decimal with scale \d+ and precision \d+ to ClickHouse type Decimal\(\d+, \d+\) with scale \d+ and precision \d+`,
 	)
 	// ID(a14c2a1c-edcd-5fcb-73be-bd04e09fccb7) not found in user directories
-	ClickHouseNotFoundInUserDirsRe    = regexp.MustCompile("ID\\([a-z0-9-]+\\) not found in `?user directories`?")
-	ClickHouseTooManyPartsTableRe     = regexp.MustCompile(`in table '(.+)'\.`)
+	ClickHouseNotFoundInUserDirsRe   = regexp.MustCompile("ID\\([a-z0-9-]+\\) not found in `?user directories`?")
+	ClickHouseTooManyPartsTableRe    = regexp.MustCompile(`in table '(.+)'\.`)
+	ClickHouseObjectStorageIOErrorRe = regexp.MustCompile(
+		`unspecified iostream_category error: while reading .+: While executing ReadFromObjectStorage`,
+	)
 	PostgresPublicationDoesNotExistRe = regexp.MustCompile(`publication ".*?" does not exist`)
 	PostgresSnapshotDoesNotExistRe    = regexp.MustCompile(`snapshot ".*?" does not exist`)
 	PostgresWalSegmentRemovedRe       = regexp.MustCompile(`requested WAL segment \w+ has already been removed`)
@@ -982,6 +985,10 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 			chproto.ErrSocketTimeout,
 			chproto.ErrTableIsReadOnly:
 			return ErrorRetryRecoverable, chErrorInfo
+		case chproto.ErrStdException:
+			if ClickHouseObjectStorageIOErrorRe.MatchString(chException.Message) {
+				return ErrorRetryRecoverable, chErrorInfo
+			}
 		case chproto.ErrTimeoutExceeded:
 			if strings.HasSuffix(chException.Message, "distributed_ddl_task_timeout") {
 				return ErrorRetryRecoverable, chErrorInfo

flow/alerting/classifier_test.go

@@ -978,3 +978,18 @@ func TestAuroraFailoverRONodeShouldBeRecoverable(t *testing.T) {
 		Code:   pgerrcode.ObjectNotInPrerequisiteState,
 	}, errInfo)
 }
+
+func TestClickHouseStdExceptionObjectStorageIOErrorShouldBeRecoverable(t *testing.T) {
+	err := &clickhouse.Exception{
+		Code: int32(chproto.ErrStdException),
+		//nolint:lll
+		Message: `std::__1::ios_base::failure: ios_base::clear: unspecified iostream_category error: while reading path/to/file.000000.avro: While executing ReadFromObjectStorage`,
+	}
+	errorClass, errInfo := GetErrorClass(t.Context(),
+		exceptions.NewClickHouseQRepSyncError(fmt.Errorf("failed to sync records: %w", err), "tbl", "db"))
+	assert.Equal(t, ErrorRetryRecoverable, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceClickHouse,
+		Code:   strconv.Itoa(int(chproto.ErrStdException)),
+	}, errInfo)
+}

flow/connectors/core.go

@@ -669,6 +669,8 @@ var (
 
 	_ GetTableSchemaConnector = &connpostgres.PostgresConnector{}
 	_ GetTableSchemaConnector = &connmysql.MySqlConnector{}
+	_ GetTableSchemaConnector = &connmongo.MongoConnector{}
+	_ GetTableSchemaConnector = &connbigquery.BigQueryConnector{}
 	_ GetTableSchemaConnector = &connsnowflake.SnowflakeConnector{}
 	_ GetTableSchemaConnector = &connclickhouse.ClickHouseConnector{}
 
@@ -696,6 +698,7 @@ var (
 	_ QRepSyncConnector = &conns3.S3Connector{}
 	_ QRepSyncConnector = &connclickhouse.ClickHouseConnector{}
 	_ QRepSyncConnector = &connelasticsearch.ElasticsearchConnector{}
+	_ QRepSyncConnector = &connpubsub.PubSubConnector{}
 
 	_ QRepSyncPgConnector = &connpostgres.PostgresConnector{}
 
@@ -721,6 +724,7 @@ var (
 	_ ValidationConnector = &connbigquery.BigQueryConnector{}
 	_ ValidationConnector = &conns3.S3Connector{}
 	_ ValidationConnector = &connmysql.MySqlConnector{}
+	_ ValidationConnector = &connmongo.MongoConnector{}
 
 	_ MirrorSourceValidationConnector = &connpostgres.PostgresConnector{}
 	_ MirrorSourceValidationConnector = &connmysql.MySqlConnector{}
@@ -729,6 +733,7 @@ var (
 
 	_ MirrorDestinationValidationConnector = &connclickhouse.ClickHouseConnector{}
 	_ MirrorDestinationValidationConnector = &connpostgres.PostgresConnector{}
+	_ MirrorDestinationValidationConnector = &connbigquery.BigQueryConnector{}
 
 	_ GetFlagsConnector = &connclickhouse.ClickHouseConnector{}
 

flow/connectors/mongo/codec_test.go

@@ -250,6 +250,11 @@ func TestMarshalDocument(t *testing.T) {
 			input:    bson.D{{Key: "date", Value: bson.DateTime(0)}},
 			expected: `{"date":"1970-01-01T00:00:00Z"}`,
 		},
+		{
+			desc:     "bson.DateTime pre-1970",
+			input:    bson.D{{Key: "date", Value: bson.DateTime(-1)}},
+			expected: `{"date":"1969-12-31T23:59:59.999Z"}`,
+		},
 		{
 			desc:     "bson.DateTime",
 			input:    bson.D{{Key: "date", Value: bson.DateTime(1672531200000)}}, // 2023-01-01 00:00:00 UTC

flow/connectors/mysql/mysql.go

@@ -168,7 +168,9 @@ func (c *MySqlConnector) connect(ctx context.Context) (*client.Conn, error) {
 	if conn == nil {
 		argF := []client.Option{func(conn *client.Conn) error {
 			if c.config.Compression > 0 {
-				conn.SetCapability(mysql.CLIENT_COMPRESS)
+				if err := conn.SetCapability(mysql.CLIENT_COMPRESS); err != nil {
+					return err
+				}
 			}
 			if !c.config.DisableTls {
 				config, err := common.CreateTlsConfig(