From 7d727497ec652fff59e53381a502e9098b8becb8 Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Mon, 31 Aug 2026 15:36:37 +1000 Subject: [PATCH 1/4] fix(storage): canonicalize lock and check identity keys MySQL's ai_ci collation deduplicates mixed-case identity keys through idx_locks_database and idx_checks_check_key; PostgreSQL compares byte-wise, so the same event stream can double-book locks or duplicate check rows. Fold identity args Go-side at the store boundary (writes and predicates, including the lock-intent guard) as a backstop behind ingress canonicalization, with cross-dialect parity subtests. --- pkg/storage/canonical.go | 16 +++++++++ pkg/storage/internal/sqlstore/applies.go | 4 ++- pkg/storage/internal/sqlstore/checks.go | 28 ++++++++++++++++ pkg/storage/internal/sqlstore/locks.go | 19 +++++++++++ pkg/storage/storagetest/checks.go | 39 ++++++++++++++++++++++ pkg/storage/storagetest/locks.go | 41 ++++++++++++++++++++++++ 6 files changed, 146 insertions(+), 1 deletion(-) create mode 100644 pkg/storage/canonical.go diff --git a/pkg/storage/canonical.go b/pkg/storage/canonical.go new file mode 100644 index 000000000..0d02a5acf --- /dev/null +++ b/pkg/storage/canonical.go @@ -0,0 +1,16 @@ +package storage + +import "strings" + +// CanonicalKey folds an identity string — repository full name, database +// name, database type, environment — to its canonical single-spelling form. +// Identity strings are canonicalized at ingress (webhook payload extraction, +// API request decode, comment-command parsing) and backstopped at the store +// boundary, so byte-wise comparisons and unique indexes behave identically on +// MySQL and PostgreSQL: MySQL's utf8mb4_0900_ai_ci storage collation forgives +// case drift while PostgreSQL compares bytes. Folding is lowercase-only; +// accent folding is deliberately excluded because identity strings are ASCII +// in practice. +func CanonicalKey(s string) string { + return strings.ToLower(s) +} diff --git a/pkg/storage/internal/sqlstore/applies.go b/pkg/storage/internal/sqlstore/applies.go index ffa9e289e..a8793cb70 100644 --- a/pkg/storage/internal/sqlstore/applies.go +++ b/pkg/storage/internal/sqlstore/applies.go @@ -772,13 +772,15 @@ func verifyExpectedLockIntent(ctx context.Context, tx *rebindTx, apply *storage. // one: it matches only a lock whose pending_plan_id is unset (the column is // NOT NULL DEFAULT ''), so an unpinned lock that a rollback re-pins mid-flight // still fails this check. + database := storage.CanonicalKey(apply.Database) + databaseType := storage.CanonicalKey(apply.DatabaseType) var lockID int64 err := tx.QueryRowContext(ctx, ` SELECT id FROM locks WHERE database_name = ? AND database_type = ? AND owner = ? AND pending_plan_id = ? FOR UPDATE - `, apply.Database, apply.DatabaseType, apply.ExpectedLockOwner, apply.ExpectedPendingPlanID).Scan(&lockID) + `, database, databaseType, apply.ExpectedLockOwner, apply.ExpectedPendingPlanID).Scan(&lockID) if errors.Is(err, sql.ErrNoRows) { return storage.ErrLockIntentChanged } diff --git a/pkg/storage/internal/sqlstore/checks.go b/pkg/storage/internal/sqlstore/checks.go index 4687784b7..0f954a1e0 100644 --- a/pkg/storage/internal/sqlstore/checks.go +++ b/pkg/storage/internal/sqlstore/checks.go @@ -30,8 +30,19 @@ type checkStore struct { classifier ErrorClassifier } +func canonicalizeCheck(check *storage.Check) { + if check == nil { + return + } + check.Repository = storage.CanonicalKey(check.Repository) + check.Environment = storage.CanonicalKey(check.Environment) + check.DatabaseType = storage.CanonicalKey(check.DatabaseType) + check.DatabaseName = storage.CanonicalKey(check.DatabaseName) +} + // Upsert creates or updates stored check state. func (s *checkStore) Upsert(ctx context.Context, check *storage.Check) error { + canonicalizeCheck(check) // Convert CheckRunID=0 to NULL (0 is Go's zero value, not a valid check run ID) var checkRunID any if check.CheckRunID != 0 { @@ -82,6 +93,7 @@ func (s *checkStore) Upsert(ctx context.Context, check *storage.Check) error { // only a write that re-ran the rollup and found the deployments clean may clear // it. See storage.PlanDriftState. func (s *checkStore) UpsertPlanResult(ctx context.Context, check *storage.Check, drift storage.PlanDriftState) error { + canonicalizeCheck(check) var checkRunID any if check.CheckRunID != 0 { checkRunID = check.CheckRunID @@ -182,6 +194,7 @@ func (s *checkStore) UpsertPlanResult(ctx context.Context, check *storage.Check, // RecoverApplyOwnedCheckWithNoOpPlan updates same-head apply-owned stored check // state when a successful no-op plan proves the target already matches the PR schema. func (s *checkStore) RecoverApplyOwnedCheckWithNoOpPlan(ctx context.Context, check *storage.Check) (bool, error) { + canonicalizeCheck(check) if !successfulNoOpPlanResult(check) { return false, nil } @@ -240,6 +253,7 @@ func successfulNoOpPlanResult(check *storage.Check) bool { // plan is no longer part of the merge gate, so the plan-only drift block should // stop blocking. A started apply still owns the row and is left untouched. func (s *checkStore) MarkStalePlanSuccessful(ctx context.Context, check *storage.Check) (bool, error) { + canonicalizeCheck(check) var checkRunID any if check.CheckRunID != 0 { checkRunID = check.CheckRunID @@ -300,6 +314,7 @@ func (s *checkStore) MarkStalePlanSuccessful(ctx context.Context, check *storage // commit) does not match and is preserved. Returns true when the row was // cleared. func (s *checkStore) ClearAggregateBlock(ctx context.Context, check *storage.Check) (bool, error) { + canonicalizeCheck(check) result, err := s.db.ExecContext(ctx, ` UPDATE checks SET blocking_reason = '', @@ -335,6 +350,7 @@ func isPlanOnlySuccessful(check *storage.Check) bool { // CompleteForApply updates stored check state to a terminal state only if it // still belongs to the apply being completed. func (s *checkStore) CompleteForApply(ctx context.Context, check *storage.Check, apply *storage.Apply) (bool, error) { + canonicalizeCheck(check) var checkRunID any if check.CheckRunID != 0 { checkRunID = check.CheckRunID @@ -407,6 +423,7 @@ func (s *checkStore) CompleteForApply(ctx context.Context, check *storage.Check, // Cancelled forward applies additionally require that no completed forward // task exists under this or an earlier apply for the target. func (s *checkStore) MarkActionRequiredForApply(ctx context.Context, check *storage.Check, apply *storage.Apply) (bool, error) { + canonicalizeCheck(check) var checkRunID any if check.CheckRunID != 0 { checkRunID = check.CheckRunID @@ -483,6 +500,7 @@ func (s *checkStore) MarkActionRequiredForApply(ctx context.Context, check *stor // after the cancellation, and the completed-task predicate keeps a // cancellation that is safe to release from being retained here. func (s *checkStore) MarkCancelledApplyFailed(ctx context.Context, check *storage.Check, apply *storage.Apply) (bool, error) { + canonicalizeCheck(check) var checkRunID any if check.CheckRunID != 0 { checkRunID = check.CheckRunID @@ -568,6 +586,10 @@ func (s *checkStore) completedForwardTaskPredicate(required bool) string { // Get returns a check by its unique key (PR + env + database), or nil if not found. func (s *checkStore) Get(ctx context.Context, repo string, pr int, environment, dbType, database string) (*storage.Check, error) { + repo = storage.CanonicalKey(repo) + environment = storage.CanonicalKey(environment) + dbType = storage.CanonicalKey(dbType) + database = storage.CanonicalKey(database) row := s.db.QueryRowContext(ctx, ` SELECT `+checkColumns+` FROM checks @@ -591,6 +613,7 @@ func (s *checkStore) GetByCheckRunID(ctx context.Context, checkRunID int64) (*st // GetByPR returns all checks for a PR. func (s *checkStore) GetByPR(ctx context.Context, repo string, pr int) ([]*storage.Check, error) { + repo = storage.CanonicalKey(repo) rows, err := s.db.QueryContext(ctx, ` SELECT `+checkColumns+` FROM checks @@ -608,6 +631,10 @@ func (s *checkStore) GetByPR(ctx context.Context, repo string, pr int) ([]*stora // GetByDatabase returns all checks for a database across all PRs. // Used for cross-PR coordination (blocking other PRs when one is applying). func (s *checkStore) GetByDatabase(ctx context.Context, repo, environment, dbType, database string) ([]*storage.Check, error) { + repo = storage.CanonicalKey(repo) + environment = storage.CanonicalKey(environment) + dbType = storage.CanonicalKey(dbType) + database = storage.CanonicalKey(database) rows, err := s.db.QueryContext(ctx, ` SELECT `+checkColumns+` FROM checks @@ -665,6 +692,7 @@ func (s *checkStore) Delete(ctx context.Context, id int64) error { // action_required when the schema change is gone from the PR, or a fresh // plan result replaces it when the change is still present. func (s *checkStore) DeleteByPRRetainingBlockingApplyOwned(ctx context.Context, repo string, pr int, merged bool) error { + repo = storage.CanonicalKey(repo) if merged { _, err := s.db.ExecContext(ctx, ` DELETE FROM checks diff --git a/pkg/storage/internal/sqlstore/locks.go b/pkg/storage/internal/sqlstore/locks.go index 571d46978..81ef6c218 100644 --- a/pkg/storage/internal/sqlstore/locks.go +++ b/pkg/storage/internal/sqlstore/locks.go @@ -22,6 +22,12 @@ type lockStore struct { classifier ErrorClassifier } +func canonicalizeLock(lock *storage.Lock) { + lock.DatabaseName = storage.CanonicalKey(lock.DatabaseName) + lock.DatabaseType = storage.CanonicalKey(lock.DatabaseType) + lock.Repository = storage.CanonicalKey(lock.Repository) +} + // Acquire attempts to acquire a lock. Returns ErrLockHeld if held by another owner. // Acquiring a lock the same owner already holds is a success (idempotent): two // concurrent applies for the same PR and database (e.g. staging and production @@ -31,6 +37,7 @@ type lockStore struct { // confirm command loads, and its disclosure record travels with it. A re-acquire // that passes an empty PendingPlanID (CLI) leaves the existing values intact. func (s *lockStore) Acquire(ctx context.Context, lock *storage.Lock) error { + canonicalizeLock(lock) op := fmt.Sprintf("acquire lock for %s/%s owner=%s", lock.DatabaseName, lock.DatabaseType, lock.Owner) return withLockRetry(ctx, s.classifier, op, func() error { return s.acquireOnce(ctx, lock) @@ -41,6 +48,7 @@ func (s *lockStore) Acquire(ctx context.Context, lock *storage.Lock) error { // racing to claim the same key can hit a transient InnoDB lock conflict on the // INSERT below; Acquire retries those. func (s *lockStore) acquireOnce(ctx context.Context, lock *storage.Lock) error { + canonicalizeLock(lock) existing, err := s.Get(ctx, lock.DatabaseName, lock.DatabaseType) if err != nil { return fmt.Errorf("read existing lock for %s/%s: %w", lock.DatabaseName, lock.DatabaseType, err) @@ -104,6 +112,7 @@ func (s *lockStore) acquireOnce(ctx context.Context, lock *storage.Lock) error { // case, so the refresh has succeeded. To distinguish that from a genuine ownership // change, re-read the lock and branch on its actual state. func (s *lockStore) refreshPendingConfirmation(ctx context.Context, lock, existing *storage.Lock) error { + canonicalizeLock(lock) if lock.PendingPlanID == "" || lock.PendingPlanID == existing.PendingPlanID { return nil } @@ -143,6 +152,8 @@ func (s *lockStore) refreshPendingConfirmation(ctx context.Context, lock, existi // Release releases a lock. Only succeeds if caller is the owner. func (s *lockStore) Release(ctx context.Context, database, dbType, owner string) error { + database = storage.CanonicalKey(database) + dbType = storage.CanonicalKey(dbType) result, err := s.db.ExecContext(ctx, ` DELETE FROM locks WHERE database_name = ? AND database_type = ? AND owner = ? @@ -172,6 +183,8 @@ func (s *lockStore) Release(ctx context.Context, database, dbType, owner string) // apply lock can become a rollback lock), so owner-only release is insufficient // after a network call or other long-running operation. func (s *lockStore) ReleaseIfPendingPlanID(ctx context.Context, database, dbType, owner, pendingPlanID string) (bool, error) { + database = storage.CanonicalKey(database) + dbType = storage.CanonicalKey(dbType) result, err := s.db.ExecContext(ctx, ` DELETE FROM locks WHERE database_name = ? AND database_type = ? AND owner = ? AND pending_plan_id = ? @@ -189,6 +202,8 @@ func (s *lockStore) ReleaseIfPendingPlanID(ctx context.Context, database, dbType // ForceRelease releases a lock regardless of owner (admin override). func (s *lockStore) ForceRelease(ctx context.Context, database, dbType string) error { + database = storage.CanonicalKey(database) + dbType = storage.CanonicalKey(dbType) result, err := s.db.ExecContext(ctx, ` DELETE FROM locks WHERE database_name = ? AND database_type = ? @@ -202,6 +217,8 @@ func (s *lockStore) ForceRelease(ctx context.Context, database, dbType string) e // Get returns a lock by database name and type, or nil if not found. func (s *lockStore) Get(ctx context.Context, database, dbType string) (*storage.Lock, error) { + database = storage.CanonicalKey(database) + dbType = storage.CanonicalKey(dbType) row := s.db.QueryRowContext(ctx, ` SELECT `+lockColumns+` FROM locks @@ -240,6 +257,7 @@ func (s *lockStore) List(ctx context.Context) ([]*storage.Lock, error) { // ErrLockNotFound when it is gone or ErrLockNotOwned when another owner holds // it. func (s *lockStore) Update(ctx context.Context, lock *storage.Lock) error { + canonicalizeLock(lock) result, err := s.db.ExecContext(ctx, ` UPDATE locks SET updated_at = NOW() @@ -273,6 +291,7 @@ func (s *lockStore) Update(ctx context.Context, lock *storage.Lock) error { // GetByPR returns all locks associated with a PR. func (s *lockStore) GetByPR(ctx context.Context, repo string, pr int) ([]*storage.Lock, error) { + repo = storage.CanonicalKey(repo) rows, err := s.db.QueryContext(ctx, ` SELECT `+lockColumns+` FROM locks diff --git a/pkg/storage/storagetest/checks.go b/pkg/storage/storagetest/checks.go index 6fa9e3627..e4e464b10 100644 --- a/pkg/storage/storagetest/checks.go +++ b/pkg/storage/storagetest/checks.go @@ -15,6 +15,45 @@ import ( // upserts, aggregate and per-database row isolation, apply ownership, and // review-time deployment drift disposition. func TestChecks(t *testing.T, h Harness) { + t.Run("CanonicalIdentityKey", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + check := &storage.Check{ + Repository: "MixedCase/Sample-Repo", PullRequest: 123, HeadSHA: "abc123", + Environment: "Staging", DatabaseType: "MySQL", DatabaseName: "Orders_DB", + Status: "completed", Conclusion: "success", + } + require.NoError(t, store.Checks().Upsert(ctx, check)) + + stored, err := store.Checks().Get(ctx, "MIXEDCASE/SAMPLE-REPO", 123, "STAGING", "MYSQL", "ORDERS_DB") + require.NoError(t, err) + require.NotNil(t, stored) + assert.Equal(t, "mixedcase/sample-repo", stored.Repository) + assert.Equal(t, "staging", stored.Environment) + assert.Equal(t, "mysql", stored.DatabaseType) + assert.Equal(t, "orders_db", stored.DatabaseName) + + check.Repository = "MIXEDCASE/SAMPLE-REPO" + check.Environment = "STAGING" + check.DatabaseType = "MYSQL" + check.DatabaseName = "ORDERS_DB" + check.HeadSHA = "def456" + require.NoError(t, store.Checks().Upsert(ctx, check)) + + checks, err := store.Checks().GetByDatabase(ctx, "MixedCase/Sample-Repo", "Staging", "MySQL", "Orders_DB") + require.NoError(t, err) + require.Len(t, checks, 1) + assert.Equal(t, "def456", checks[0].HeadSHA) + + checks, err = store.Checks().GetByPR(ctx, "MIXEDCASE/SAMPLE-REPO", 123) + require.NoError(t, err) + require.Len(t, checks, 1) + require.NoError(t, store.Checks().DeleteByPRRetainingBlockingApplyOwned(ctx, "MixedCase/Sample-Repo", 123, false)) + checks, err = store.Checks().GetByPR(ctx, "mixedcase/sample-repo", 123) + require.NoError(t, err) + assert.Empty(t, checks) + }) + t.Run("Upsert", func(t *testing.T) { ctx := t.Context() store := h.NewStorage(t) diff --git a/pkg/storage/storagetest/locks.go b/pkg/storage/storagetest/locks.go index edbc4d220..017bbb978 100644 --- a/pkg/storage/storagetest/locks.go +++ b/pkg/storage/storagetest/locks.go @@ -16,6 +16,47 @@ import ( // requires backdating the stored row via SQL, which is dialect-specific and // lives in each implementation's own integration tests. func TestLocks(t *testing.T, h Harness) { + t.Run("CanonicalIdentityKey", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + lock := &storage.Lock{ + DatabaseName: "Orders_DB", + DatabaseType: "MySQL", + Repository: "MixedCase/Sample-Repo", + PullRequest: 123, + Owner: "MixedCase/Sample-Repo#123", + PendingPlanID: "plan-1", + } + require.NoError(t, store.Locks().Acquire(ctx, lock)) + + stored, err := store.Locks().Get(ctx, "ORDERS_DB", "MYSQL") + require.NoError(t, err) + require.NotNil(t, stored) + assert.Equal(t, "orders_db", stored.DatabaseName) + assert.Equal(t, "mysql", stored.DatabaseType) + assert.Equal(t, "mixedcase/sample-repo", stored.Repository) + assert.Equal(t, "MixedCase/Sample-Repo#123", stored.Owner) + + locks, err := store.Locks().GetByPR(ctx, "MIXEDCASE/SAMPLE-REPO", 123) + require.NoError(t, err) + require.Len(t, locks, 1) + + require.NoError(t, store.Locks().Acquire(ctx, &storage.Lock{ + DatabaseName: "orders_db", DatabaseType: "mysql", + Repository: "mixedcase/sample-repo", PullRequest: 123, + Owner: "MixedCase/Sample-Repo#123", + })) + require.ErrorIs(t, store.Locks().Acquire(ctx, &storage.Lock{ + DatabaseName: "ORDERS_DB", DatabaseType: "MYSQL", + Repository: "MIXEDCASE/SAMPLE-REPO", PullRequest: 123, + Owner: "different-owner", + }), storage.ErrLockHeld) + + released, err := store.Locks().ReleaseIfPendingPlanID(ctx, "ORDERS_DB", "MYSQL", "MixedCase/Sample-Repo#123", "plan-1") + require.NoError(t, err) + assert.True(t, released) + }) + t.Run("Acquire", func(t *testing.T) { ctx := t.Context() store := h.NewStorage(t) From 85d3035f1f725165b433400441a554ebcecc840c Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Mon, 31 Aug 2026 16:59:51 +1000 Subject: [PATCH 2/4] fix(storage): single canonicalization boundary for lock acquire Acquire is now the one fold point on its call chain (internal helpers no longer refold), canonicalizeLock gains the same nil guard as its twin, and the in-place identity canonicalization contract is documented on the store godocs. Addresses external review of pull/1216. --- pkg/storage/canonical.go | 17 ++++++++++------- pkg/storage/internal/sqlstore/locks.go | 8 +++++--- pkg/storage/storage.go | 2 ++ 3 files changed, 17 insertions(+), 10 deletions(-) diff --git a/pkg/storage/canonical.go b/pkg/storage/canonical.go index 0d02a5acf..cf893ba4f 100644 --- a/pkg/storage/canonical.go +++ b/pkg/storage/canonical.go @@ -4,13 +4,16 @@ import "strings" // CanonicalKey folds an identity string — repository full name, database // name, database type, environment — to its canonical single-spelling form. -// Identity strings are canonicalized at ingress (webhook payload extraction, -// API request decode, comment-command parsing) and backstopped at the store -// boundary, so byte-wise comparisons and unique indexes behave identically on -// MySQL and PostgreSQL: MySQL's utf8mb4_0900_ai_ci storage collation forgives -// case drift while PostgreSQL compares bytes. Folding is lowercase-only; -// accent folding is deliberately excluded because identity strings are ASCII -// in practice. +// Identity strings are the cross-dialect row-identity keys: MySQL's +// utf8mb4_0900_ai_ci storage collation forgives case drift while PostgreSQL +// compares bytes, so every boundary that accepts an identity string folds it +// before matching or persisting. Folding is lowercase-only; accent folding is +// deliberately excluded because identity strings are ASCII in practice. +// +// Deliberately not folded: lock owner strings (audit metadata such as +// "Org/Repo#42", compared byte-wise on both sides of every ownership check), +// deployment names, table names, SHAs, and GitHub node IDs — these are either +// case-significant or opaque values, not row-identity keys. func CanonicalKey(s string) string { return strings.ToLower(s) } diff --git a/pkg/storage/internal/sqlstore/locks.go b/pkg/storage/internal/sqlstore/locks.go index 81ef6c218..21a686c1f 100644 --- a/pkg/storage/internal/sqlstore/locks.go +++ b/pkg/storage/internal/sqlstore/locks.go @@ -23,6 +23,9 @@ type lockStore struct { } func canonicalizeLock(lock *storage.Lock) { + if lock == nil { + return + } lock.DatabaseName = storage.CanonicalKey(lock.DatabaseName) lock.DatabaseType = storage.CanonicalKey(lock.DatabaseType) lock.Repository = storage.CanonicalKey(lock.Repository) @@ -46,9 +49,8 @@ func (s *lockStore) Acquire(ctx context.Context, lock *storage.Lock) error { // acquireOnce performs a single claim attempt. Concurrent same-owner callers // racing to claim the same key can hit a transient InnoDB lock conflict on the -// INSERT below; Acquire retries those. +// INSERT below; Acquire retries those. Acquire canonicalizes the lock first. func (s *lockStore) acquireOnce(ctx context.Context, lock *storage.Lock) error { - canonicalizeLock(lock) existing, err := s.Get(ctx, lock.DatabaseName, lock.DatabaseType) if err != nil { return fmt.Errorf("read existing lock for %s/%s: %w", lock.DatabaseName, lock.DatabaseType, err) @@ -111,8 +113,8 @@ func (s *lockStore) acquireOnce(ctx context.Context, lock *storage.Lock) error { // between this caller's read and its write. The owner still holds the lock in that // case, so the refresh has succeeded. To distinguish that from a genuine ownership // change, re-read the lock and branch on its actual state. +// Acquire canonicalizes the lock before reaching this helper. func (s *lockStore) refreshPendingConfirmation(ctx context.Context, lock, existing *storage.Lock) error { - canonicalizeLock(lock) if lock.PendingPlanID == "" || lock.PendingPlanID == existing.PendingPlanID { return nil } diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index e2e1cf4b2..30829c82c 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -112,6 +112,7 @@ type Storage interface { // Locks prevent concurrent schema changes to the same database. // Lock key is database:type (not per-environment) to block concurrent changes // across environments and PRs. +// Methods accepting a *Lock canonicalize its repository, database name, and database type in place before persisting. type LockStore interface { // Acquire attempts to acquire a lock. Returns ErrLockHeld if already held by another owner. // If the same owner already holds the lock, this is a no-op (idempotent). @@ -150,6 +151,7 @@ type LockStore interface { // CheckStore manages SchemaBot's stored check state. // Per-database rows track internal status for a PR/environment/database. // Aggregate rows store the GitHub check_run_id for the visible GitHub Check Run. +// Methods accepting a *Check canonicalize its repository, database name, database type, and environment in place before persisting. type CheckStore interface { // Upsert creates or updates stored check state. Upsert(ctx context.Context, check *Check) error From b0458d0377f8039eb3b8f83b75092d38d75b300a Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Tue, 1 Sep 2026 11:59:08 +1000 Subject: [PATCH 3/4] fix(storage): fold apply-target lock name components before hashing The sha256-derived advisory lock name is byte-sensitive on both dialects, so case drift in database, type, or environment would defeat mutual exclusion across differently-spelled callers. --- pkg/storage/canonical.go | 13 +++++++---- pkg/storage/internal/sqlstore/applies.go | 4 ++++ .../sqlstore/apply_target_lock_test.go | 8 +++++++ pkg/storage/storagetest/locks.go | 23 ++++++++++--------- 4 files changed, 33 insertions(+), 15 deletions(-) diff --git a/pkg/storage/canonical.go b/pkg/storage/canonical.go index cf893ba4f..3cd960951 100644 --- a/pkg/storage/canonical.go +++ b/pkg/storage/canonical.go @@ -10,10 +10,15 @@ import "strings" // before matching or persisting. Folding is lowercase-only; accent folding is // deliberately excluded because identity strings are ASCII in practice. // -// Deliberately not folded: lock owner strings (audit metadata such as -// "Org/Repo#42", compared byte-wise on both sides of every ownership check), -// deployment names, table names, SHAs, and GitHub node IDs — these are either -// case-significant or opaque values, not row-identity keys. +// Deliberately not folded here: lock owner strings ("org/repo#42") are the +// ownership predicate on lock acquire, release, and intent verification, but +// the repository they are derived from is folded at ingress, so owners are +// canonical by construction rather than re-folded at this boundary. +// Deployment names are identity keys too, but they are operator-controlled +// configuration that doubles as routing and schema-directory path components, +// so config validation rejects non-canonical spellings instead of silently +// rewriting them. Table names, SHAs, and GitHub node IDs are case-significant +// or opaque values, not row-identity keys. func CanonicalKey(s string) string { return strings.ToLower(s) } diff --git a/pkg/storage/internal/sqlstore/applies.go b/pkg/storage/internal/sqlstore/applies.go index a8793cb70..1e86153e5 100644 --- a/pkg/storage/internal/sqlstore/applies.go +++ b/pkg/storage/internal/sqlstore/applies.go @@ -229,6 +229,10 @@ func (w *applyWriteTx) commit() error { } func applyTargetLockName(database, dbType, environment string) string { + // Fold every component so caller spelling cannot change the derived lock name. + database = storage.CanonicalKey(database) + dbType = storage.CanonicalKey(dbType) + environment = storage.CanonicalKey(environment) sum := sha256.Sum256([]byte(database + "\x00" + dbType + "\x00" + environment)) return "schemabot_apply_" + hex.EncodeToString(sum[:16]) } diff --git a/pkg/storage/internal/sqlstore/apply_target_lock_test.go b/pkg/storage/internal/sqlstore/apply_target_lock_test.go index 4f5f6a5f4..f524824d3 100644 --- a/pkg/storage/internal/sqlstore/apply_target_lock_test.go +++ b/pkg/storage/internal/sqlstore/apply_target_lock_test.go @@ -3,9 +3,17 @@ package sqlstore import ( "testing" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +func TestApplyTargetLockName_CanonicalComponents(t *testing.T) { + lowercase := applyTargetLockName("orders_db", "mysql", "staging") + mixedCase := applyTargetLockName("Orders_DB", "MySQL", "Staging") + + assert.Equal(t, lowercase, mixedCase) +} + // TestAcquireApplyTargetLockConn_NilLockerFailsClosed verifies an apply // target without an advisory-lock implementation is rejected before any // connection is opened: applies must not proceed unserialized across diff --git a/pkg/storage/storagetest/locks.go b/pkg/storage/storagetest/locks.go index 017bbb978..d6e055884 100644 --- a/pkg/storage/storagetest/locks.go +++ b/pkg/storage/storagetest/locks.go @@ -22,9 +22,9 @@ func TestLocks(t *testing.T, h Harness) { lock := &storage.Lock{ DatabaseName: "Orders_DB", DatabaseType: "MySQL", - Repository: "MixedCase/Sample-Repo", - PullRequest: 123, - Owner: "MixedCase/Sample-Repo#123", + Repository: "Org/Repo", + PullRequest: 42, + Owner: "Org/Repo#42", PendingPlanID: "plan-1", } require.NoError(t, store.Locks().Acquire(ctx, lock)) @@ -34,25 +34,26 @@ func TestLocks(t *testing.T, h Harness) { require.NotNil(t, stored) assert.Equal(t, "orders_db", stored.DatabaseName) assert.Equal(t, "mysql", stored.DatabaseType) - assert.Equal(t, "mixedcase/sample-repo", stored.Repository) - assert.Equal(t, "MixedCase/Sample-Repo#123", stored.Owner) + assert.Equal(t, "org/repo", stored.Repository) + assert.Equal(t, "Org/Repo#42", stored.Owner) - locks, err := store.Locks().GetByPR(ctx, "MIXEDCASE/SAMPLE-REPO", 123) + locks, err := store.Locks().GetByPR(ctx, "ORG/REPO", 42) require.NoError(t, err) require.Len(t, locks, 1) require.NoError(t, store.Locks().Acquire(ctx, &storage.Lock{ DatabaseName: "orders_db", DatabaseType: "mysql", - Repository: "mixedcase/sample-repo", PullRequest: 123, - Owner: "MixedCase/Sample-Repo#123", + Repository: "org/repo", PullRequest: 42, + Owner: "Org/Repo#42", })) require.ErrorIs(t, store.Locks().Acquire(ctx, &storage.Lock{ DatabaseName: "ORDERS_DB", DatabaseType: "MYSQL", - Repository: "MIXEDCASE/SAMPLE-REPO", PullRequest: 123, - Owner: "different-owner", + Repository: "ORG/REPO", PullRequest: 42, + Owner: "org/repo#42", }), storage.ErrLockHeld) + require.ErrorIs(t, store.Locks().Release(ctx, "ORDERS_DB", "MYSQL", "org/repo#42"), storage.ErrLockNotOwned) - released, err := store.Locks().ReleaseIfPendingPlanID(ctx, "ORDERS_DB", "MYSQL", "MixedCase/Sample-Repo#123", "plan-1") + released, err := store.Locks().ReleaseIfPendingPlanID(ctx, "ORDERS_DB", "MYSQL", "Org/Repo#42", "plan-1") require.NoError(t, err) assert.True(t, released) }) From fd2306ee3538295e18a93ab8d63c43c55050cb6d Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Tue, 1 Sep 2026 12:45:30 +1000 Subject: [PATCH 4/4] fix(storage): compare lock owners byte-wise on every dialect MySQL's accent- and case-insensitive default collation let a differently-cased owner release or touch a lock it does not hold, diverging from the byte-wise ownership contract the acquire path and PostgreSQL already enforce. --- pkg/storage/internal/sqlstore/dialect.go | 19 +++++++++++++++++++ pkg/storage/internal/sqlstore/dialect_test.go | 8 ++++++++ pkg/storage/internal/sqlstore/locks.go | 9 +++++---- pkg/storage/internal/sqlstore/locks_test.go | 6 +++--- pkg/storage/internal/sqlstore/storage.go | 2 +- pkg/storage/storagetest/locks.go | 10 +++++++++- 6 files changed, 45 insertions(+), 9 deletions(-) diff --git a/pkg/storage/internal/sqlstore/dialect.go b/pkg/storage/internal/sqlstore/dialect.go index 233c1f77f..632fcb858 100644 --- a/pkg/storage/internal/sqlstore/dialect.go +++ b/pkg/storage/internal/sqlstore/dialect.go @@ -53,6 +53,12 @@ type Dialect interface { // placed immediately after a table name or alias; dialects without index-hint // syntax return an empty string. IndexHint(index string) string + // BinaryEquals returns a predicate that is true only when column's stored + // value equals the single bound placeholder byte-for-byte. Opaque identity + // values such as lock owners compare byte-wise by contract, so dialects + // whose default column collation folds case or accents must force a binary + // comparison rather than inherit the column's collation. + BinaryEquals(column string) string // JoinedUpdate returns an UPDATE that changes target rows selected through a // join. Aliases qualify join and predicate expressions, while assignments to // target columns must be unqualified so the statement is valid across @@ -252,6 +258,13 @@ func (MySQLDialect) IndexHint(index string) string { return " FORCE INDEX (`" + strings.ReplaceAll(index, "`", "``") + "`)" } +// BinaryEquals compares under the binary collation: the schema's table default +// is accent- and case-insensitive utf8mb4_0900_ai_ci, under which +// differently-spelled opaque values would match. +func (MySQLDialect) BinaryEquals(column string) string { + return column + " COLLATE utf8mb4_0900_bin = ?" +} + // JoinedUpdate builds a MySQL multi-table UPDATE statement. func (MySQLDialect) JoinedUpdate(targetTable, targetAlias, joinTable, joinAlias, joinCondition string, assignments []JoinedUpdateAssignment, predicate string) string { if len(assignments) == 0 { @@ -445,6 +458,12 @@ func (PostgresDialect) JSONBooleanIsTrue(expression string, path []string) strin // relies on the planner to choose the access path. func (PostgresDialect) IndexHint(string) string { return "" } +// BinaryEquals returns plain equality: PostgreSQL's deterministic collations +// already compare equality byte-for-byte. +func (PostgresDialect) BinaryEquals(column string) string { + return column + " = ?" +} + // JoinedUpdate builds a PostgreSQL UPDATE … FROM statement. The join condition // moves into the WHERE clause alongside the residual predicate; SET // placeholders still precede predicate placeholders, so the placeholder-free diff --git a/pkg/storage/internal/sqlstore/dialect_test.go b/pkg/storage/internal/sqlstore/dialect_test.go index da774bcaa..e0ebaa841 100644 --- a/pkg/storage/internal/sqlstore/dialect_test.go +++ b/pkg/storage/internal/sqlstore/dialect_test.go @@ -33,6 +33,10 @@ func TestMySQLDialectIndexHint(t *testing.T) { assert.Equal(t, " FORCE INDEX (`idx_database_env_deployment`)", MySQLDialect{}.IndexHint("idx_database_env_deployment")) } +func TestMySQLDialectBinaryEquals(t *testing.T) { + assert.Equal(t, "owner COLLATE utf8mb4_0900_bin = ?", MySQLDialect{}.BinaryEquals("owner")) +} + func TestMySQLDialectJoinedUpdate(t *testing.T) { assert.Equal(t, "UPDATE apply_comments c JOIN applies a ON a.id = c.apply_id SET c.edit_count = c.edit_count + 1, c.updated_at = NOW() WHERE c.apply_id = ? AND a.lease_token = ?", @@ -293,6 +297,10 @@ func TestPostgresDialect(t *testing.T) { d.RelativeTime(TimestampPrecisionDefault, AfterCurrentTime, LiteralIntervalAmount(2), IntervalDay)) } +func TestPostgresDialectBinaryEquals(t *testing.T) { + assert.Equal(t, "owner = ?", PostgresDialect{}.BinaryEquals("owner")) +} + func TestPostgresDialectInsertIfAbsent(t *testing.T) { assert.Equal(t, InsertIfAbsentSyntax{Suffix: " ON CONFLICT (apply_id, comment_state) DO NOTHING"}, diff --git a/pkg/storage/internal/sqlstore/locks.go b/pkg/storage/internal/sqlstore/locks.go index 21a686c1f..d8c649664 100644 --- a/pkg/storage/internal/sqlstore/locks.go +++ b/pkg/storage/internal/sqlstore/locks.go @@ -19,6 +19,7 @@ const lockColumns = `id, database_name, database_type, repository, pull_request, // lockStore implements storage.LockStore using MySQL. type lockStore struct { db *rebindDB + dialect Dialect classifier ErrorClassifier } @@ -121,7 +122,7 @@ func (s *lockStore) refreshPendingConfirmation(ctx context.Context, lock, existi result, err := s.db.ExecContext(ctx, ` UPDATE locks SET pending_plan_id = ?, disclosed_copy_discard = ?, updated_at = NOW() - WHERE database_name = ? AND database_type = ? AND owner = ? + WHERE database_name = ? AND database_type = ? AND `+s.dialect.BinaryEquals("owner")+` `, lock.PendingPlanID, lock.DisclosedCopyDiscard, lock.DatabaseName, lock.DatabaseType, lock.Owner) if err != nil { return fmt.Errorf("refresh pending confirmation for %s/%s owner=%s: %w", @@ -158,7 +159,7 @@ func (s *lockStore) Release(ctx context.Context, database, dbType, owner string) dbType = storage.CanonicalKey(dbType) result, err := s.db.ExecContext(ctx, ` DELETE FROM locks - WHERE database_name = ? AND database_type = ? AND owner = ? + WHERE database_name = ? AND database_type = ? AND `+s.dialect.BinaryEquals("owner")+` `, database, dbType, owner) if err != nil { return err @@ -189,7 +190,7 @@ func (s *lockStore) ReleaseIfPendingPlanID(ctx context.Context, database, dbType dbType = storage.CanonicalKey(dbType) result, err := s.db.ExecContext(ctx, ` DELETE FROM locks - WHERE database_name = ? AND database_type = ? AND owner = ? AND pending_plan_id = ? + WHERE database_name = ? AND database_type = ? AND `+s.dialect.BinaryEquals("owner")+` AND pending_plan_id = ? `, database, dbType, owner, pendingPlanID) if err != nil { return false, err @@ -263,7 +264,7 @@ func (s *lockStore) Update(ctx context.Context, lock *storage.Lock) error { result, err := s.db.ExecContext(ctx, ` UPDATE locks SET updated_at = NOW() - WHERE database_name = ? AND database_type = ? AND owner = ? + WHERE database_name = ? AND database_type = ? AND `+s.dialect.BinaryEquals("owner")+` `, lock.DatabaseName, lock.DatabaseType, lock.Owner) if err != nil { return fmt.Errorf("touch lock for %s/%s: %w", lock.DatabaseName, lock.DatabaseType, err) diff --git a/pkg/storage/internal/sqlstore/locks_test.go b/pkg/storage/internal/sqlstore/locks_test.go index 10d567036..471922071 100644 --- a/pkg/storage/internal/sqlstore/locks_test.go +++ b/pkg/storage/internal/sqlstore/locks_test.go @@ -89,7 +89,7 @@ func TestLockStore_Acquire_RefreshSameOwnerValueAlreadyMatches(t *testing.T) { require.NoError(t, db.Close()) }) require.NoError(t, db.PingContext(ctx)) - store := &lockStore{db: newRebindDB(db, MySQLDialect{}), classifier: NewMySQLErrorClassifier()} + store := &lockStore{db: newRebindDB(db, MySQLDialect{}), dialect: MySQLDialect{}, classifier: NewMySQLErrorClassifier()} require.NoError(t, store.Acquire(ctx, &storage.Lock{ DatabaseName: "testdb", @@ -150,7 +150,7 @@ func TestLockStore_Acquire_RefreshSameOwnerValueAlreadyMatches(t *testing.T) { func TestLockStore_Acquire_RefreshOwnerNoLongerMatches(t *testing.T) { clearTables(t) ctx := t.Context() - store := &lockStore{db: newRebindDB(testDB, MySQLDialect{}), classifier: NewMySQLErrorClassifier()} + store := &lockStore{db: newRebindDB(testDB, MySQLDialect{}), dialect: MySQLDialect{}, classifier: NewMySQLErrorClassifier()} require.NoError(t, store.Acquire(ctx, &storage.Lock{ DatabaseName: "testdb", @@ -280,7 +280,7 @@ func TestLockStore_UpdateSameSecondSucceeds(t *testing.T) { _, err = db.ExecContext(ctx, "SET TIMESTAMP = 1700000000") require.NoError(t, err) - store := &lockStore{db: newRebindDB(db, MySQLDialect{}), classifier: NewMySQLErrorClassifier()} + store := &lockStore{db: newRebindDB(db, MySQLDialect{}), dialect: MySQLDialect{}, classifier: NewMySQLErrorClassifier()} // Acquire seeds the row via the locks table's DEFAULT CURRENT_TIMESTAMP, // which resolves to the frozen NOW(). diff --git a/pkg/storage/internal/sqlstore/storage.go b/pkg/storage/internal/sqlstore/storage.go index 64b15c87c..a7c107439 100644 --- a/pkg/storage/internal/sqlstore/storage.go +++ b/pkg/storage/internal/sqlstore/storage.go @@ -85,7 +85,7 @@ func NewWithDependencies(deps Dependencies) *Storage { } return &Storage{ db: rdb, - locks: &lockStore{db: rdb, classifier: deps.Classifier}, + locks: &lockStore{db: rdb, dialect: deps.Dialect, classifier: deps.Classifier}, plans: &planStore{db: rdb, identity: deps.Identity, classifier: deps.Classifier}, applies: &applyStore{db: rdb, dialect: deps.Dialect, identity: deps.Identity, locker: deps.Locker, classifier: deps.Classifier}, tasks: &taskStore{db: rdb, dialect: deps.Dialect, identity: deps.Identity, locker: deps.Locker}, diff --git a/pkg/storage/storagetest/locks.go b/pkg/storage/storagetest/locks.go index d6e055884..1eb000dd0 100644 --- a/pkg/storage/storagetest/locks.go +++ b/pkg/storage/storagetest/locks.go @@ -52,8 +52,16 @@ func TestLocks(t *testing.T, h Harness) { Owner: "org/repo#42", }), storage.ErrLockHeld) require.ErrorIs(t, store.Locks().Release(ctx, "ORDERS_DB", "MYSQL", "org/repo#42"), storage.ErrLockNotOwned) + require.ErrorIs(t, store.Locks().Update(ctx, &storage.Lock{ + DatabaseName: "ORDERS_DB", DatabaseType: "MYSQL", + Owner: "org/repo#42", + }), storage.ErrLockNotOwned) + + released, err := store.Locks().ReleaseIfPendingPlanID(ctx, "ORDERS_DB", "MYSQL", "org/repo#42", "plan-1") + require.NoError(t, err) + assert.False(t, released) - released, err := store.Locks().ReleaseIfPendingPlanID(ctx, "ORDERS_DB", "MYSQL", "Org/Repo#42", "plan-1") + released, err = store.Locks().ReleaseIfPendingPlanID(ctx, "ORDERS_DB", "MYSQL", "Org/Repo#42", "plan-1") require.NoError(t, err) assert.True(t, released) })