Skip to content

Commit 2f48bd2

Browse files
committed
fix(cron): fire Postgres jobs on time by promoting time columns to timestamptz
On Postgres the cron_jobs time columns were TIMESTAMP WITHOUT TIME ZONE. lib/pq sends each Go time.Time carrying its zone offset, but a zone-less column silently drops the offset and keeps only the wall clock — so a Beijing "09:00" job was stored as 09:00 and read back as 09:00 UTC = 17:00 Beijing, firing ~8h late (or never matching next_run <= now() at all). SQLite was unaffected (its driver stores the full instant). Fix: promote next_run / last_run / locked_at / created_at to timestamptz. The type preserves the instant across write/read regardless of offset or session TimeZone, verified end-to-end against lib/pq. Read paths (GetNextDueTime, GetDueCronJobs) need no change — lib/pq restores the moment via its currentLocation for timestamptz. Migration is Postgres-only and idempotent: it probes pg_catalog.format_type and short-circuits once a column is already timestamptz, so daemon restarts and fresh installs are no-ops. BREAKING for existing Postgres deployments: existing cron_jobs rows are truncated on first migration. Their stored next_run wall-clocks already carry the wrong timezone, so converting them would freeze the bug into the new type; a clean reschedule is the correct recovery. Affected recurring schedules must be recreated. Chat history and agent config are untouched. The gateway logs a WARN with the wiped row count before truncating. SQLite deployments are unaffected. Adds Postgres integration coverage (internal/store/cron_pg_test.go), opt-in via FASTCLAW_TEST_PG_DSN — the missing guard that let this bug ship. Documents the upgrade impact in CHANGELOG.md.
1 parent 2833ab5 commit 2f48bd2

3 files changed

Lines changed: 407 additions & 0 deletions

File tree

CHANGELOG.md

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
# Changelog
2+
3+
Notable changes to FastClaw. Items marked **BREAKING** require operator
4+
action on upgrade — read those notes before deploying.
5+
6+
## [Unreleased]
7+
8+
### Fixed
9+
10+
- **Cron jobs fired ~hours late on Postgres when the server ran in a
11+
non-UTC timezone.** The `cron_jobs` time columns were declared
12+
`TIMESTAMP WITHOUT TIME ZONE`. A Go `time.Time` carrying a non-UTC
13+
offset (e.g. `Asia/Shanghai`) was written with its offset silently
14+
dropped, so a Beijing "09:00" job was stored as `09:00` and read back
15+
as `09:00 UTC` = `17:00 Beijing` — firing 8h late (or, in the opposite
16+
direction, never matching `next_run <= now()` at all). SQLite was
17+
unaffected. The columns are now `TIMESTAMPTZ`, which preserves the
18+
instant across write/read regardless of offset or session `TimeZone`.
19+
20+
### BREAKING — cron schedule state is reset on upgrade (Postgres only)
21+
22+
When a Postgres deployment runs the schema migration for the first time,
23+
**every existing row in `cron_jobs` is deleted.** This is deliberate:
24+
the stored `next_run` wall-clocks already carry the wrong timezone, so
25+
converting them would freeze the bug into the new column type. A clean
26+
reschedule is the correct recovery.
27+
28+
- **What is lost:** pending scheduled jobs (recurring `cron`, `interval`,
29+
and not-yet-fired `once` reminders).
30+
- **What is NOT lost:** chat history (`sessions`, `session_messages`),
31+
agent identity files, provider/channel config — none of these are
32+
touched.
33+
- **Operator action required:** after upgrading, any recurring schedule
34+
a user relies on (e.g. "every day at 9am") must be recreated by asking
35+
the agent again, or via the dashboard's Scheduler tab. The original
36+
`create_cron_job` tool calls are still visible in chat history and can
37+
serve as a reference for what to rebuild.
38+
- **Visibility:** the gateway logs a single
39+
`level=WARN msg="resetting cron schedule state for timestamptz migration …"`
40+
line with the row count before wiping, so operators can tell from the
41+
upgrade log whether any jobs were affected.
42+
- **SQLite deployments:** unaffected — the migration is skipped entirely.
43+
- **Idempotent:** re-running the migration (e.g. on every daemon boot)
44+
is a no-op once the columns are already `timestamptz`.

