From cd36b557eee9d1753c281599955ff1f499c537ae Mon Sep 17 00:00:00 2001 From: Mitchell Hashimoto Date: Fri, 1 May 2026 22:06:01 -0700 Subject: [PATCH] store: order latest buildkite build by monotonic int LookupBuildkiteBuildByTuple sorted on created_at, an RFC3339Nano text column. Lexical comparison of nanosecond timestamps is not reliable: time.Format trims trailing zeros, so an instant on the exact second renders as '...:00Z' while one nanosecond later renders as '...:00.000000001Z' and lex-sorts before it. The practical effect was that /logs could resolve the wrong run for a workflow that had been triggered more than once. Add a created_unix_ns INTEGER column to buildkite_builds, populate it from time.Now().UnixNano() on insert, and switch the lookup to ORDER BY created_unix_ns DESC with created_at and build_number as deterministic tiebreakers for legacy rows that pre-date the column. The migration path is covered: an additive ALTER widens existing databases, and a one-shot Go-side backfill parses each row's created_at and writes the corresponding UnixNano. Rows whose text fails to parse are left at the default 0 so a single corrupt row cannot wedge startup. New tests in store_migrate_test.go open a hand-crafted pre-migration database through openStore and assert the upgrade is correct, idempotent, and tolerant of bad data. --- store.go | 38 ++++--- store_migrate.go | 94 ++++++++++++++-- store_migrate_test.go | 242 ++++++++++++++++++++++++++++++++++++++++++ store_test.go | 94 ++++++++++++++++ 4 files changed, 447 insertions(+), 21 deletions(-) create mode 100644 store_migrate_test.go diff --git a/store.go b/store.go index 9455fbd..69cf76f 100644 --- a/store.go +++ b/store.go @@ -352,24 +352,31 @@ type BuildkiteBuildRef struct { // Buildkite-side rebuild that re-fires us) just refreshes the row // instead of failing. func (s *store) InsertBuildkiteBuild(ctx context.Context, ref BuildkiteBuildRef) error { + // Capture wall-clock and monotonic-friendly forms once so the two + // columns agree on the same instant. created_at is the + // human-readable RFC3339Nano string; created_unix_ns is the + // integer the lookup orders on (text comparison of nanosecond + // timestamps isn't reliable, so we sort on the int instead). + now := time.Now().UTC() _, err := s.db.ExecContext(ctx, `INSERT INTO buildkite_builds ( build_uuid, build_number, pipeline_slug, org, knot, pipeline_rkey, workflow, - pipeline_uri, created_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + pipeline_uri, created_at, created_unix_ns + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(build_uuid) DO UPDATE SET - build_number = excluded.build_number, - pipeline_slug = excluded.pipeline_slug, - org = excluded.org, - knot = excluded.knot, - pipeline_rkey = excluded.pipeline_rkey, - workflow = excluded.workflow, - pipeline_uri = excluded.pipeline_uri, - created_at = excluded.created_at`, + build_number = excluded.build_number, + pipeline_slug = excluded.pipeline_slug, + org = excluded.org, + knot = excluded.knot, + pipeline_rkey = excluded.pipeline_rkey, + workflow = excluded.workflow, + pipeline_uri = excluded.pipeline_uri, + created_at = excluded.created_at, + created_unix_ns = excluded.created_unix_ns`, ref.BuildUUID, ref.BuildNumber, ref.PipelineSlug, ref.Org, ref.Knot, ref.PipelineRkey, ref.Workflow, - ref.PipelineURI, time.Now().UTC().Format(time.RFC3339Nano), + ref.PipelineURI, now.Format(time.RFC3339Nano), now.UnixNano(), ) if err != nil { return fmt.Errorf("insert buildkite_build: %w", err) @@ -411,6 +418,13 @@ func (s *store) LookupBuildkiteBuildByUUID(ctx context.Context, buildUUID string // over time (rebuilds, re-triggers). We always serve logs for the // latest run; older runs are still queryable by build UUID directly // if anyone ever wants that. +// +// Ordering is on created_unix_ns (a monotonic int) rather than +// created_at. Text comparison of RFC3339Nano timestamps is not +// reliable across nanosecond precision, which used to make this +// query occasionally pick the wrong run. created_at and build_number +// are kept as deterministic tiebreakers for legacy rows that pre-date +// the new column and still scan as 0. func (s *store) LookupBuildkiteBuildByTuple(ctx context.Context, knot, pipelineRkey, workflow string) (*BuildkiteBuildRef, error) { var ref BuildkiteBuildRef err := s.db.QueryRowContext(ctx, @@ -418,7 +432,7 @@ func (s *store) LookupBuildkiteBuildByTuple(ctx context.Context, knot, pipelineR knot, pipeline_rkey, workflow, pipeline_uri FROM buildkite_builds WHERE knot = ? AND pipeline_rkey = ? AND workflow = ? - ORDER BY created_at DESC + ORDER BY created_unix_ns DESC, created_at DESC, build_number DESC LIMIT 1`, knot, pipelineRkey, workflow, ).Scan( diff --git a/store_migrate.go b/store_migrate.go index 7133357..572e4d1 100644 --- a/store_migrate.go +++ b/store_migrate.go @@ -8,6 +8,7 @@ import ( "context" "fmt" "strings" + "time" ) // schema is the full set of CREATE statements applied at startup. It is @@ -100,16 +101,23 @@ CREATE TABLE IF NOT EXISTS events ( -- string means "use the provider's defaultOrg" — that's both the -- usual single-org case and what every row written before this -- column existed will scan as. +-- created_unix_ns is the monotonic recency key. created_at is kept +-- (RFC3339Nano text) for human-readable inspection, but it must NOT +-- be used for ordering: text comparison of nanosecond timestamps is +-- not reliable, which used to make /logs occasionally resolve the +-- wrong run. created_unix_ns is the integer the latest-build lookup +-- sorts on instead. CREATE TABLE IF NOT EXISTS buildkite_builds ( - build_uuid TEXT PRIMARY KEY, - build_number INTEGER NOT NULL, - pipeline_slug TEXT NOT NULL, - org TEXT NOT NULL DEFAULT '', - knot TEXT NOT NULL, - pipeline_rkey TEXT NOT NULL, - workflow TEXT NOT NULL, - pipeline_uri TEXT NOT NULL, - created_at TEXT NOT NULL + build_uuid TEXT PRIMARY KEY, + build_number INTEGER NOT NULL, + pipeline_slug TEXT NOT NULL, + org TEXT NOT NULL DEFAULT '', + knot TEXT NOT NULL, + pipeline_rkey TEXT NOT NULL, + workflow TEXT NOT NULL, + pipeline_uri TEXT NOT NULL, + created_at TEXT NOT NULL, + created_unix_ns INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS buildkite_builds_lookup ON buildkite_builds (knot, pipeline_rkey, workflow); @@ -133,6 +141,15 @@ func (s *store) migrate(ctx context.Context) error { // chose. Pre-existing rows scan as empty string, which // the provider treats as "use defaultOrg". `ALTER TABLE buildkite_builds ADD COLUMN org TEXT NOT NULL DEFAULT ''`, + + // Monotonic integer ordering key for buildkite_builds. + // Replaces ORDER BY created_at (RFC3339Nano text), whose + // lexical order isn't reliable across nanosecond precision + // and could make /logs resolve the wrong run. Default 0 + // covers pre-existing rows; the backfill below promotes + // them to their parsed timestamp so ordering stays stable + // across the upgrade. + `ALTER TABLE buildkite_builds ADD COLUMN created_unix_ns INTEGER NOT NULL DEFAULT 0`, } { if _, err := s.db.ExecContext(ctx, alter); err != nil { if strings.Contains(err.Error(), "duplicate column name") { @@ -141,5 +158,64 @@ func (s *store) migrate(ctx context.Context) error { return fmt.Errorf("apply alter %q: %w", alter, err) } } + + if err := s.backfillBuildkiteCreatedUnixNS(ctx); err != nil { + return fmt.Errorf("backfill buildkite created_unix_ns: %w", err) + } + return nil +} + +// backfillBuildkiteCreatedUnixNS walks every buildkite_builds row whose +// created_unix_ns is still the post-ALTER default (0) and sets it from +// the RFC3339Nano text in created_at. SQLite has no native nanosecond +// parser, so the conversion has to happen in Go. +// +// Rows whose created_at can't be parsed are left at 0; that keeps a +// single corrupt row from blocking startup, and ordering between two +// 0-keyed rows still falls through to created_at as a deterministic +// tiebreaker in the lookup query. +func (s *store) backfillBuildkiteCreatedUnixNS(ctx context.Context) error { + rows, err := s.db.QueryContext(ctx, + `SELECT build_uuid, created_at FROM buildkite_builds + WHERE created_unix_ns = 0`, + ) + if err != nil { + return fmt.Errorf("query rows to backfill: %w", err) + } + type pending struct { + uuid string + ns int64 + } + var todo []pending + for rows.Next() { + var uuid, createdAt string + if err := rows.Scan(&uuid, &createdAt); err != nil { + rows.Close() + return fmt.Errorf("scan row: %w", err) + } + t, perr := time.Parse(time.RFC3339Nano, createdAt) + if perr != nil { + // Skip unparseable rows rather than failing the whole + // migration. The lookup query still has a fallback + // ordering for rows that share the default 0 key. + continue + } + todo = append(todo, pending{uuid: uuid, ns: t.UnixNano()}) + } + if err := rows.Err(); err != nil { + rows.Close() + return fmt.Errorf("iterate rows: %w", err) + } + rows.Close() + + for _, p := range todo { + if _, err := s.db.ExecContext(ctx, + `UPDATE buildkite_builds SET created_unix_ns = ? + WHERE build_uuid = ?`, + p.ns, p.uuid, + ); err != nil { + return fmt.Errorf("update row %q: %w", p.uuid, err) + } + } return nil } diff --git a/store_migrate_test.go b/store_migrate_test.go new file mode 100644 index 0000000..bf59168 --- /dev/null +++ b/store_migrate_test.go @@ -0,0 +1,242 @@ +package main + +// Migration tests for the SQLite store. These specifically cover the +// upgrade path from a pre-`created_unix_ns` database (the shape every +// production tack.db that pre-dates this commit will have on first +// boot) to the current schema. The fresh-database path is exercised +// implicitly by every other test via newTestStore. These tests are +// about the *transition*. + +import ( + "context" + "database/sql" + "fmt" + "path/filepath" + "testing" + "time" +) + +// legacyBuildkiteSchema is the buildkite_builds table as it looked +// before created_unix_ns was added. We hand-craft a database in this +// shape so the migration has something realistic to widen and backfill. +// +// Note that org is also absent: the migrate() path adds it via ALTER +// too, so the test doubles as coverage for two stacked column adds +// applying to the same table on the same upgrade. +const legacyBuildkiteSchema = ` +CREATE TABLE buildkite_builds ( + build_uuid TEXT PRIMARY KEY, + build_number INTEGER NOT NULL, + pipeline_slug TEXT NOT NULL, + knot TEXT NOT NULL, + pipeline_rkey TEXT NOT NULL, + workflow TEXT NOT NULL, + pipeline_uri TEXT NOT NULL, + created_at TEXT NOT NULL +); +` + +// openLegacyStore opens a brand-new sqlite file, hand-installs the +// pre-migration schema, and seeds it with the supplied rows. It +// returns the file path so the caller can re-open it through the +// real openStore (which runs migrate()) and observe the upgrade. +// +// We deliberately don't go through openStore for the seeding step, +// since that would apply the current schema and defeat the point. +func openLegacyStore(t *testing.T, rows []legacyRow) string { + t.Helper() + path := filepath.Join(t.TempDir(), "tack.db") + dsn := fmt.Sprintf("file:%s?_journal_mode=WAL&_synchronous=NORMAL&_foreign_keys=on", path) + db, err := sql.Open("sqlite3", dsn) + if err != nil { + t.Fatalf("open legacy db: %v", err) + } + defer db.Close() + + if _, err := db.Exec(legacyBuildkiteSchema); err != nil { + t.Fatalf("install legacy schema: %v", err) + } + for _, r := range rows { + if _, err := db.Exec( + `INSERT INTO buildkite_builds ( + build_uuid, build_number, pipeline_slug, + knot, pipeline_rkey, workflow, + pipeline_uri, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + r.uuid, r.buildNumber, "p", + r.knot, r.rkey, r.workflow, + "at://x", r.createdAt, + ); err != nil { + t.Fatalf("seed legacy row %q: %v", r.uuid, err) + } + } + return path +} + +// legacyRow is the minimal set of fields we need to seed for the +// migration tests. Anything unspecified in the legacy schema (org, +// created_unix_ns) is filled in by the migration itself. +type legacyRow struct { + uuid string + knot, rkey, workflow string + buildNumber int64 + createdAt string +} + +// TestMigrateAddsAndBackfillsCreatedUnixNS exercises the full upgrade: +// a database in the old schema gets opened through openStore, which +// runs migrate(), which should (a) add the created_unix_ns column and +// (b) populate it from each row's existing created_at text. +func TestMigrateAddsAndBackfillsCreatedUnixNS(t *testing.T) { + older := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + newer := older.Add(time.Nanosecond) // see store_test.go for why this matters + + path := openLegacyStore(t, []legacyRow{ + { + uuid: "older", knot: "k", rkey: "r", workflow: "w", + buildNumber: 1, createdAt: older.Format(time.RFC3339Nano), + }, + { + uuid: "newer", knot: "k", rkey: "r", workflow: "w", + buildNumber: 2, createdAt: newer.Format(time.RFC3339Nano), + }, + }) + + s, err := openStore(path) + if err != nil { + t.Fatalf("openStore (migrate): %v", err) + } + defer s.Close() + + // Column must now exist and be populated for both seeded rows. + got := map[string]int64{} + rows, err := s.db.Query(`SELECT build_uuid, created_unix_ns FROM buildkite_builds`) + if err != nil { + t.Fatalf("select after migrate: %v", err) + } + for rows.Next() { + var uuid string + var ns int64 + if err := rows.Scan(&uuid, &ns); err != nil { + t.Fatalf("scan: %v", err) + } + got[uuid] = ns + } + if err := rows.Err(); err != nil { + t.Fatalf("iterate: %v", err) + } + + if got["older"] != older.UnixNano() { + t.Errorf("older: created_unix_ns = %d; want %d", got["older"], older.UnixNano()) + } + if got["newer"] != newer.UnixNano() { + t.Errorf("newer: created_unix_ns = %d; want %d", got["newer"], newer.UnixNano()) + } + + // And the lookup query (the actual reason the column exists) + // must surface the newer row, which is the case the original bug + // report says was getting it wrong before. + ref, err := s.LookupBuildkiteBuildByTuple(context.Background(), "k", "r", "w") + if err != nil { + t.Fatalf("lookup: %v", err) + } + if ref == nil || ref.BuildUUID != "newer" { + t.Fatalf("lookup picked %+v; want uuid=newer", ref) + } +} + +// TestMigrateIdempotent makes sure running migrate() repeatedly is +// safe: re-applying the ALTERs returns "duplicate column name" which +// is swallowed, and the backfill sees no rows with created_unix_ns=0 +// the second time around so it must not clobber the values written +// on the first pass. +func TestMigrateIdempotent(t *testing.T) { + when := time.Date(2026, 5, 1, 12, 34, 56, 789, time.UTC) + path := openLegacyStore(t, []legacyRow{{ + uuid: "u1", knot: "k", rkey: "r", workflow: "w", + buildNumber: 1, createdAt: when.Format(time.RFC3339Nano), + }}) + + // First open: migrate runs, column is added and backfilled. + s, err := openStore(path) + if err != nil { + t.Fatalf("first openStore: %v", err) + } + + // Run migrate() again on the same handle. Should be a no-op and + // must not return an error from a redundant ALTER. + if err := s.migrate(context.Background()); err != nil { + t.Fatalf("second migrate: %v", err) + } + s.Close() + + // Re-open from scratch (which also re-runs migrate) and confirm + // the backfilled value is still intact. + s2, err := openStore(path) + if err != nil { + t.Fatalf("re-open: %v", err) + } + defer s2.Close() + + var ns int64 + if err := s2.db.QueryRow( + `SELECT created_unix_ns FROM buildkite_builds WHERE build_uuid = ?`, "u1", + ).Scan(&ns); err != nil { + t.Fatalf("read back: %v", err) + } + if ns != when.UnixNano() { + t.Fatalf("created_unix_ns = %d after repeated migrate; want %d", ns, when.UnixNano()) + } +} + +// TestMigrateBackfillSkipsUnparseableCreatedAt confirms the backfill's +// "skip and keep going" behavior for malformed timestamps. A single +// corrupt row must not block startup; the row is left at 0 (the +// post-ALTER default) and the lookup's tiebreaker columns then take +// over for it. +func TestMigrateBackfillSkipsUnparseableCreatedAt(t *testing.T) { + good := time.Date(2026, 7, 8, 9, 10, 11, 12, time.UTC) + path := openLegacyStore(t, []legacyRow{ + { + uuid: "good", knot: "k", rkey: "r", workflow: "w", + buildNumber: 1, createdAt: good.Format(time.RFC3339Nano), + }, + { + uuid: "garbage", knot: "k", rkey: "r", workflow: "w", + buildNumber: 2, createdAt: "not a timestamp", + }, + }) + + s, err := openStore(path) + if err != nil { + // The whole point of swallowing parse errors is that + // migrate() must succeed regardless. If it returns an + // error here, the swallow logic regressed. + t.Fatalf("openStore must tolerate unparseable created_at: %v", err) + } + defer s.Close() + + values := map[string]int64{} + rows, err := s.db.Query(`SELECT build_uuid, created_unix_ns FROM buildkite_builds`) + if err != nil { + t.Fatalf("select: %v", err) + } + for rows.Next() { + var uuid string + var ns int64 + if err := rows.Scan(&uuid, &ns); err != nil { + t.Fatalf("scan: %v", err) + } + values[uuid] = ns + } + if err := rows.Err(); err != nil { + t.Fatalf("iterate: %v", err) + } + + if values["good"] != good.UnixNano() { + t.Errorf("good row: created_unix_ns = %d; want %d", values["good"], good.UnixNano()) + } + if values["garbage"] != 0 { + t.Errorf("garbage row: created_unix_ns = %d; want 0 (left at default)", values["garbage"]) + } +} diff --git a/store_test.go b/store_test.go index d979cdf..c07c977 100644 --- a/store_test.go +++ b/store_test.go @@ -12,6 +12,7 @@ import ( "context" "path/filepath" "testing" + "time" ) // newTestStore opens a fresh store in a per-test temp dir and registers @@ -284,6 +285,99 @@ func TestKnotsForSpindle(t *testing.T) { } } +// TestLookupBuildkiteBuildByTuplePicksMostRecent verifies that +// LookupBuildkiteBuildByTuple returns the build with the largest +// created_unix_ns even when text-ordering created_at would lie. +// +// The motivating bug: time.Format(RFC3339Nano) trims trailing zeros, +// so an exact-second instant renders as "...:00Z" while one nanosecond +// later renders as "...:00.000000001Z". Lex ordering puts ".000000001Z" +// *before* "Z" because '.' (0x2E) < 'Z' (0x5A), so ORDER BY created_at +// DESC would surface the *earlier* row as the latest build. Sorting +// on created_unix_ns avoids that and is what /logs depends on to +// resolve the right run. +func TestLookupBuildkiteBuildByTuplePicksMostRecent(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + // Two rows for the same (knot, rkey, workflow). The "newer" one + // is one nanosecond later but its RFC3339Nano text sorts BEFORE + // the older row's, which is exactly the failure mode we're + // guarding against. + older := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + newer := older.Add(time.Nanosecond) + + insert := func(uuid string, ts time.Time) { + t.Helper() + _, err := s.db.ExecContext(ctx, + `INSERT INTO buildkite_builds ( + build_uuid, build_number, pipeline_slug, org, + knot, pipeline_rkey, workflow, + pipeline_uri, created_at, created_unix_ns + ) VALUES (?, ?, ?, '', ?, ?, ?, ?, ?, ?)`, + uuid, int64(1), "p", + "k", "r", "w", "at://x", + ts.Format(time.RFC3339Nano), ts.UnixNano(), + ) + if err != nil { + t.Fatalf("insert %s: %v", uuid, err) + } + } + insert("older", older) + insert("newer", newer) + + ref, err := s.LookupBuildkiteBuildByTuple(ctx, "k", "r", "w") + if err != nil { + t.Fatalf("lookup: %v", err) + } + if ref == nil { + t.Fatal("lookup returned nil; want a row") + } + if ref.BuildUUID != "newer" { + t.Fatalf("lookup picked %q; want %q (text-ordering bug regression)", + ref.BuildUUID, "newer") + } +} + +// TestBuildkiteCreatedUnixNSBackfill simulates the upgrade path: rows +// inserted before the column existed (created_unix_ns = 0) get +// promoted to the parsed UnixNano of their created_at by the migration. +// Without the backfill the lookup would tie on 0 across all legacy +// rows and have to fall back to the (still-imperfect) text ordering. +func TestBuildkiteCreatedUnixNSBackfill(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + // Pretend this row was written by an older binary: created_at + // is set, created_unix_ns is the post-ALTER default 0. + want := time.Date(2026, 3, 4, 5, 6, 7, 8, time.UTC) + if _, err := s.db.ExecContext(ctx, + `INSERT INTO buildkite_builds ( + build_uuid, build_number, pipeline_slug, org, + knot, pipeline_rkey, workflow, + pipeline_uri, created_at, created_unix_ns + ) VALUES (?, 1, 'p', '', 'k', 'r', 'w', 'at://x', ?, 0)`, + "legacy", want.Format(time.RFC3339Nano), + ); err != nil { + t.Fatalf("seed legacy row: %v", err) + } + + if err := s.backfillBuildkiteCreatedUnixNS(ctx); err != nil { + t.Fatalf("backfill: %v", err) + } + + var got int64 + if err := s.db.QueryRowContext(ctx, + `SELECT created_unix_ns FROM buildkite_builds WHERE build_uuid = ?`, + "legacy", + ).Scan(&got); err != nil { + t.Fatalf("read back: %v", err) + } + if got != want.UnixNano() { + t.Fatalf("created_unix_ns = %d; want %d", got, want.UnixNano()) + } +} + // countRows is a small SELECT COUNT(*) helper used by lifecycle tests // to verify deletes actually removed the row. Table name is interpolated // directly because callers pass a constant from the schema, not user -- 2.51.2