Timezone cleanup (#3202)

1. if-err cleanup
2. don't wait on sync to cancel when pausing, this was causing pausing
to sometimes take 10+ seconds
3. set timezone UTC on postgres connections
4. parse datetime as UTC in ClickHouse

4 addresses this:
```
select parseDateTimeBestEffort('2025-07-11 19:30:02.313462+0000')

SELECT parseDateTimeBestEffort('2025-07-11 19:30:02.313462+0000')

Query id: 0d4bbad2-fdab-414f-85c8-2c80c61ad0ab

┌─parseDateTimeBestEffort('2025-07-11 19:30:02.313462+0000')─┐
│                                        2025-07-11 15:30:02 │
└────────────────────────────────────────────────────────────┘
```
where the ClickHouse server has a non-UTC timezone.
The problem is our timezone hygiene isn't clear enough bringing data from postgres &
ingesting into ClickHouse to handle this, between initial loads & cdc,
that we should instead maintain a policy of "peerdb transmits times as UTC"

Author: serprex

flow/connectors/bigquery/bigquery.go

@@ -83,10 +83,9 @@ func NewBigQueryConnector(ctx context.Context, config *protos.BigqueryConfig) (*
 		return nil, fmt.Errorf("failed to create BigQuery client: %v", err)
 	}
 
-	_, datasetErr := client.DatasetInProject(projectID, datasetID).Metadata(ctx)
-	if datasetErr != nil {
-		logger.Error("failed to get dataset metadata", "error", datasetErr)
-		return nil, fmt.Errorf("failed to get dataset metadata: %v", datasetErr)
+	if _, err := client.DatasetInProject(projectID, datasetID).Metadata(ctx); err != nil {
+		logger.Error("failed to get dataset metadata", "error", err)
+		return nil, fmt.Errorf("failed to get dataset metadata: %v", err)
 	}
 
 	storageClient, err := bqsa.CreateStorageClient(ctx)
@@ -166,8 +165,7 @@ func (c *BigQueryConnector) Close() error {
 
 // ConnectionActive returns nil if the connection is active.
 func (c *BigQueryConnector) ConnectionActive(ctx context.Context) error {
-	_, err := c.client.DatasetInProject(c.projectID, c.datasetID).Metadata(ctx)
-	if err != nil {
+	if _, err := c.client.DatasetInProject(c.projectID, c.datasetID).Metadata(ctx); err != nil {
 		return fmt.Errorf("failed to get dataset metadata: %v", err)
 	}
 
@@ -182,12 +180,13 @@ func (c *BigQueryConnector) waitForTableReady(ctx context.Context, datasetTable
 	attempt := 0
 
 	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
 		if time.Now().After(deadline) {
 			return fmt.Errorf("timeout reached while waiting for table %s to be ready", datasetTable)
 		}
-
-		_, err := table.Metadata(ctx)
-		if err == nil {
+		if _, err := table.Metadata(ctx); err == nil {
 			return nil
 		}
 
@@ -768,8 +767,7 @@ func (c *BigQueryConnector) RenameTables(
 
 		// if _resync table does not exist, log and continue.
 		dataset := c.client.DatasetInProject(c.projectID, srcDatasetTable.dataset)
-		_, err := dataset.Table(srcDatasetTable.table).Metadata(ctx)
-		if err != nil {
+		if _, err := dataset.Table(srcDatasetTable.table).Metadata(ctx); err != nil {
 			if !strings.Contains(err.Error(), "notFound") {
 				return nil, fmt.Errorf("[renameTable] failed to get metadata for _resync table %s: %w", srcDatasetTable.string(), err)
 			}
@@ -779,8 +777,7 @@ func (c *BigQueryConnector) RenameTables(
 
 		// if the original table does not exist, log and skip soft delete step
 		originalTableExists := true
-		_, err = dataset.Table(dstDatasetTable.table).Metadata(ctx)
-		if err != nil {
+		if _, err := dataset.Table(dstDatasetTable.table).Metadata(ctx); err != nil {
 			if !strings.Contains(err.Error(), "notFound") {
 				return nil, fmt.Errorf("[renameTable] failed to get metadata for original table %s: %w", dstDatasetTable.string(), err)
 			}
@@ -860,8 +857,7 @@ func (c *BigQueryConnector) RenameTables(
 
 				query.DefaultProjectID = c.projectID
 				query.DefaultDatasetID = c.datasetID
-				_, err := query.Read(ctx)
-				if err != nil {
+				if _, err := query.Read(ctx); err != nil {
 					return nil, fmt.Errorf("unable to handle soft-deletes for table %s: %w", dstDatasetTable.string(), err)
 				}
 			}
@@ -876,8 +872,7 @@ func (c *BigQueryConnector) RenameTables(
 
 			query.DefaultProjectID = c.projectID
 			query.DefaultDatasetID = c.datasetID
-			_, err := query.Read(ctx)
-			if err != nil {
+			if _, err := query.Read(ctx); err != nil {
 				return nil, fmt.Errorf("unable to set synced at column for table %s: %w", srcDatasetTable.string(), err)
 			}
 		}
@@ -886,8 +881,7 @@ func (c *BigQueryConnector) RenameTables(
 		dropQuery := c.queryWithLogging("DROP TABLE IF EXISTS " + dstDatasetTable.string())
 		dropQuery.DefaultProjectID = c.projectID
 		dropQuery.DefaultDatasetID = c.datasetID
-		_, err = dropQuery.Read(ctx)
-		if err != nil {
+		if _, err := dropQuery.Read(ctx); err != nil {
 			return nil, fmt.Errorf("unable to drop table %s: %w", dstDatasetTable.string(), err)
 		}
 
@@ -896,8 +890,7 @@ func (c *BigQueryConnector) RenameTables(
 			srcDatasetTable.string(), dstDatasetTable.table))
 		query.DefaultProjectID = c.projectID
 		query.DefaultDatasetID = c.datasetID
-		_, err = query.Read(ctx)
-		if err != nil {
+		if _, err := query.Read(ctx); err != nil {
 			return nil, fmt.Errorf("unable to rename table %s to %s: %w", srcDatasetTable.string(),
 				dstDatasetTable.string(), err)
 		}
@@ -925,8 +918,7 @@ func (c *BigQueryConnector) CreateTablesFromExisting(
 			newDatasetTable.string(), existingDatasetTable.string()))
 		query.DefaultProjectID = c.projectID
 		query.DefaultDatasetID = c.datasetID
-		_, err := query.Read(ctx)
-		if err != nil {
+		if _, err := query.Read(ctx); err != nil {
 			return nil, fmt.Errorf("unable to create table %s: %w", newTable, err)
 		}
 
@@ -950,8 +942,7 @@ func (c *BigQueryConnector) RemoveTableEntriesFromRawTable(
 			rawTableIdentifier, tableName, req.NormalizeBatchId, req.SyncBatchId))
 		deleteCmd.DefaultProjectID = c.projectID
 		deleteCmd.DefaultDatasetID = c.datasetID
-		_, err := deleteCmd.Read(ctx)
-		if err != nil {
+		if _, err := deleteCmd.Read(ctx); err != nil {
 			c.logger.Error("failed to remove entries from raw table", "error", err)
 		}
 

flow/connectors/bigquery/qrep.go

@@ -103,8 +103,7 @@ func (c *BigQueryConnector) SetupQRepMetadataTables(ctx context.Context, config
 		query := c.queryWithLogging("TRUNCATE TABLE " + config.DestinationTableIdentifier)
 		query.DefaultDatasetID = c.datasetID
 		query.DefaultProjectID = c.projectID
-		_, err := query.Read(ctx)
-		if err != nil {
+		if _, err := query.Read(ctx); err != nil {
 			return fmt.Errorf("failed to TRUNCATE table before query replication: %w", err)
 		}
 	}

flow/connectors/clickhouse/normalize_query.go

@@ -128,13 +128,13 @@ func (t *NormalizeQueryGenerator) BuildQuery(ctx context.Context) (string, error
 		switch clickHouseType {
 		case "Date32", "Nullable(Date32)":
 			fmt.Fprintf(&projection,
-				"toDate32(parseDateTime64BestEffortOrNull(JSONExtractString(_peerdb_data, %s),6)) AS %s,",
+				"toDate32(parseDateTime64BestEffortOrNull(JSONExtractString(_peerdb_data, %s),6,'UTC')) AS %s,",
 				peerdb_clickhouse.QuoteLiteral(colName),
 				peerdb_clickhouse.QuoteIdentifier(dstColName),
 			)
 			if t.enablePrimaryUpdate {
 				fmt.Fprintf(&projectionUpdate,
-					"toDate32(parseDateTime64BestEffortOrNull(JSONExtractString(_peerdb_match_data, %s),6)) AS %s,",
+					"toDate32(parseDateTime64BestEffortOrNull(JSONExtractString(_peerdb_match_data, %s),6,'UTC')) AS %s,",
 					peerdb_clickhouse.QuoteLiteral(colName),
 					peerdb_clickhouse.QuoteIdentifier(dstColName),
 				)
@@ -144,40 +144,41 @@ func (t *NormalizeQueryGenerator) BuildQuery(ctx context.Context) (string, error
 				// parseDateTime64BestEffortOrNull for hh:mm:ss puts the year as current year
 				// (or previous year if result would be in future) so explicitly anchor to unix epoch
 				fmt.Fprintf(&projection,
-					"parseDateTime64BestEffortOrNull('1970-01-01 ' || JSONExtractString(_peerdb_data, %s),6) AS %s,",
+					"parseDateTime64BestEffortOrNull('1970-01-01 ' || JSONExtractString(_peerdb_data, %s),6,'UTC') AS %s,",
 					peerdb_clickhouse.QuoteLiteral(colName),
 					peerdb_clickhouse.QuoteIdentifier(dstColName),
 				)
 				if t.enablePrimaryUpdate {
 					fmt.Fprintf(&projectionUpdate,
-						"parseDateTime64BestEffortOrNull('1970-01-01 ' || JSONExtractString(_peerdb_match_data, %s),6) AS %s,",
+						"parseDateTime64BestEffortOrNull('1970-01-01 ' || JSONExtractString(_peerdb_match_data, %s),6,'UTC') AS %s,",
 						peerdb_clickhouse.QuoteLiteral(colName),
 						peerdb_clickhouse.QuoteIdentifier(dstColName),
 					)
 				}
 			} else {
 				fmt.Fprintf(&projection,
-					"parseDateTime64BestEffortOrNull(JSONExtractString(_peerdb_data, %s),6) AS %s,",
+					"parseDateTime64BestEffortOrNull(JSONExtractString(_peerdb_data, %s),6,'UTC') AS %s,",
 					peerdb_clickhouse.QuoteLiteral(colName),
 					peerdb_clickhouse.QuoteIdentifier(dstColName),
 				)
 				if t.enablePrimaryUpdate {
 					fmt.Fprintf(&projectionUpdate,
-						"parseDateTime64BestEffortOrNull(JSONExtractString(_peerdb_match_data, %s),6) AS %s,",
+						"parseDateTime64BestEffortOrNull(JSONExtractString(_peerdb_match_data, %s),6,'UTC') AS %s,",
 						peerdb_clickhouse.QuoteLiteral(colName),
 						peerdb_clickhouse.QuoteIdentifier(dstColName),
 					)
 				}
 			}
 		case "Array(DateTime64(6))", "Nullable(Array(DateTime64(6)))":
 			fmt.Fprintf(&projection,
-				`arrayMap(x -> parseDateTime64BestEffortOrNull(trimBoth(x, '"'), 6), JSONExtractArrayRaw(_peerdb_data, %s)) AS %s,`,
+				`arrayMap(x -> parseDateTime64BestEffortOrNull(trimBoth(x, '"'),6,'UTC'), JSONExtractArrayRaw(_peerdb_data, %s)) AS %s,`,
 				peerdb_clickhouse.QuoteLiteral(colName),
 				peerdb_clickhouse.QuoteIdentifier(dstColName),
 			)
 			if t.enablePrimaryUpdate {
 				fmt.Fprintf(&projectionUpdate,
-					`arrayMap(x -> parseDateTime64BestEffortOrNull(trimBoth(x, '"'), 6), JSONExtractArrayRaw(_peerdb_match_data, %s)) AS %s,`,
+					`arrayMap(x -> parseDateTime64BestEffortOrNull(trimBoth(x, '"'),6,'UTC'),`+
+						`JSONExtractArrayRaw(_peerdb_match_data, %s)) AS %s,`,
 					peerdb_clickhouse.QuoteLiteral(colName),
 					peerdb_clickhouse.QuoteIdentifier(dstColName),
 				)

flow/connectors/eventhub/hubmanager.go

@@ -62,8 +62,7 @@ func (m *EventHubManager) GetOrCreateHubClient(ctx context.Context, name ScopedE
 	hub, hubConnectOK = m.hubs.Load(name)
 	if hubConnectOK {
 		hubTmp := hub.(*azeventhubs.ProducerClient)
-		_, err := hubTmp.GetEventHubProperties(ctx, nil)
-		if err != nil {
+		if _, err := hubTmp.GetEventHubProperties(ctx, nil); err != nil {
 			logger := internal.LoggerFromCtx(ctx)
 			logger.Info(
 				fmt.Sprintf("eventhub %s not reachable. Will re-establish connection and re-create it.", name),
@@ -169,8 +168,7 @@ func (m *EventHubManager) EnsureEventHubExists(ctx context.Context, name ScopedE
 			},
 		}
 
-		_, err := hubClient.CreateOrUpdate(ctx, resourceGroup, namespace, name.Eventhub, opts, nil)
-		if err != nil {
+		if _, err := hubClient.CreateOrUpdate(ctx, resourceGroup, namespace, name.Eventhub, opts, nil); err != nil {
 			slog.Error("failed to create event hub", slog.Any("error", err))
 			return err
 		}

flow/connectors/mongo/validate.go

@@ -29,8 +29,7 @@ func (c *MongoConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.F
 		return nil
 	}
 
-	_, err := shared_mongo.GetReplSetGetStatus(ctx, c.client)
-	if err != nil {
+	if _, err := shared_mongo.GetReplSetGetStatus(ctx, c.client); err != nil {
 		return err
 	}
 

flow/connectors/postgres/postgres.go

@@ -73,10 +73,10 @@ func NewPostgresConnector(ctx context.Context, env map[string]string, pgConfig *
 		return nil, err
 	}
 
-	runtimeParams := connConfig.Config.RuntimeParams
-	runtimeParams["idle_in_transaction_session_timeout"] = "0"
-	runtimeParams["statement_timeout"] = "0"
-	runtimeParams["DateStyle"] = "ISO, DMY"
+	connConfig.Config.RuntimeParams["timezone"] = "UTC"
+	connConfig.Config.RuntimeParams["idle_in_transaction_session_timeout"] = "0"
+	connConfig.Config.RuntimeParams["statement_timeout"] = "0"
+	connConfig.Config.RuntimeParams["DateStyle"] = "ISO, DMY"
 
 	tunnel, err := utils.NewSSHTunnel(ctx, pgConfig.SshConfig)
 	if err != nil {
@@ -159,10 +159,9 @@ func (c *PostgresConnector) CreateReplConn(ctx context.Context) (*pgx.Conn, erro
 		return nil, fmt.Errorf("failed to parse connection string: %w", err)
 	}
 
-	runtimeParams := replConfig.Config.RuntimeParams
-	runtimeParams["idle_in_transaction_session_timeout"] = "0"
-	runtimeParams["statement_timeout"] = "0"
-	// ensure that replication is set to database
+	replConfig.Config.RuntimeParams["timezone"] = "UTC"
+	replConfig.Config.RuntimeParams["idle_in_transaction_session_timeout"] = "0"
+	replConfig.Config.RuntimeParams["statement_timeout"] = "0"
 	replConfig.Config.RuntimeParams["replication"] = "database"
 	replConfig.Config.RuntimeParams["bytea_output"] = "hex"
 	replConfig.Config.RuntimeParams["intervalstyle"] = "postgres"

flow/connectors/postgres/qrep_query_executor.go

@@ -34,8 +34,7 @@ func (c *PostgresConnector) NewQRepQueryExecutor(ctx context.Context, version ui
 func (c *PostgresConnector) NewQRepQueryExecutorSnapshot(ctx context.Context, version uint32,
 	snapshot string, flowJobName string, partitionID string,
 ) (*QRepQueryExecutor, error) {
-	_, err := c.fetchCustomTypeMapping(ctx)
-	if err != nil {
+	if _, err := c.fetchCustomTypeMapping(ctx); err != nil {
 		c.logger.Error("[pg_query_executor] failed to fetch custom type mapping", slog.Any("error", err))
 		return nil, fmt.Errorf("failed to fetch custom type mapping: %w", err)
 	}

flow/connectors/postgres/ssh_wrapped_conn.go

@@ -60,8 +60,7 @@ func NewPostgresConnFromConfig(
 	}
 
 	if err := retryWithBackoff(logger, func() error {
-		_, err := conn.Exec(ctx, "SELECT 1")
-		if err != nil {
+		if _, err := conn.Exec(ctx, "SELECT 1"); err != nil {
 			logger.Error("Failed to ping pool", slog.Any("error", err), slog.String("host", connConfig.Host))
 			return err
 		}

flow/connectors/pubsub/pubsub.go

@@ -59,8 +59,7 @@ func (c *PubSubConnector) Close() error {
 
 func (c *PubSubConnector) ConnectionActive(ctx context.Context) error {
 	topic := c.client.Topic("test")
-	_, err := topic.Exists(ctx)
-	if err != nil {
+	if _, err := topic.Exists(ctx); err != nil {
 		return fmt.Errorf("pubsub connection active check failure: %w", err)
 	}
 	return nil

flow/connectors/snowflake/qrep.go

@@ -75,8 +75,7 @@ func (c *SnowflakeConnector) SetupQRepMetadataTables(ctx context.Context, config
 	if schemaExists, err := c.checkIfRawSchemaExists(ctx); err != nil {
 		return fmt.Errorf("error while checking if schema %s for raw table exists: %w", c.rawSchema, err)
 	} else if !schemaExists {
-		_, err := c.execWithLogging(ctx, fmt.Sprintf(createSchemaSQL, c.rawSchema))
-		if err != nil {
+		if _, err := c.execWithLogging(ctx, fmt.Sprintf(createSchemaSQL, c.rawSchema)); err != nil {
 			return err
 		}
 	}
@@ -225,8 +224,7 @@ func (c *SnowflakeConnector) dropStage(ctx context.Context, stagingPath string,
 	stageName := c.getStageNameForJob(job)
 	stmt := "DROP STAGE IF EXISTS " + stageName
 
-	_, err := c.ExecContext(ctx, stmt)
-	if err != nil {
+	if _, err := c.ExecContext(ctx, stmt); err != nil {
 		return fmt.Errorf("failed to drop stage %s: %w", stageName, err)
 	}