internal/store/cron_pg_test.go

Lines changed: 252 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,252 @@
1+
package store
2+
3+
import (
4+
"context"
5+
"database/sql"
6+
"fmt"
7+
"os"
8+
"strings"
9+
"testing"
10+
"time"
11+
12+
_ "github.com/lib/pq"
13+
)
14+
15+
// pgTestDSN is read from the environment to opt this test in. CI without a
16+
// Postgres instance skips; a developer running ./scripts/run-postgres.sh sets
17+
// it (the helper exports a matching DSN) to exercise the real backend.
18+
const pgTestDSNEnv = "FASTCLAW_TEST_PG_DSN"
19+
20+
// openTestPG connects to the Postgres instance named by FASTCLAW_TEST_PG_DSN,
21+
// creates a uniquely-named throwaway database, and returns a migrated *DBStore
22+
// pointing at it plus a cleanup that drops the DB and closes the admin handle.
23+
// The fresh database means every test starts from a clean schema — the
24+
// timestamptz migration runs on CREATE TABLE + convert, exercising the same
25+
// path a brand-new install takes.
26+
func openTestPG(t *testing.T) (*DBStore, func()) {
27+
t.Helper()
28+
dsn := os.Getenv(pgTestDSNEnv)
29+
if dsn == "" {
30+
t.Skipf("%s not set — skipping Postgres integration test", pgTestDSNEnv)
31+
}
32+
// Connect to the maintenance DB (the DSN already targets a DB, but we need
33+
// a server-level handle to CREATE the scratch database). Reuse the creds by
34+
// rewriting the path to "postgres".
35+
adminDSN := pgRewriteDB(dsn, "postgres")
36+
admin, err := sql.Open("postgres", adminDSN)
37+
if err != nil {
38+
t.Fatalf("open admin conn: %v", err)
39+
}
40+
scratch := fmt.Sprintf("fc_test_%x", time.Now().UnixNano())
41+
// DROP+CREATE: escape the identifier just in case, though the name is hex.
42+
if _, err := admin.Exec(fmt.Sprintf(`DROP DATABASE IF EXISTS %q`, scratch)); err != nil {
43+
admin.Close()
44+
t.Fatalf("drop scratch db: %v", err)
45+
}
46+
if _, err := admin.Exec(fmt.Sprintf(`CREATE DATABASE %q`, scratch)); err != nil {
47+
admin.Close()
48+
t.Fatalf("create scratch db: %v", err)
49+
}
50+
storeDSN := pgRewriteDB(dsn, scratch)
51+
st, err := NewDBStore("postgres", storeDSN)
52+
if err != nil {
53+
admin.Exec(fmt.Sprintf(`DROP DATABASE IF EXISTS %q`, scratch))
54+
admin.Close()
55+
t.Fatalf("open store: %v", err)
56+
}
57+
if err := st.Migrate(context.Background()); err != nil {
58+
st.Close()
59+
admin.Exec(fmt.Sprintf(`DROP DATABASE IF EXISTS %q`, scratch))
60+
admin.Close()
61+
t.Fatalf("migrate: %v", err)
62+
}
63+
cleanup := func() {
64+
st.Close()
65+
_, _ = admin.Exec(fmt.Sprintf(`DROP DATABASE IF EXISTS %q`, scratch))
66+
admin.Close()
67+
}
68+
return st, cleanup
69+
}
70+
71+
// pgRewriteDB returns a copy of a lib/pq DSN with the database path replaced
72+
// by newDB. It handles both the "postgres://…/dbname?…" URL form and the
73+
// "host=… dbname=…" keyword form. Used to retarget a connection at the
74+
// maintenance "postgres" DB or a per-test scratch DB.
75+
func pgRewriteDB(dsn, newDB string) string {
76+
// URL form: scheme://user:pass@host:port/dbname?params
77+
if i := strings.Index(dsn, "://"); i >= 0 {
78+
pathStart := i + 3
79+
// find start of path after the authority (first '/' after host[:port])
80+
slash := strings.IndexByte(dsn[pathStart:], '/')
81+
if slash < 0 {
82+
return dsn + "/" + newDB
83+
}
84+
slash += pathStart
85+
query := strings.IndexByte(dsn[slash+1:], '?')
86+
if query < 0 {
87+
return dsn[:slash+1] + newDB
88+
}
89+
return dsn[:slash+1] + newDB + dsn[slash+1+query:]
90+
}
91+
// Keyword form: replace dbname=… value (space-terminated).
92+
if k := strings.Index(dsn, "dbname="); k >= 0 {
93+
rest := dsn[k+len("dbname="):]
94+
sp := strings.IndexByte(rest, ' ')
95+
if sp < 0 {
96+
return dsn[:k] + "dbname=" + newDB
97+
}
98+
return dsn[:k] + "dbname=" + newDB + rest[sp:]
99+
}
100+
return dsn + " dbname=" + newDB
101+
}
102+
103+
// TestCronJobsPGTimestampTZ verifies the timezone fix on Postgres. The cron
104+
// schedule columns must be timestamptz so a Go time.Time carrying a non-UTC
105+
// offset survives the write/read round-trip as the exact same instant.
106+
//
107+
// This is the regression guard for the bug where a Beijing-time "09:00" job
108+
// was stored as 09:00 wall-clock and reinterpreted as 09:00 UTC = 17:00
109+
// Beijing, firing 8h late. Before the fix the drift assertion fails.
110+
func TestCronJobsPGTimestampTZ(t *testing.T) {
111+
st, cleanup := openTestPG(t)
112+
defer cleanup()
113+
ctx := context.Background()
114+
db := st.DB()
115+
116+
// Schema assertion: the four time columns are timestamptz, not timestamp.
117+
for _, col := range []string{"next_run", "last_run", "locked_at", "created_at"} {
118+
isTZ, err := st.columnIsTimestampTZ(ctx, "cron_jobs", col)
119+
if err != nil {
120+
t.Fatalf("probe %s: %v", col, err)
121+
}
122+
if !isTZ {
123+
t.Errorf("cron_jobs.%s is not timestamptz — timezone bug not migrated", col)
124+
}
125+
}
126+
127+
// Regression: write a non-UTC-offset instant (Beijing 09:00 = 01:00 UTC)
128+
// and confirm the stored instant reads back exactly, with no drift.
129+
beijing, err := time.LoadLocation("Asia/Shanghai")
130+
if err != nil {
131+
t.Fatalf("load Asia/Shanghai: %v", err)
132+
}
133+
want := time.Date(2026, 7, 2, 9, 0, 0, 0, beijing) // 09:00 +08 = 01:00 UTC
134+
135+
job := &CronJobRecord{
136+
ID: "tz-test-1",
137+
AgentID: "a_tz",
138+
Name: "beijing 9am",
139+
Type: "cron",
140+
Schedule: "0 9 * * *",
141+
Message: "wake up",
142+
Channel: "web",
143+
ChatID: "c",
144+
Timezone: "Asia/Shanghai",
145+
Enabled: true,
146+
NextRun: &want,
147+
CreatedAt: time.Now().UTC(),
148+
}
149+
if err := st.SaveCronJob(ctx, job); err != nil {
150+
t.Fatalf("save: %v", err)
151+
}
152+
153+
got, err := st.GetCronJob(ctx, "tz-test-1")
154+
if err != nil {
155+
t.Fatalf("get: %v", err)
156+
}
157+
if got.NextRun == nil {
158+
t.Fatal("NextRun nil after round-trip")
159+
}
160+
drift := got.NextRun.Sub(want)
161+
if drift < -time.Second || drift > time.Second {
162+
t.Errorf("next_run drift after round-trip = %v (want ~0). wrote %s, read %s",
163+
drift, want.Format(time.RFC3339), got.NextRun.Format(time.RFC3339))
164+
}
165+
166+
// GetNextDueTime must return the same instant — the scheduler sleeps
167+
// until it, so any drift here directly shifts the fire time.
168+
nextDue, err := st.GetNextDueTime(ctx)
169+
if err != nil {
170+
t.Fatalf("GetNextDueTime: %v", err)
171+
}
172+
dueDrift := nextDue.Sub(want)
173+
if dueDrift < -time.Second || dueDrift > time.Second {
174+
t.Errorf("GetNextDueTime drift = %v (want ~0). want %s, got %s",
175+
dueDrift, want.Format(time.RFC3339), nextDue.Format(time.RFC3339))
176+
}
177+
178+
// Sanity: also confirm a second column (last_run via UpdateCronJobRun)
179+
// round-trips. last_run is written through UpdateCronJobRun.
180+
lastRun := time.Date(2026, 7, 1, 9, 0, 0, 0, beijing)
181+
future := want.Add(24 * time.Hour)
182+
if err := st.UpdateCronJobRun(ctx, "tz-test-1", lastRun, future); err != nil {
183+
t.Fatalf("UpdateCronJobRun: %v", err)
184+
}
185+
got2, err := st.GetCronJob(ctx, "tz-test-1")
186+
if err != nil {
187+
t.Fatalf("get after update: %v", err)
188+
}
189+
if got2.LastRun == nil {
190+
t.Fatal("LastRun nil after update")
191+
}
192+
lastDrift := got2.LastRun.Sub(lastRun)
193+
if lastDrift < -time.Second || lastDrift > time.Second {
194+
t.Errorf("last_run drift = %v. wrote %s, read %s",
195+
lastDrift, lastRun.Format(time.RFC3339), got2.LastRun.Format(time.RFC3339))
196+
}
197+
198+
// locked_at is exercised by LockCronJob. Verify it too for completeness.
199+
if _, err := db.ExecContext(ctx,
200+
`UPDATE cron_jobs SET locked_at = $1 WHERE id = $2`, want, "tz-test-1"); err != nil {
201+
t.Fatalf("set locked_at: %v", err)
202+
}
203+
var lockedAt time.Time
204+
if err := db.QueryRowContext(ctx,
205+
`SELECT locked_at FROM cron_jobs WHERE id = $1`, "tz-test-1").Scan(&lockedAt); err != nil {
206+
t.Fatalf("read locked_at: %v", err)
207+
}
208+
lockDrift := lockedAt.Sub(want)
209+
if lockDrift < -time.Second || lockDrift > time.Second {
210+
t.Errorf("locked_at drift = %v. wrote %s, read %s",
211+
lockDrift, want.Format(time.RFC3339), lockedAt.Format(time.RFC3339))
212+
}
213+
}
214+
215+
// TestCronJobsPGMigrationIdempotent asserts that re-running Migrate on an
216+
// already-converted schema is a no-op (no error, columns stay timestamptz,
217+
// no data loss on the row we leave behind). The migration must be safe to
218+
// run repeatedly — daemon restarts call Migrate on every boot.
219+
func TestCronJobsPGMigrationIdempotent(t *testing.T) {
220+
st, cleanup := openTestPG(t)
221+
defer cleanup()
222+
ctx := context.Background()
223+
224+
// Leave a row in place, then re-run Migrate. The idempotent path must
225+
// detect timestamptz and skip — it must NOT truncate.
226+
keep := time.Now().Add(time.Hour).UTC()
227+
job := &CronJobRecord{
228+
ID: "keep-1", AgentID: "a", Type: "interval", Schedule: "1h",
229+
Message: "x", Channel: "web", ChatID: "c", Timezone: "UTC",
230+
Enabled: true, NextRun: &keep, CreatedAt: time.Now().UTC(),
231+
}
232+
if err := st.SaveCronJob(ctx, job); err != nil {
233+
t.Fatalf("save: %v", err)
234+
}
235+
236+
if err := st.Migrate(ctx); err != nil {
237+
t.Fatalf("second Migrate failed: %v", err)
238+
}
239+
240+
// Row must survive (no truncation on the idempotent path).
241+
got, err := st.GetCronJob(ctx, "keep-1")
242+
if err != nil {
243+
t.Fatalf("get after re-migrate: %v (row was wrongly truncated)", err)
244+
}
245+
if got.NextRun == nil {
246+
t.Fatal("NextRun nil after re-migrate")
247+
}
248+
drift := got.NextRun.Sub(keep)
249+
if drift < -time.Second || drift > time.Second {
250+
t.Errorf("next_run changed on idempotent re-migrate: drift=%v", drift)
251+
}
252+
}

0 commit comments

Comments
 (0)