From c757e12320394b7b7c531bcf3dad97c7c6ae3474 Mon Sep 17 00:00:00 2001 From: Bilal Akhtar Date: Mon, 31 Aug 2026 14:38:53 -0400 Subject: [PATCH] fix(flow): upgrade to go 1.27 Bumps version of Go up to 1.27. Only changes were some transitive dependency bumps (pebble, swiss) and transposing the `Error()` method on some errors that also have an Unwrap() method so that the pointer type itself also implements the `error` interface, which govet now enforces more strongly when an error is formatted with %w. Other lint changes were removing excess parenthesis and always-nil return values. --- .github/workflows/cleanup.yml | 2 +- .github/workflows/golang-lint.yml | 8 ++++---- .github/workflows/tilt-flow.yml | 2 +- e2e_cleanup/main.go | 2 +- .../clickhouse/staging_validate_test.go | 2 +- flow/connectors/mysql/qvalue_convert.go | 2 +- flow/connectors/postgres/cdc.go | 2 +- flow/connectors/snowflake/snowflake.go | 2 +- flow/connectors/utils/avro_writer.go | 9 +++------ flow/go.mod | 2 +- flow/shared/exceptions/clickhouse.go | 8 ++++++-- flow/shared/exceptions/cockroachdb.go | 18 +++++++++++++----- stacks/flow.Dockerfile | 2 +- 13 files changed, 35 insertions(+), 26 deletions(-) diff --git a/.github/workflows/cleanup.yml b/.github/workflows/cleanup.yml index 9015970d0a..26af03ffb7 100644 --- a/.github/workflows/cleanup.yml +++ b/.github/workflows/cleanup.yml @@ -14,7 +14,7 @@ jobs: - uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 # v6 with: - go-version: '1.26.6' + go-version: '1.27.0' cache-dependency-path: e2e_cleanup/go.sum - name: download go modules diff --git a/.github/workflows/golang-lint.yml b/.github/workflows/golang-lint.yml index c3df637f0e..f5fedd6cd3 100644 --- a/.github/workflows/golang-lint.yml +++ b/.github/workflows/golang-lint.yml @@ -24,7 +24,7 @@ jobs: sudo apt-get install libgeos-dev - uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 # v6 with: - go-version: '1.26.6' + go-version: '1.27.0' cache: false # flow/pkg's Ruleguard config imports flow/generated/protos, so lint needs # both modules in a workspace. e2e_cleanup is linted under it too. @@ -33,18 +33,18 @@ jobs: - name: golangci-lint flow uses: golangci/golangci-lint-action@ba0d7d2ec06a0ea1cb5fa41b2e4a3ab91d21278a # v9 with: - version: v2.12.2 + version: v2.13.2 working-directory: ./flow args: --timeout=10m - name: golangci-lint flow/pkg uses: golangci/golangci-lint-action@ba0d7d2ec06a0ea1cb5fa41b2e4a3ab91d21278a # v9 with: - version: v2.12.2 + version: v2.13.2 working-directory: ./flow/pkg args: --config=../.golangci.yml --timeout=10m - name: golangci-lint e2e_cleanup uses: golangci/golangci-lint-action@ba0d7d2ec06a0ea1cb5fa41b2e4a3ab91d21278a # v9 with: - version: v2.12.2 + version: v2.13.2 working-directory: ./e2e_cleanup args: --timeout=10m diff --git a/.github/workflows/tilt-flow.yml b/.github/workflows/tilt-flow.yml index b8b79eaa99..43aef0150b 100644 --- a/.github/workflows/tilt-flow.yml +++ b/.github/workflows/tilt-flow.yml @@ -232,7 +232,7 @@ jobs: - uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 # v6 with: - go-version: '1.26.6' + go-version: '1.27.0' cache-dependency-path: | flow/go.sum flow/pkg/go.sum diff --git a/e2e_cleanup/main.go b/e2e_cleanup/main.go index affba662c9..8c8e0a1e29 100644 --- a/e2e_cleanup/main.go +++ b/e2e_cleanup/main.go @@ -188,7 +188,7 @@ func CleanupSF(ctx context.Context) { Database: config.Database, Warehouse: config.Warehouse, Role: config.Role, - RequestTimeout: time.Minute, + RequestTimeout: time.Minute, //nolint:staticcheck // deprecated upstream, no replacement Params: map[string]*string{ "CLIENT_TELEMETRY_ENABLED": new("false"), }, diff --git a/flow/connectors/clickhouse/staging_validate_test.go b/flow/connectors/clickhouse/staging_validate_test.go index be68b040d8..c641c4fdba 100644 --- a/flow/connectors/clickhouse/staging_validate_test.go +++ b/flow/connectors/clickhouse/staging_validate_test.go @@ -161,7 +161,7 @@ func TestGCSStagingStoreValidate_HappyPath(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { t.Logf("GCS request: %s %s", r.Method, r.URL.Path) switch { - case r.Method == http.MethodPost && (strings.Contains(r.URL.Path, "/b/my-bucket/o")): + case r.Method == http.MethodPost && strings.Contains(r.URL.Path, "/b/my-bucket/o"): uploads.Add(1) w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) diff --git a/flow/connectors/mysql/qvalue_convert.go b/flow/connectors/mysql/qvalue_convert.go index 42949bfd05..153ac5c801 100644 --- a/flow/connectors/mysql/qvalue_convert.go +++ b/flow/connectors/mysql/qvalue_convert.go @@ -428,7 +428,7 @@ func binaryColumnLength(mytype byte, meta uint16) int { borrowedBitsMask := uint8(0x30) extraBits := higherMetaByte & borrowedBitsMask if extraBits != borrowedBitsMask { // More than 255 bytes - return int(lowerMetaByte) | (int((extraBits)^borrowedBitsMask) << 4) + return int(lowerMetaByte) | (int(extraBits^borrowedBitsMask) << 4) } return int(meta & 0xFF) } diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go index 75d474b00c..6432596cf5 100644 --- a/flow/connectors/postgres/cdc.go +++ b/flow/connectors/postgres/cdc.go @@ -752,7 +752,7 @@ func PullCdcRecords[Items model.Items]( waitingForCommit = true } } else { - logger.Info(("standby deadline reached, no records accumulated, continuing to wait")) + logger.Info("standby deadline reached, no records accumulated, continuing to wait") } nextRecordDeadline = time.Now().Add(req.IdleTimeout) } diff --git a/flow/connectors/snowflake/snowflake.go b/flow/connectors/snowflake/snowflake.go index 8963003da0..5b2c1ae3cc 100644 --- a/flow/connectors/snowflake/snowflake.go +++ b/flow/connectors/snowflake/snowflake.go @@ -98,7 +98,7 @@ func NewSnowflakeConnector( Database: snowflakeProtoConfig.Database, Warehouse: snowflakeProtoConfig.Warehouse, Role: snowflakeProtoConfig.Role, - RequestTimeout: time.Duration(snowflakeProtoConfig.QueryTimeout), + RequestTimeout: time.Duration(snowflakeProtoConfig.QueryTimeout), //nolint:staticcheck // deprecated upstream, no replacement Params: map[string]*string{ "CLIENT_SESSION_KEEP_ALIVE": new("true"), "CLIENT_TELEMETRY_ENABLED": new("false"), diff --git a/flow/connectors/utils/avro_writer.go b/flow/connectors/utils/avro_writer.go index f2ed2b8d43..e680c30d68 100644 --- a/flow/connectors/utils/avro_writer.go +++ b/flow/connectors/utils/avro_writer.go @@ -194,13 +194,13 @@ func (p *peerDBOCFWriter) WriteRecordsToAvroFile(ctx context.Context, env map[st }, nil } -func (p *peerDBOCFWriter) getAvroFieldNamesFromSchema() ([]string, error) { +func (p *peerDBOCFWriter) getAvroFieldNamesFromSchema() []string { fields := p.avroSchema.Schema.Fields() avroFieldNames := make([]string, len(fields)) for i, field := range fields { avroFieldNames[i] = field.Name() } - return avroFieldNames, nil + return avroFieldNames } func (p *peerDBOCFWriter) writeRecordsToOCFWriter( @@ -212,10 +212,7 @@ func (p *peerDBOCFWriter) writeRecordsToOCFWriter( ) (int64, error) { logger := internal.LoggerFromCtx(ctx) - avroFieldNames, err := p.getAvroFieldNamesFromSchema() - if err != nil { - return 0, fmt.Errorf("failed to get Avro field names from schema: %w", err) - } + avroFieldNames := p.getAvroFieldNamesFromSchema() avroConverter, err := model.NewQRecordAvroConverter( ctx, env, p.avroSchema, p.targetDWH, avroFieldNames, logger, ) diff --git a/flow/go.mod b/flow/go.mod index d2f9b576ec..e3d37ddf30 100644 --- a/flow/go.mod +++ b/flow/go.mod @@ -1,6 +1,6 @@ module github.com/PeerDB-io/peerdb/flow -go 1.26.0 +go 1.27.0 require ( cloud.google.com/go v0.123.0 diff --git a/flow/shared/exceptions/clickhouse.go b/flow/shared/exceptions/clickhouse.go index 66a058cac9..29e832f511 100644 --- a/flow/shared/exceptions/clickhouse.go +++ b/flow/shared/exceptions/clickhouse.go @@ -19,7 +19,7 @@ func (e *ClickHouseQRepSyncError) Unwrap() error { } type ClickHouseNormalizedTableCreationError struct { - error + err error DestinationTable string } @@ -27,6 +27,10 @@ func NewClickHouseNormalizedTableCreationError(err error, destinationTable strin return &ClickHouseNormalizedTableCreationError{err, destinationTable} } +func (e *ClickHouseNormalizedTableCreationError) Error() string { + return e.err.Error() +} + func (e *ClickHouseNormalizedTableCreationError) Unwrap() error { - return e.error + return e.err } diff --git a/flow/shared/exceptions/cockroachdb.go b/flow/shared/exceptions/cockroachdb.go index 4314b49e5c..21040284f5 100644 --- a/flow/shared/exceptions/cockroachdb.go +++ b/flow/shared/exceptions/cockroachdb.go @@ -4,15 +4,19 @@ package exceptions // speaks the Postgres wire protocol and reuses its SQLSTATE codes, so without // this wrapper error classification would attribute CockroachDB errors to Postgres. type CockroachDBError struct { - error + err error } func NewCockroachDBError(err error) *CockroachDBError { return &CockroachDBError{err} } +func (e *CockroachDBError) Error() string { + return e.err.Error() +} + func (e *CockroachDBError) Unwrap() error { - return e.error + return e.err } // CockroachChangefeedIrrecoverableError marks changefeed failures no retry can @@ -20,16 +24,20 @@ func (e *CockroachDBError) Unwrap() error { // truncated or dropped. The mirror needs operator action, typically a resync, // and the alerting classifier notifies the user instead of retrying silently. type CockroachChangefeedIrrecoverableError struct { - error + err error // Code is a stable machine-readable reason: CURSOR_PAST_GC, // TABLE_TRUNCATED, TABLE_DROPPED or TABLE_NOT_AT_CURSOR. Code string } func NewCockroachChangefeedIrrecoverableError(code string, err error) *CockroachChangefeedIrrecoverableError { - return &CockroachChangefeedIrrecoverableError{error: err, Code: code} + return &CockroachChangefeedIrrecoverableError{err: err, Code: code} +} + +func (e *CockroachChangefeedIrrecoverableError) Error() string { + return e.err.Error() } func (e *CockroachChangefeedIrrecoverableError) Unwrap() error { - return e.error + return e.err } diff --git a/stacks/flow.Dockerfile b/stacks/flow.Dockerfile index f76b27233d..fc76e91351 100644 --- a/stacks/flow.Dockerfile +++ b/stacks/flow.Dockerfile @@ -1,6 +1,6 @@ # syntax=docker/dockerfile:1.26@sha256:ecfaec9ed6d810b56388c508f4121597bfbba70d41a6dfeee4d8cad5f295fc32 -FROM golang:1.26-alpine@sha256:0178a641fbb4858c5f1b48e34bdaabe0350a330a1b1149aabd498d0699ff5fb2 AS builder +FROM golang:1.27-alpine@sha256:4c9fe60190a2a3350ddc51de80d0224b8a6698d12bdfc999fee45ea9d6c46dbc AS builder # Allow build flags to be passed in at build time, for example debug flags ARG DEBUG_BUILD ENV DEBUG_BUILD=${DEBUG_BUILD}