dbconn: scope the table lock holder lookup to the session's own database

docs/invariants.md

@@ -604,7 +604,11 @@ Migrations serialize per table via a **session-scoped advisory lock** — the an
 *Enforced today:* `pkg/dbconn` `AcquireTableLock` (`TableLock` proof carried by
 `TableLockSession`; affinity refusal behind transaction-mode and statement-mode PgBouncer,
 two-instance refusal, keepalive-loss, and `Bind`
-cancellation tests); `pkg/schemachange` shadow build, drop, and inspect each require the
+cancellation tests), and `TableLockSession.Confirm` and `LookupTableLockHolder` read
+`pg_locks` scoped to the connection's own database, so a lock the session holds in another
+database of the same server — where it excludes nothing on this database's table — confirms
+as `ErrTableLockNotHeld` and is not reported as the table's holder (same-key-in-another-database
+test); `pkg/schemachange` shadow build, drop, and inspect each require the
 session for the proven table, run under its `Bind` context, and confirm from their own
 transaction that the session's backend holds the lock before the first write (nil-session,
 wrong-table, reported-loss, gone-session, rival-backend, mid-build-loss, and mid-drop-loss

pkg/dbconn/table_lock.go

@@ -232,13 +232,24 @@ func LookupTableLockHolder(ctx context.Context, conn AdvisoryLockHolder, schema,
 	return lookupTableLockHolder(ctx, conn, TableLockKey{ClassID: TableLockClassID, ObjID: objID})
 }
 
+// lookupTableLockHolder reads the holder of key from pg_locks on conn's
+// connection, scoped to conn's own database. pg_locks is cluster-wide and
+// an advisory lock is per database, so the same key may be granted in
+// several databases of one server at once; only the grant in the database
+// conn is connected to is the lock on this table.
 func lookupTableLockHolder(ctx context.Context, conn AdvisoryLockHolder, key TableLockKey) (holder TableLockHolder, found bool, err error) {
 	ref := pairAdvisoryLock(key)
+	// INV: LK-1 — the database predicate is what makes a confirmation from
+	// a working connection mean "held in the database this transaction
+	// writes to": without it, a lock session dialled into another database
+	// of the same server would confirm the lock it holds there.
 	const query = `SELECT l.pid, COALESCE(a.application_name, ''), a.backend_start, COALESCE(a.state, '')
 		  FROM pg_catalog.pg_locks l
 		  LEFT JOIN pg_catalog.pg_stat_activity a ON a.pid = l.pid
 		 WHERE l.locktype = 'advisory'
 		   AND l.granted
+		   AND l.database = (SELECT d.oid FROM pg_catalog.pg_database d
+		                      WHERE d.datname = pg_catalog.current_database())
 		   AND l.objsubid = $1
 		   AND l.classid::bigint = $2
 		   AND l.objid::bigint = $3

pkg/dbconn/table_lock_confirm_integration_test.go

@@ -56,6 +56,48 @@ func TestTableLockConfirmDistinguishesHolderGoneFromHolderElsewhere(t *testing.T
 	assert.NoError(t, rival.Confirm(t.Context(), tx), "the rival confirms its own lock on the same connection")
 }
 
+// An advisory lock is per database, but pg_locks lists every database's
+// grants. A lock session dialled into another database of the same server
+// holds the key there, not in the database the working transaction writes
+// to, so Confirm must report the table unheld; a lookup from the working
+// database must not find that session either, while a lookup from the lock's
+// own database does.
+func TestTableLockConfirmIgnoresTheSameKeyHeldInAnotherDatabase(t *testing.T) {
+	serverURL := testutil.StartPostgres(t)
+	lockURL := testutil.NewDatabase(t, serverURL)
+	workURL := testutil.NewDatabase(t, serverURL)
+	workPool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: workURL})
+	require.NoError(t, err)
+	t.Cleanup(workPool.Close)
+	lockPool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: lockURL})
+	require.NoError(t, err)
+	t.Cleanup(lockPool.Close)
+
+	held, err := dbconn.AcquireTableLock(t.Context(), dbconn.Config{URL: lockURL}, "app", "orders")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		assert.NoError(t, held.Release(context.WithoutCancel(t.Context())))
+	})
+
+	tx, err := workPool.Begin(t.Context())
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		assert.NoError(t, tx.Rollback(context.WithoutCancel(t.Context())))
+	})
+	err = held.Confirm(t.Context(), tx)
+	assert.ErrorIs(t, err, dbconn.ErrTableLockNotHeld, "the lock in another database is not this database's table lock")
+	var heldErr *dbconn.TableLockHeldError
+	assert.NotErrorAs(t, err, &heldErr, "the other database's holder is not a rival for this table")
+
+	_, found, err := dbconn.LookupTableLockHolder(t.Context(), workPool, "app", "orders")
+	require.NoError(t, err)
+	assert.False(t, found, "the working database sees no holder")
+	holder, found, err := dbconn.LookupTableLockHolder(t.Context(), lockPool, "app", "orders")
+	require.NoError(t, err)
+	require.True(t, found, "the lock's own database sees its holder")
+	assert.Equal(t, held.BackendPID(), holder.PID)
+}
+
 // goneLock acquires the table lock and then terminates the session's
 // backend, so the server no longer grants the lock while the session still
 // believes it holds it. The cleanup waits for the keepalive to notice the