fix(flow): upgrade to go 1.27

.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

.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

.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

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"),
 		},

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)

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)
 }

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)
 		}

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"),

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,
 	)

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

flow/shared/exceptions/clickhouse.go

@@ -19,14 +19,18 @@ func (e *ClickHouseQRepSyncError) Unwrap() error {
 }
 
 type ClickHouseNormalizedTableCreationError struct {
-	error
+	err              error
 	DestinationTable string
 }
 
 func NewClickHouseNormalizedTableCreationError(err error, destinationTable string) *ClickHouseNormalizedTableCreationError {
 	return &ClickHouseNormalizedTableCreationError{err, destinationTable}
 }
 
+func (e *ClickHouseNormalizedTableCreationError) Error() string {
+	return e.err.Error()
+}
+
 func (e *ClickHouseNormalizedTableCreationError) Unwrap() error {
-	return e.error
+	return e.err
 }

flow/shared/exceptions/cockroachdb.go

@@ -4,32 +4,40 @@ 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
 // fix: the cursor fell behind the replica GC threshold, or a watched table was
 // 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
 }

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}