Update to Go 1.26, golangci-lint 2.10.1

.github/workflows/golang-lint.yml

@@ -29,12 +29,12 @@ jobs:
       - name: golangci-lint flow
         uses: golangci/golangci-lint-action@1e7e51e771db61008b38414a730f564565cf7c20 # v9
         with:
-          version: v2.9.0
+          version: v2.10.1
           working-directory: ./flow
           args: --timeout=10m
       - name: golangci-lint e2e_cleanup
         uses: golangci/golangci-lint-action@1e7e51e771db61008b38414a730f564565cf7c20 # v9
         with:
-          version: v2.9.0
+          version: v2.10.1
           working-directory: ./e2e_cleanup
           args: --timeout=10m

e2e_cleanup/go.mod

@@ -1,6 +1,6 @@
 module github.com/PeerDB-io/peerdb/e2e_cleanup
 
-go 1.25.0
+go 1.26.0
 
 require (
 	cloud.google.com/go/bigquery v1.72.0

flow/activities/flowable.go

@@ -305,7 +305,7 @@ func (a *FlowableActivity) SyncFlow(
 	var normalizeWaiting atomic.Bool
 	var syncingBatchID atomic.Int64
 	var syncState atomic.Pointer[string]
-	syncState.Store(shared.Ptr("setup"))
+	syncState.Store(new("setup"))
 	shutdown := common.HeartbeatRoutine(ctx, func() string {
 		// Must load Waiting after BatchID to avoid race saying we're waiting on currently processing batch
 		sBatchID := syncingBatchID.Load()
@@ -399,7 +399,7 @@ func (a *FlowableActivity) SyncFlow(
 				break
 			}
 			logger.Error("failed to sync records", slog.Any("error", syncErr))
-			syncState.Store(shared.Ptr("cleanup"))
+			syncState.Store(new("cleanup"))
 			close(syncDone)
 			normRequests.Close()
 			normResponses.Close()
@@ -418,7 +418,7 @@ func (a *FlowableActivity) SyncFlow(
 		}
 	}
 
-	syncState.Store(shared.Ptr("cleanup"))
+	syncState.Store(new("cleanup"))
 	close(syncDone)
 	normRequests.Close()
 	normResponses.Close()

flow/activities/flowable_core.go

@@ -194,7 +194,7 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 	}
 
 	startTime := time.Now()
-	syncState.Store(shared.Ptr("syncing"))
+	syncState.Store(new("syncing"))
 	errGroup, errCtx := errgroup.WithContext(ctx)
 	errGroup.Go(func() error {
 		return pull(srcConn, errCtx, a.CatalogPool, a.OtelManager, &model.PullRecordsRequest[Items]{
@@ -235,7 +235,7 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 		}
 		defer dstClose(ctx)
 
-		syncState.Store(shared.Ptr("updating schema"))
+		syncState.Store(new("updating schema"))
 		if err := dstConn.ReplayTableSchemaDeltas(
 			ctx, config.Env, flowName, options.TableMappings, recordBatchSync.SchemaDeltas, config.Flags,
 		); err != nil {
@@ -319,7 +319,7 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 		}
 		return nil, fmt.Errorf("[cdc] failed to pull records: %w", err)
 	}
-	syncState.Store(shared.Ptr("bookkeeping"))
+	syncState.Store(new("bookkeeping"))
 
 	syncDuration := time.Since(syncStartTime)
 	lastCheckpoint := recordBatchSync.GetLastCheckpoint()
@@ -350,13 +350,13 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 
 	a.OtelManager.Metrics.CurrentBatchIdGauge.Record(ctx, res.CurrentSyncBatchID)
 
-	syncState.Store(shared.Ptr("updating schema"))
+	syncState.Store(new("updating schema"))
 	if err := a.applySchemaDeltas(ctx, config, res.TableSchemaDeltas); err != nil {
 		return nil, err
 	}
 
 	if recordBatchSync.NeedsNormalize() {
-		syncState.Store(shared.Ptr("normalizing"))
+		syncState.Store(new("normalizing"))
 		normRequests.Update(res.CurrentSyncBatchID)
 		normWaitThreshold := res.CurrentSyncBatchID - normBufferSize
 		if normResponses.Load() <= normWaitThreshold {

flow/alerting/slack_alert_sender.go

@@ -26,7 +26,7 @@ func (s *SlackAlertSender) getOpenConnectionsAlertThreshold() uint32 {
 }
 
 type slackAlertConfig struct {
-	AuthToken                     string   `json:"auth_token"`
+	AuthToken                     string   `json:"auth_token"` //nolint:gosec // G117: credential field by design, encrypted outside
 	ChannelIDs                    []string `json:"channel_ids"`
 	Members                       []string `json:"members"`
 	SlotLagMBAlertThreshold       uint32   `json:"slot_lag_mb_alert_threshold"`

flow/cmd/maintenance.go

@@ -9,7 +9,6 @@ import (
 	"os"
 	"strings"
 
-	"github.com/aws/smithy-go/ptr"
 	"go.temporal.io/sdk/client"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -73,7 +72,7 @@ func MaintenanceMain(ctx context.Context, args *MaintenanceCLIParams) error {
 			slog.InfoContext(ctx, "Assuming maintenance workflows were skipped")
 			return WriteMaintenanceOutputToCatalog(ctx, StartMaintenanceResult{
 				Skipped:       true,
-				SkippedReason: ptr.String("Assumed skipped by CLI Flag"),
+				SkippedReason: new("Assumed skipped by CLI Flag"),
 				CLIVersion:    internal.PeerDBVersionShaShort(),
 			})
 		}
@@ -143,7 +142,7 @@ func skipStartMaintenanceIfNeeded(ctx context.Context, args *MaintenanceCLIParam
 			slog.InfoContext(ctx, "Skipping maintenance workflow due to missing k8s service", "service", args.SkipIfK8sServiceMissing)
 			return true, WriteMaintenanceOutputToCatalog(ctx, StartMaintenanceResult{
 				Skipped:          true,
-				SkippedReason:    ptr.String(fmt.Sprintf("K8s service %s missing", args.SkipIfK8sServiceMissing)),
+				SkippedReason:    new(fmt.Sprintf("K8s service %s missing", args.SkipIfK8sServiceMissing)),
 				CLIVersion:       internal.PeerDBVersionShaShort(),
 				CLIDeployVersion: internal.PeerDBDeploymentVersion(),
 			})
@@ -189,11 +188,11 @@ func skipStartMaintenanceIfNeeded(ctx context.Context, args *MaintenanceCLIParam
 					"deployApiVersion", version.DeploymentVersion, "cliDeployVersion", internal.PeerDBDeploymentVersion())
 				return true, WriteMaintenanceOutputToCatalog(ctx, StartMaintenanceResult{
 					Skipped:          true,
-					SkippedReason:    ptr.String("Version Mismatch: " + strings.Join(skippedReasons, ", ")),
+					SkippedReason:    new("Version Mismatch: " + strings.Join(skippedReasons, ", ")),
 					CLIVersion:       internal.PeerDBVersionShaShort(),
 					CLIDeployVersion: internal.PeerDBDeploymentVersion(),
 					APIVersion:       version.Version,
-					APIDeployVersion: ptr.ToString(version.DeploymentVersion),
+					APIDeployVersion: shared.Val(version.DeploymentVersion),
 				})
 			}
 		}
@@ -208,7 +207,7 @@ func skipStartMaintenanceIfNeeded(ctx context.Context, args *MaintenanceCLIParam
 				slog.InfoContext(ctx, "Skipping maintenance workflow due to no mirrors")
 				return true, WriteMaintenanceOutputToCatalog(ctx, StartMaintenanceResult{
 					Skipped:       true,
-					SkippedReason: ptr.String("No mirrors found"),
+					SkippedReason: new("No mirrors found"),
 				})
 			}
 		}

flow/cmd/settings.go

@@ -35,8 +35,7 @@ func (h *FlowRequestHandler) GetDynamicSettings(
 	if _, err := pgx.ForEachRow(rows, []any{&name, &value}, func() error {
 		if idx, ok := internal.DynamicIndex[name]; ok {
 			settings[idx] = proto.CloneOf(settings[idx])
-			newValue := value // create a new string reference as value can be overwritten by the next iteration.
-			settings[idx].Value = &newValue
+			settings[idx].Value = new(value) // create a new string reference as value can be overwritten by the next iteration.
 		}
 		return nil
 	}); err != nil {

flow/connectors/mysql/rds_test.go

@@ -4,7 +4,6 @@ import (
 	"os"
 	"testing"
 
-	"github.com/aws/smithy-go/ptr"
 	"github.com/stretchr/testify/require"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
@@ -29,7 +28,7 @@ func TestAwsRDSIAMAuthConnectForMYSQL(t *testing.T) {
 				AuthConfig: &protos.AwsAuthenticationConfig_Role{
 					Role: &protos.AWSAuthAssumeRoleConfig{
 						AssumeRoleArn:  os.Getenv("FLOW_TESTS_RDS_IAM_AUTH_ASSUME_ROLE"),
-						ChainedRoleArn: ptr.String(os.Getenv("FLOW_TESTS_RDS_IAM_AUTH_CHAINED_ROLE")),
+						ChainedRoleArn: new(os.Getenv("FLOW_TESTS_RDS_IAM_AUTH_CHAINED_ROLE")),
 					},
 				},
 			},
@@ -60,7 +59,7 @@ func TestAwsRDSIAMAuthConnectForMYSQLViaProxy(t *testing.T) {
 				AuthConfig: &protos.AwsAuthenticationConfig_Role{
 					Role: &protos.AWSAuthAssumeRoleConfig{
 						AssumeRoleArn:  os.Getenv("FLOW_TESTS_RDS_IAM_AUTH_ASSUME_ROLE"),
-						ChainedRoleArn: ptr.String(os.Getenv("FLOW_TESTS_RDS_IAM_AUTH_CHAINED_ROLE")),
+						ChainedRoleArn: new(os.Getenv("FLOW_TESTS_RDS_IAM_AUTH_CHAINED_ROLE")),
 					},
 				},
 			},

flow/connectors/postgres/rds_test.go

@@ -4,7 +4,6 @@ import (
 	"os"
 	"testing"
 
-	"github.com/aws/smithy-go/ptr"
 	"github.com/stretchr/testify/require"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
@@ -31,7 +30,7 @@ func TestAwsRDSIAMAuthConnectForPostgres(t *testing.T) {
 				AuthConfig: &protos.AwsAuthenticationConfig_Role{
 					Role: &protos.AWSAuthAssumeRoleConfig{
 						AssumeRoleArn:  os.Getenv("FLOW_TESTS_RDS_IAM_AUTH_ASSUME_ROLE"),
-						ChainedRoleArn: ptr.String(os.Getenv("FLOW_TESTS_RDS_IAM_AUTH_CHAINED_ROLE")),
+						ChainedRoleArn: new(os.Getenv("FLOW_TESTS_RDS_IAM_AUTH_CHAINED_ROLE")),
 					},
 				},
 			},
@@ -74,7 +73,7 @@ func TestAwsRDSIAMAuthConnectForPostgresViaProxy(t *testing.T) {
 				AuthConfig: &protos.AwsAuthenticationConfig_Role{
 					Role: &protos.AWSAuthAssumeRoleConfig{
 						AssumeRoleArn:  os.Getenv("FLOW_TESTS_RDS_IAM_AUTH_ASSUME_ROLE"),
-						ChainedRoleArn: ptr.String(os.Getenv("FLOW_TESTS_RDS_IAM_AUTH_CHAINED_ROLE")),
+						ChainedRoleArn: new(os.Getenv("FLOW_TESTS_RDS_IAM_AUTH_CHAINED_ROLE")),
 					},
 				},
 			},

flow/connectors/snowflake/snowflake.go

@@ -11,7 +11,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/aws/smithy-go/ptr"
 	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/snowflakedb/gosnowflake"
 	"go.temporal.io/sdk/log"
@@ -92,7 +91,7 @@ func NewSnowflakeConnector(
 	}
 
 	additionalParams := make(map[string]*string)
-	additionalParams["CLIENT_SESSION_KEEP_ALIVE"] = ptr.String("true")
+	additionalParams["CLIENT_SESSION_KEEP_ALIVE"] = new("true")
 
 	snowflakeConfig := gosnowflake.Config{
 		Account:          snowflakeProtoConfig.AccountId,

flow/connectors/utils/aws.go

@@ -19,7 +19,6 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/sts"
 	smithyendpoints "github.com/aws/smithy-go/endpoints"
-	"github.com/aws/smithy-go/ptr"
 	"github.com/google/uuid"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
@@ -33,15 +32,6 @@ const (
 
 var s3CompatibleServiceEndpointPattern = regexp.MustCompile(`^https?://[a-zA-Z0-9.-]+(:\d+)?$`)
 
-type AWSSecrets struct {
-	AccessKeyID     string
-	SecretAccessKey string
-	AwsRoleArn      string
-	Region          string
-	Endpoint        string
-	SessionToken    string
-}
-
 type PeerAWSCredentials struct {
 	Credentials    aws.Credentials
 	RoleArn        *string
@@ -199,7 +189,7 @@ func (a *AssumeRoleBasedAWSCredentialsProvider) Retrieve(ctx context.Context) (A
 	}
 	return AWSCredentials{
 		AWS:         retrieved,
-		EndpointUrl: ptr.String(a.GetEndpointURL()),
+		EndpointUrl: new(a.GetEndpointURL()),
 	}, nil
 }
 

flow/connectors/utils/gcp.go

@@ -18,7 +18,7 @@ type GcpServiceAccount struct {
 	Type                    string `json:"type"`
 	ProjectID               string `json:"project_id"`
 	PrivateKeyID            string `json:"private_key_id"`
-	PrivateKey              string `json:"private_key"`
+	PrivateKey              string `json:"private_key"` //nolint:gosec // G117: deserialized from Google credential files, by design
 	ClientEmail             string `json:"client_email"`
 	ClientID                string `json:"client_id"`
 	AuthURI                 string `json:"auth_uri"`

flow/e2e/clickhouse_mysql_test.go

@@ -560,7 +560,6 @@ func (s ClickHouseSuite) Test_MySQL_Specific_Geometric_Types() {
 		require.Len(s.t, row, 8, "expected 8 columns")
 		for j := 1; j < 8; j++ {
 			geometryVal := row[j].Value()
-			//nolint:gosec
 			require.Equal(s.t, expectedValues[i][j-1], geometryVal, "geometry value mismatch at row %d column %d", i+1, j)
 		}
 	}

flow/e2e/congen.go

@@ -30,7 +30,7 @@ func TableMappings(s GenericSuite, tables ...string) []*protos.TableMapping {
 	for i := 0; i < len(tables); i += 2 {
 		tm = append(tm, &protos.TableMapping{
 			SourceTableIdentifier:      AttachSchema(s, tables[i]),
-			DestinationTableIdentifier: s.DestinationTable(tables[i+1]),
+			DestinationTableIdentifier: s.DestinationTable(tables[i+1]), //nolint:gosec // G602: even length enforced above
 			ShardingKey:                "id",
 		})
 	}

flow/e2e/s3_helper.go

@@ -30,7 +30,7 @@ type S3PeerCredentials struct {
 	AccessKeyID     string `json:"accessKeyId"`
 	SecretAccessKey string `json:"secretAccessKey"`
 	AwsRoleArn      string `json:"awsRoleArn"`
-	SessionToken    string `json:"sessionToken"`
+	SessionToken    string `json:"sessionToken"` //nolint:gosec // G117: sometimes provided from file
 	Region          string `json:"region"`
 	Endpoint        string `json:"endpoint"`
 }
@@ -73,7 +73,7 @@ func NewS3TestHelper(ctx context.Context, s3environment S3Environment) (*S3TestH
 		if err != nil {
 			return nil, err
 		}
-		rootCA = shared.Ptr(base64.StdEncoding.EncodeToString(bytes))
+		rootCA = new(base64.StdEncoding.EncodeToString(bytes))
 		tlsHost = "minio.local"
 	default:
 		panic(fmt.Sprintf("invalid s3environment %d", s3environment))
@@ -97,7 +97,7 @@ func NewS3TestHelper(ctx context.Context, s3environment S3Environment) (*S3TestH
 		AccessKeyId:     &config.AccessKeyID,
 		SecretAccessKey: &config.SecretAccessKey,
 		Region:          &config.Region,
-		Endpoint:        shared.Ptr(endpoint),
+		Endpoint:        new(endpoint),
 		RootCa:          rootCA,
 		TlsHost:         tlsHost,
 	}

flow/go.mod

@@ -1,6 +1,6 @@
 module github.com/PeerDB-io/peerdb/flow
 
-go 1.25.5
+go 1.26.0
 
 require (
 	cloud.google.com/go v0.123.0

flow/internal/dynamicconf.go

@@ -10,7 +10,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/aws/smithy-go/ptr"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgtype"
 	"golang.org/x/exp/constraints"
@@ -725,7 +724,7 @@ func PeerDBMaintenanceModeEnabled(ctx context.Context, env map[string]string) (b
 }
 
 func UpdatePeerDBMaintenanceModeEnabled(ctx context.Context, pool shared.CatalogPool, enabled bool) error {
-	return UpdateDynamicSetting(ctx, pool, "PEERDB_MAINTENANCE_MODE_ENABLED", ptr.String(strconv.FormatBool(enabled)))
+	return UpdateDynamicSetting(ctx, pool, "PEERDB_MAINTENANCE_MODE_ENABLED", new(strconv.FormatBool(enabled)))
 }
 
 func PeerDBPKMEmptyBatchThrottleThresholdSeconds(ctx context.Context, env map[string]string) (int64, error) {

flow/model/qrecord_avro_size_test.go

@@ -75,7 +75,7 @@ func TestMongoDBAvroSizeComputation(t *testing.T) {
 				"Estimated size should be within 5% of actual (upper bound)")
 
 			// for completion, print out the compression ratio
-			fileInfo, err := os.Stat(tmpfile.Name())
+			fileInfo, err := os.Stat(tmpfile.Name()) //nolint:gosec // G703: temp file path
 			require.NoError(t, err)
 			compressedSize := fileInfo.Size()
 			compressionRatio := float64(actualSize) / float64(compressedSize)
@@ -106,7 +106,7 @@ func writeAvroFileCompressed(
 		ConstructColumnNameAvroFieldMap(schema.Fields),
 	)
 	require.NoError(t, err)
-	file, err := os.Create(filePath)
+	file, err := os.Create(filePath) //nolint:gosec // G703: temp file path
 	require.NoError(t, err)
 	defer file.Close()
 
@@ -233,7 +233,7 @@ func (w *MeteredWriter) Write(p []byte) (int, error) {
 func getActualUncompressedSize(t *testing.T, filePath string, avroSchema *QRecordAvroSchemaDefinition, numRecords int) int64 {
 	t.Helper()
 
-	file, err := os.Open(filePath)
+	file, err := os.Open(filePath) //nolint:gosec // G703: temp file path
 	require.NoError(t, err)
 	defer file.Close()
 	decoder, err := ocf.NewDecoder(file)

flow/pkg/go.mod

@@ -1,6 +1,6 @@
 module github.com/PeerDB-io/peerdb/flow/pkg
 
-go 1.25.0
+go 1.26.0
 
 require (
 	github.com/ClickHouse/ch-go v0.68.0

flow/pkg/mongo/client_options_test.go

@@ -5,9 +5,10 @@ import (
 	"testing"
 	"time"
 
-	"github.com/PeerDB-io/peerdb/flow/pkg/common"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/PeerDB-io/peerdb/flow/pkg/common"
 )
 
 func TestBuildClientOptions_TLS(t *testing.T) {

flow/pkg/mongo/validation.go

@@ -7,8 +7,9 @@ import (
 	"slices"
 	"strings"
 
-	"github.com/PeerDB-io/peerdb/flow/pkg/common"
 	"go.mongodb.org/mongo-driver/v2/mongo"
+
+	"github.com/PeerDB-io/peerdb/flow/pkg/common"
 )
 
 const (