diff --git a/cmd/archiver/main.go b/cmd/archiver/main.go index addbc78..6023bda 100644 --- a/cmd/archiver/main.go +++ b/cmd/archiver/main.go @@ -100,7 +100,7 @@ func dispatchCLI(ctx context.Context) bool { // setupConnection connects to PostgreSQL, verifies the connection, and ensures // the watermark table exists. Any failure log.Fatalf's — this is the cron body. func setupConnection(ctx context.Context, cfg *config.Config) (*pgx.Conn, *watermark.Store) { - conn, err := pgx.Connect(ctx, cfg.Postgres.DSN) + conn, err := partition.Connect(ctx, cfg.Postgres.DSN) if err != nil { log.Fatalf("connect pg: %v", err) } diff --git a/cmd/partitioner/main.go b/cmd/partitioner/main.go index f9f4985..f180a9b 100644 --- a/cmd/partitioner/main.go +++ b/cmd/partitioner/main.go @@ -58,7 +58,7 @@ func main() { log.Fatalf("load config: %v", err) } - conn, err := pgx.Connect(ctx, cfg.Postgres.DSN) + conn, err := partition.Connect(ctx, cfg.Postgres.DSN) if err != nil { log.Fatalf("connect: %v", err) } diff --git a/internal/partition/partition.go b/internal/partition/partition.go index 6caa42a..eefe3ea 100644 --- a/internal/partition/partition.go +++ b/internal/partition/partition.go @@ -80,14 +80,18 @@ func ParseBoundExpr(expr string) (lower, upper time.Time, err error) { return parseBoundPair(expr, TimeBoundary{}) } -// parseTimestamp parses a pg_get_expr-style bound value (the quoted string -// inside FOR VALUES FROM/TO) into a UTC time. Accepts the common layouts -// PostgreSQL emits. +// parseTimestamp parses a pg_get_expr bound value (the quoted string inside +// FOR VALUES FROM/TO) into a UTC time. PostgreSQL renders bounds ISO-style but +// not as strict ISO 8601 (space separator, bare +00 offset), so time.RFC3339 +// does not apply. The layouts are the stdlib time.DateTime (timestamp without +// time zone) and time.DateOnly (date), plus a numeric-offset layout for +// timestamptz; time.Parse consumes any trailing fractional second on its own. +// Callers connect via Connect, which pins the session so the render stays in this set. func parseTimestamp(s string) (time.Time, error) { for _, layout := range []string{ - "2006-01-02 15:04:05+00", "2006-01-02 15:04:05-07", - "2006-01-02", + time.DateTime, + time.DateOnly, } { if t, err := time.Parse(layout, s); err == nil { return t.UTC(), nil @@ -96,6 +100,32 @@ func parseTimestamp(s string) (time.Time, error) { return time.Time{}, fmt.Errorf("unrecognized timestamp format: %q", s) } +// boundConnConfig parses dsn and pins DateStyle and TimeZone as connection +// runtime params, so pg_get_expr(relpartbound) renders partition bounds +// deterministically (ISO date order, UTC) and parseTimestamp's layout set stays +// complete regardless of the server's defaults. The params ride the startup +// packet, so they apply to the connection (and any reconnect) without a SET. +func boundConnConfig(dsn string) (*pgx.ConnConfig, error) { + cfg, err := pgx.ParseConfig(dsn) + if err != nil { + return nil, fmt.Errorf("parse dsn: %w", err) + } + cfg.RuntimeParams["datestyle"] = "ISO, YMD" + cfg.RuntimeParams["timezone"] = "UTC" + return cfg, nil +} + +// Connect opens a pgx connection with the bound-rendering params pinned (see +// boundConnConfig). Both the archiver and the partitioner read partition bounds, +// so both connect through here. +func Connect(ctx context.Context, dsn string) (*pgx.Conn, error) { + cfg, err := boundConnConfig(dsn) + if err != nil { + return nil, err + } + return pgx.ConnectConfig(ctx, cfg) +} + // PartitionName generates a partition table name for the given date and period. func PartitionName(t time.Time, period string) string { if period == PeriodDaily { diff --git a/internal/partition/partition_test.go b/internal/partition/partition_test.go index 940617d..b6cde4e 100644 --- a/internal/partition/partition_test.go +++ b/internal/partition/partition_test.go @@ -106,6 +106,30 @@ func TestParseBoundExpr(t *testing.T) { wantLow: time.Date(2025, 11, 1, 0, 0, 0, 0, time.UTC), wantHigh: time.Date(2025, 12, 1, 0, 0, 0, 0, time.UTC), }, + { + name: "timestamp without time zone (no offset)", + expr: "FOR VALUES FROM ('2026-04-01 00:00:00') TO ('2026-05-01 00:00:00')", + wantLow: time.Date(2026, 4, 1, 0, 0, 0, 0, time.UTC), + wantHigh: time.Date(2026, 5, 1, 0, 0, 0, 0, time.UTC), + }, + { + name: "timestamp without time zone, fractional seconds", + expr: "FOR VALUES FROM ('2026-04-01 00:00:00.123456') TO ('2026-05-01 00:00:00')", + wantLow: time.Date(2026, 4, 1, 0, 0, 0, 123456000, time.UTC), + wantHigh: time.Date(2026, 5, 1, 0, 0, 0, 0, time.UTC), + }, + { + name: "timestamptz non-UTC offset", + expr: "FOR VALUES FROM ('2026-04-01 00:00:00-07') TO ('2026-05-01 00:00:00-07')", + wantLow: time.Date(2026, 4, 1, 7, 0, 0, 0, time.UTC), + wantHigh: time.Date(2026, 5, 1, 7, 0, 0, 0, time.UTC), + }, + { + name: "timestamptz fractional seconds with offset", + expr: "FOR VALUES FROM ('2026-04-01 00:00:00.123456+00') TO ('2026-05-01 00:00:00+00')", + wantLow: time.Date(2026, 4, 1, 0, 0, 0, 123456000, time.UTC), + wantHigh: time.Date(2026, 5, 1, 0, 0, 0, 0, time.UTC), + }, { name: "invalid format", expr: "not a valid expression", @@ -127,6 +151,21 @@ func TestParseBoundExpr(t *testing.T) { } } +func TestBoundConnConfig(t *testing.T) { + cfg, err := boundConnConfig("postgres://u@localhost:5432/db") + require.NoError(t, err) + assert.Equal(t, "ISO, YMD", cfg.RuntimeParams["datestyle"]) + assert.Equal(t, "UTC", cfg.RuntimeParams["timezone"]) +} + +// A malformed DSN fails in boundConnConfig (pgx.ParseConfig) before any network +// I/O, so Connect's error path is unit-testable without a live server. +func TestConnect_MalformedDSN(t *testing.T) { + _, err := Connect(context.Background(), "postgres://u@localhost:notaport/db") + require.Error(t, err) + assert.Contains(t, err.Error(), "parse dsn") +} + func TestPartitionName(t *testing.T) { assert.Equal(t, "p_2026_03", PartitionName(time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC), "monthly")) assert.Equal(t, "p_2026_03_15", PartitionName(time.Date(2026, 3, 15, 0, 0, 0, 0, time.UTC), "daily"))