Add the golib scaffold and the pg module

Stand up the shared library with its first module: the Postgres plumbing
every service currently copy-pastes — the DSN builder, the pool
constructor, and the migration runner arrproxy proved out.

- Add pg.DSNFromEnv, generalising the identical Sprintf builders in
  encapi, artifactapi and forgebot into one prefixed lookup with
  DATABASE_URL passthrough and libpq fallbacks.
- Add pg.New and pg.NewMigrated, which ping before returning so an
  unreachable server fails at startup rather than on the first query.
- Add pg.Migrate, lifting arrproxy's runner verbatim in semantics and
  generalising the hardcoded advisory-lock key to FNV-1a/64 of a
  caller-supplied name and the embedded set to an fs.FS.
- Add pg/pgtest, unifying the encapi and artifactapi testcontainers
  helpers, with SkipIfShort so container-backed tests self-skip on the
  Docker-less Kubernetes runners.
- Add the Makefile, README and pre-commit config, plus test, pre-commit
  and build pipelines on golib-ci.

pg/dsn.go

@@ -0,0 +1,125 @@
+package pg
+
+import (
+	"fmt"
+	"net/url"
+	"os"
+	"strconv"
+	"strings"
+)
+
+// Default connection settings applied when neither the prefixed nor the libpq
+// variable for a field is set. User and database have no default: a service
+// connecting to "postgres" as "postgres" is a bug, not a default.
+const (
+	defaultHost    = "localhost"
+	defaultPort    = 5432
+	defaultSSLMode = "disable"
+)
+
+// dsnField is one connection setting and the variable names it is read from.
+type dsnField struct {
+	// suffix is appended to the caller's prefix, e.g. "DBHOST".
+	suffix string
+	// libpq is the standard libpq variable for the same setting.
+	libpq string
+}
+
+var (
+	fieldHost = dsnField{"DBHOST", "PGHOST"}
+	fieldPort = dsnField{"DBPORT", "PGPORT"}
+	fieldUser = dsnField{"DBUSER", "PGUSER"}
+	fieldPass = dsnField{"DBPASS", "PGPASSWORD"}
+	fieldName = dsnField{"DBNAME", "PGDATABASE"}
+	fieldSSL  = dsnField{"DBSSL", "PGSSLMODE"}
+)
+
+// DSNFromEnv builds a libpq/pgx connection string from the environment.
+//
+// prefix is prepended to every custom variable name, so a service can namespace
+// its settings ("ENCAPI_" reads ENCAPI_DBHOST); an empty prefix reads the bare
+// names the estate's services already use (DBHOST, DBPORT, ...).
+//
+// Precedence, highest first:
+//
+//  1. <PREFIX>DATABASE_URL — returned verbatim, no parsing or validation.
+//  2. DATABASE_URL — likewise. Identical to 1 when prefix is empty.
+//  3. <PREFIX>DBHOST, <PREFIX>DBPORT, <PREFIX>DBUSER, <PREFIX>DBPASS,
+//     <PREFIX>DBNAME, <PREFIX>DBSSL.
+//  4. libpq's PGHOST, PGPORT, PGUSER, PGPASSWORD, PGDATABASE, PGSSLMODE.
+//  5. Built-in defaults: localhost, 5432, sslmode=disable.
+//
+// Levels 3 to 5 resolve per field, so a deployment may set DBPASS from a secret
+// and leave the rest to PG* variables. An unset variable and one set to the
+// empty string are treated alike, except for the password, where the empty
+// string is a legitimate value and only distinguishable from unset if set
+// explicitly — both render the same DSN, so the distinction does not matter.
+//
+// User and database name have no default: DSNFromEnv reports an error naming
+// the variables it looked at rather than connecting somewhere unintended.
+func DSNFromEnv(prefix string) (string, error) {
+	if v := os.Getenv(prefix + "DATABASE_URL"); v != "" {
+		return v, nil
+	}
+	if v := os.Getenv("DATABASE_URL"); v != "" {
+		return v, nil
+	}
+
+	host := lookup(prefix, fieldHost, defaultHost)
+	portStr := lookup(prefix, fieldPort, strconv.Itoa(defaultPort))
+	user := lookup(prefix, fieldUser, "")
+	pass := lookup(prefix, fieldPass, "")
+	name := lookup(prefix, fieldName, "")
+	ssl := lookup(prefix, fieldSSL, defaultSSLMode)
+
+	port, err := strconv.Atoi(portStr)
+	if err != nil {
+		return "", fmt.Errorf("invalid %s%s: %w", prefix, fieldPort.suffix, err)
+	}
+	if port < 1 || port > 65535 {
+		return "", fmt.Errorf("invalid %s%s: port %d out of range", prefix, fieldPort.suffix, port)
+	}
+	if user == "" {
+		return "", fmt.Errorf("no database user: set %s%s or %s", prefix, fieldUser.suffix, fieldUser.libpq)
+	}
+	if name == "" {
+		return "", fmt.Errorf("no database name: set %s%s or %s", prefix, fieldName.suffix, fieldName.libpq)
+	}
+
+	return DSN(host, port, user, pass, name, ssl), nil
+}
+
+// DSN renders the estate's standard connection string. User and password are
+// percent-escaped, so a password containing "@" or "/" does not truncate the
+// host; for values without reserved characters the result is byte-identical to
+// the fmt.Sprintf builders this replaces.
+func DSN(host string, port int, user, pass, name, sslMode string) string {
+	u := url.URL{
+		Scheme:   "postgres",
+		User:     url.UserPassword(user, pass),
+		Host:     hostPort(host, port),
+		Path:     "/" + name,
+		RawQuery: "sslmode=" + url.QueryEscape(sslMode),
+	}
+	return u.String()
+}
+
+// hostPort joins host and port, bracketing a bare IPv6 literal so the colons in
+// the address are not read as the port separator.
+func hostPort(host string, port int) string {
+	if host != "" && host[0] != '[' && strings.Contains(host, ":") {
+		return "[" + host + "]:" + strconv.Itoa(port)
+	}
+	return host + ":" + strconv.Itoa(port)
+}
+
+// lookup resolves one field: prefixed variable, then libpq variable, then def.
+func lookup(prefix string, f dsnField, def string) string {
+	if v := os.Getenv(prefix + f.suffix); v != "" {
+		return v
+	}
+	if v := os.Getenv(f.libpq); v != "" {
+		return v
+	}
+	return def
+}

pg/dsn_test.go

@@ -0,0 +1,236 @@
+package pg
+
+import (
+	"net/url"
+	"strings"
+	"testing"
+)
+
+// dsnVars is every variable DSNFromEnv reads, for both the bare and the
+// prefixed namespace used by these tests. Each case starts from all of them
+// unset so an inherited PGHOST on a developer's machine cannot change a result.
+var dsnVars = []string{
+	"DATABASE_URL", "DBHOST", "DBPORT", "DBUSER", "DBPASS", "DBNAME", "DBSSL",
+	"PGHOST", "PGPORT", "PGUSER", "PGPASSWORD", "PGDATABASE", "PGSSLMODE",
+	"APP_DATABASE_URL", "APP_DBHOST", "APP_DBPORT", "APP_DBUSER", "APP_DBPASS",
+	"APP_DBNAME", "APP_DBSSL",
+}
+
+// setEnv clears every variable DSNFromEnv consults, then sets the given ones.
+func setEnv(t *testing.T, env map[string]string) {
+	t.Helper()
+	for _, k := range dsnVars {
+		t.Setenv(k, "")
+	}
+	for k, v := range env {
+		t.Setenv(k, v)
+	}
+}
+
+func TestDSNFromEnv(t *testing.T) {
+	tests := []struct {
+		name   string
+		prefix string
+		env    map[string]string
+		want   string
+	}{
+		{
+			name:   "prefixed custom vars",
+			prefix: "APP_",
+			env: map[string]string{
+				"APP_DBHOST": "db.internal", "APP_DBPORT": "6432",
+				"APP_DBUSER": "app", "APP_DBPASS": "s3cret",
+				"APP_DBNAME": "appdb", "APP_DBSSL": "require",
+			},
+			want: "postgres://app:s3cret@db.internal:6432/appdb?sslmode=require",
+		},
+		{
+			name:   "bare custom vars with an empty prefix",
+			prefix: "",
+			env: map[string]string{
+				"DBHOST": "pg", "DBPORT": "5432", "DBUSER": "encapi",
+				"DBPASS": "encapi", "DBNAME": "encapi", "DBSSL": "disable",
+			},
+			// Byte-identical to the fmt.Sprintf builder this replaces.
+			want: "postgres://encapi:encapi@pg:5432/encapi?sslmode=disable",
+		},
+		{
+			name:   "libpq vars fill in",
+			prefix: "APP_",
+			env: map[string]string{
+				"PGHOST": "libpq.host", "PGPORT": "5433", "PGUSER": "pguser",
+				"PGPASSWORD": "pgpass", "PGDATABASE": "pgdb", "PGSSLMODE": "verify-full",
+			},
+			want: "postgres://pguser:pgpass@libpq.host:5433/pgdb?sslmode=verify-full",
+		},
+		{
+			name:   "prefixed vars beat libpq vars per field",
+			prefix: "APP_",
+			env: map[string]string{
+				"APP_DBPASS": "from-secret",
+				"PGHOST":     "libpq.host", "PGUSER": "pguser",
+				"PGPASSWORD": "ignored", "PGDATABASE": "pgdb",
+			},
+			want: "postgres://pguser:from-secret@libpq.host:5432/pgdb?sslmode=disable",
+		},
+		{
+			name:   "defaults for host, port and sslmode",
+			prefix: "",
+			env:    map[string]string{"DBUSER": "u", "DBNAME": "d"},
+			want:   "postgres://u:@localhost:5432/d?sslmode=disable",
+		},
+		{
+			name:   "prefixed DATABASE_URL passes through verbatim",
+			prefix: "APP_",
+			env: map[string]string{
+				"APP_DATABASE_URL": "postgres://who:cares@elsewhere/db?sslmode=require&application_name=x",
+				"APP_DBHOST":       "ignored", "APP_DBUSER": "ignored", "APP_DBNAME": "ignored",
+			},
+			want: "postgres://who:cares@elsewhere/db?sslmode=require&application_name=x",
+		},
+		{
+			name:   "bare DATABASE_URL wins over the field vars",
+			prefix: "APP_",
+			env: map[string]string{
+				"DATABASE_URL": "postgres://u:p@h:5432/d",
+				"APP_DBHOST":   "ignored", "APP_DBUSER": "ignored", "APP_DBNAME": "ignored",
+			},
+			want: "postgres://u:p@h:5432/d",
+		},
+		{
+			name:   "prefixed DATABASE_URL wins over the bare one",
+			prefix: "APP_",
+			env: map[string]string{
+				"APP_DATABASE_URL": "postgres://app@app-host/app",
+				"DATABASE_URL":     "postgres://bare@bare-host/bare",
+			},
+			want: "postgres://app@app-host/app",
+		},
+		{
+			name:   "reserved characters in the password are escaped",
+			prefix: "",
+			env: map[string]string{
+				"DBHOST": "h", "DBUSER": "u", "DBPASS": "p@ss/w:rd", "DBNAME": "d",
+			},
+			want: "postgres://u:p%40ss%2Fw%3Ard@h:5432/d?sslmode=disable",
+		},
+		{
+			name:   "IPv6 host is bracketed",
+			prefix: "",
+			env: map[string]string{
+				"DBHOST": "fd00::1", "DBUSER": "u", "DBNAME": "d",
+			},
+			want: "postgres://u:@[fd00::1]:5432/d?sslmode=disable",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			setEnv(t, tc.env)
+			got, err := DSNFromEnv(tc.prefix)
+			if err != nil {
+				t.Fatalf("DSNFromEnv: %v", err)
+			}
+			if got != tc.want {
+				t.Fatalf("DSNFromEnv = %q, want %q", got, tc.want)
+			}
+		})
+	}
+}
+
+// Every DSN this builds must survive the parse pgx will do on it.
+func TestDSNFromEnv_ResultParses(t *testing.T) {
+	setEnv(t, map[string]string{
+		"DBHOST": "fd00::1", "DBPORT": "6432", "DBUSER": "us er",
+		"DBPASS": "p@ss/w:rd", "DBNAME": "app", "DBSSL": "verify-full",
+	})
+	dsn, err := DSNFromEnv("")
+	if err != nil {
+		t.Fatalf("DSNFromEnv: %v", err)
+	}
+	u, err := url.Parse(dsn)
+	if err != nil {
+		t.Fatalf("parse %q: %v", dsn, err)
+	}
+	if u.Hostname() != "fd00::1" {
+		t.Errorf("host = %q, want fd00::1", u.Hostname())
+	}
+	if u.Port() != "6432" {
+		t.Errorf("port = %q, want 6432", u.Port())
+	}
+	if u.User.Username() != "us er" {
+		t.Errorf("user = %q, want %q", u.User.Username(), "us er")
+	}
+	pass, _ := u.User.Password()
+	if pass != "p@ss/w:rd" {
+		t.Errorf("password = %q, want %q", pass, "p@ss/w:rd")
+	}
+	if got := strings.TrimPrefix(u.Path, "/"); got != "app" {
+		t.Errorf("database = %q, want app", got)
+	}
+	if got := u.Query().Get("sslmode"); got != "verify-full" {
+		t.Errorf("sslmode = %q, want verify-full", got)
+	}
+}
+
+func TestDSNFromEnv_Errors(t *testing.T) {
+	tests := []struct {
+		name    string
+		env     map[string]string
+		wantSub string
+	}{
+		{
+			name:    "unparseable port",
+			env:     map[string]string{"DBPORT": "not-a-port", "DBUSER": "u", "DBNAME": "d"},
+			wantSub: "invalid DBPORT",
+		},
+		{
+			name:    "port out of range",
+			env:     map[string]string{"DBPORT": "70000", "DBUSER": "u", "DBNAME": "d"},
+			wantSub: "out of range",
+		},
+		{
+			name:    "no user",
+			env:     map[string]string{"DBNAME": "d"},
+			wantSub: "set DBUSER or PGUSER",
+		},
+		{
+			name:    "no database name",
+			env:     map[string]string{"DBUSER": "u"},
+			wantSub: "set DBNAME or PGDATABASE",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			setEnv(t, tc.env)
+			got, err := DSNFromEnv("")
+			if err == nil {
+				t.Fatalf("expected an error, got DSN %q", got)
+			}
+			if !strings.Contains(err.Error(), tc.wantSub) {
+				t.Fatalf("error %q does not mention %q", err, tc.wantSub)
+			}
+		})
+	}
+}
+
+// The error names the prefixed variable the caller is expected to set, not the
+// bare one, or the message sends them looking for the wrong knob.
+func TestDSNFromEnv_ErrorNamesPrefixedVar(t *testing.T) {
+	setEnv(t, map[string]string{"APP_DBPORT": "x", "APP_DBUSER": "u", "APP_DBNAME": "d"})
+	_, err := DSNFromEnv("APP_")
+	if err == nil {
+		t.Fatal("expected an error")
+	}
+	if !strings.Contains(err.Error(), "APP_DBPORT") {
+		t.Fatalf("error %q does not name APP_DBPORT", err)
+	}
+}
+
+func TestDSN_EmptyPasswordMatchesLegacyFormat(t *testing.T) {
+	// The Sprintf builders rendered an unset password as an empty string
+	// between the colon and the "@"; keep that shape so DSNs do not churn.
+	if got, want := DSN("h", 5432, "u", "", "d", "disable"), "postgres://u:@h:5432/d?sslmode=disable"; got != want {
+		t.Fatalf("DSN = %q, want %q", got, want)
+	}
+}

pg/fakes_test.go

@@ -0,0 +1,173 @@
+package pg
+
+import (
+	"context"
+	"errors"
+	"io/fs"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+// brokenFS fails every Open, so fs.ReadDir(".") fails.
+type brokenFS struct{}
+
+func (brokenFS) Open(string) (fs.File, error) { return nil, fs.ErrPermission }
+
+// missingFileFS lists a migration that cannot then be read, the shape a
+// mis-built embed or a racing file deletion produces.
+type missingFileFS struct{}
+
+func (missingFileFS) Open(string) (fs.File, error) { return nil, fs.ErrNotExist }
+
+func (missingFileFS) ReadDir(string) ([]fs.DirEntry, error) {
+	return []fs.DirEntry{fakeDirEntry{name: "0001_first.sql"}}, nil
+}
+
+type fakeDirEntry struct {
+	name string
+	dir  bool
+}
+
+func (e fakeDirEntry) Name() string { return e.name }
+func (e fakeDirEntry) IsDir() bool  { return e.dir }
+func (e fakeDirEntry) Type() fs.FileMode {
+	if e.dir {
+		return fs.ModeDir
+	}
+	return 0
+}
+func (e fakeDirEntry) Info() (fs.FileInfo, error) { return nil, errors.New("no info") }
+
+// execCall records one statement the migrator issued.
+type execCall struct {
+	sql  string
+	args []any
+}
+
+// fakeSession is a pgx connection that records statements instead of running
+// them. Embedding nothing: it implements the whole session interface.
+type fakeSession struct {
+	execs    []execCall
+	execErrs map[string]error
+
+	queries  []string
+	rows     *fakeRows
+	queryErr error
+
+	tx       *fakeTx
+	beginErr error
+}
+
+func (s *fakeSession) Exec(_ context.Context, sql string, args ...any) (pgconn.CommandTag, error) {
+	s.execs = append(s.execs, execCall{sql: sql, args: args})
+	return pgconn.NewCommandTag("SELECT 1"), s.execErrs[sql]
+}
+
+func (s *fakeSession) Query(_ context.Context, sql string, _ ...any) (pgx.Rows, error) {
+	s.queries = append(s.queries, sql)
+	if s.queryErr != nil {
+		return nil, s.queryErr
+	}
+	return s.rows, nil
+}
+
+func (s *fakeSession) Begin(context.Context) (pgx.Tx, error) {
+	if s.beginErr != nil {
+		return nil, s.beginErr
+	}
+	if s.tx == nil {
+		s.tx = &fakeTx{}
+	}
+	return s.tx, nil
+}
+
+// execSQL returns just the statements issued, in order.
+func (s *fakeSession) execSQL() []string {
+	out := make([]string, 0, len(s.execs))
+	for _, c := range s.execs {
+		out = append(out, c.sql)
+	}
+	return out
+}
+
+// fakeRows serves a fixed column of strings. The embedded interface supplies
+// the pgx.Rows methods the migrator never calls; calling one panics, which is
+// the intent — it would mean the migrator grew an untested dependency.
+type fakeRows struct {
+	pgx.Rows
+
+	values  []string
+	i       int
+	scanErr error
+	err     error
+	closed  bool
+}
+
+func (r *fakeRows) Next() bool {
+	if r.i >= len(r.values) {
+		return false
+	}
+	r.i++
+	return true
+}
+
+func (r *fakeRows) Scan(dest ...any) error {
+	if r.scanErr != nil {
+		return r.scanErr
+	}
+	if len(dest) != 1 {
+		return errors.New("expected exactly one scan destination")
+	}
+	p, ok := dest[0].(*string)
+	if !ok {
+		return errors.New("expected a *string scan destination")
+	}
+	*p = r.values[r.i-1]
+	return nil
+}
+
+func (r *fakeRows) Close()     { r.closed = true }
+func (r *fakeRows) Err() error { return r.err }
+
+// fakeTx records the statements and the terminal call of a transaction.
+type fakeTx struct {
+	pgx.Tx
+
+	execs     []execCall
+	execErrs  map[string]error
+	commitErr error
+	committed bool
+	rolled    int
+}
+
+func (t *fakeTx) Exec(_ context.Context, sql string, args ...any) (pgconn.CommandTag, error) {
+	t.execs = append(t.execs, execCall{sql: sql, args: args})
+	return pgconn.NewCommandTag("INSERT 0 1"), t.execErrs[sql]
+}
+
+func (t *fakeTx) Commit(context.Context) error {
+	if t.commitErr != nil {
+		return t.commitErr
+	}
+	t.committed = true
+	return nil
+}
+
+func (t *fakeTx) Rollback(context.Context) error {
+	t.rolled++
+	if t.committed {
+		// What pgx returns for a rollback after a successful commit; the
+		// deferred rollback in Apply must tolerate it.
+		return pgx.ErrTxClosed
+	}
+	return nil
+}
+
+func (t *fakeTx) execSQL() []string {
+	out := make([]string, 0, len(t.execs))
+	for _, c := range t.execs {
+		out = append(out, c.sql)
+	}
+	return out
+}

pg/integration_test.go

@@ -0,0 +1,186 @@
+package pg_test
+
+import (
+	"context"
+	"net/url"
+	"strings"
+	"testing"
+	"testing/fstest"
+	"time"
+
+	"git.unkin.net/unkin/golib/pg"
+	"git.unkin.net/unkin/golib/pg/pgtest"
+)
+
+// migrations is a two-file set exercising the things the runner promises:
+// multi-statement files (simple protocol) and IF NOT EXISTS re-runnability.
+var migrations = fstest.MapFS{
+	"0001_widgets.sql": {Data: []byte(`
+CREATE TABLE IF NOT EXISTS widgets (id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL);
+CREATE INDEX IF NOT EXISTS widgets_name_idx ON widgets (name);
+`)},
+	"0002_gadgets.sql": {Data: []byte(`CREATE TABLE IF NOT EXISTS gadgets (id BIGSERIAL PRIMARY KEY);`)},
+	"notes.md":         {Data: []byte("ignored")},
+}
+
+func testCtx(t *testing.T) context.Context {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	t.Cleanup(cancel)
+	return ctx
+}
+
+// The full path against a real server: connect, migrate, and confirm the schema
+// and the bookkeeping table both landed.
+func TestNewMigrated_AgainstRealPostgres(t *testing.T) {
+	ctx := testCtx(t)
+	dsn := pgtest.MustStartPostgres(ctx, t)
+
+	pool, err := pg.NewMigrated(ctx, dsn, migrations, pg.MigrateOptions{LockName: "golib-pg-integration"})
+	if err != nil {
+		t.Fatalf("NewMigrated: %v", err)
+	}
+	t.Cleanup(pool.Close)
+
+	for _, table := range []string{"widgets", "gadgets", "schema_migrations"} {
+		var exists bool
+		err := pool.QueryRow(ctx,
+			`SELECT EXISTS (SELECT 1 FROM information_schema.tables
+			                WHERE table_schema = 'public' AND table_name = $1)`, table).Scan(&exists)
+		if err != nil {
+			t.Fatalf("check %s: %v", table, err)
+		}
+		if !exists {
+			t.Errorf("table %s was not created", table)
+		}
+	}
+
+	// The second statement of the multi-statement file must have run too.
+	var idx bool
+	if err := pool.QueryRow(ctx,
+		`SELECT EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'widgets_name_idx')`).Scan(&idx); err != nil {
+		t.Fatalf("check index: %v", err)
+	}
+	if !idx {
+		t.Error("the second statement of the multi-statement migration did not run")
+	}
+
+	var versions []string
+	rows, err := pool.Query(ctx, "SELECT version FROM schema_migrations ORDER BY version")
+	if err != nil {
+		t.Fatalf("read schema_migrations: %v", err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var v string
+		if err := rows.Scan(&v); err != nil {
+			t.Fatalf("scan: %v", err)
+		}
+		versions = append(versions, v)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatalf("rows: %v", err)
+	}
+	want := []string{"0001_widgets.sql", "0002_gadgets.sql"}
+	if len(versions) != len(want) {
+		t.Fatalf("recorded versions %v, want %v", versions, want)
+	}
+	for i := range want {
+		if versions[i] != want[i] {
+			t.Fatalf("recorded versions %v, want %v", versions, want)
+		}
+	}
+}
+
+// Every replica calls Migrate at startup. Running it concurrently against one
+// server must apply the set exactly once and leave no lock held.
+func TestMigrate_ConcurrentRepliesConverge(t *testing.T) {
+	ctx := testCtx(t)
+	dsn := pgtest.MustStartPostgres(ctx, t)
+
+	pool, err := pg.New(ctx, dsn, nil)
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	t.Cleanup(pool.Close)
+
+	const replicas = 4
+	errs := make(chan error, replicas)
+	for range replicas {
+		go func() {
+			errs <- pg.Migrate(ctx, pool, migrations, pg.MigrateOptions{LockName: "golib-pg-integration"})
+		}()
+	}
+	for range replicas {
+		if err := <-errs; err != nil {
+			t.Fatalf("Migrate: %v", err)
+		}
+	}
+
+	var n int
+	if err := pool.QueryRow(ctx, "SELECT count(*) FROM schema_migrations").Scan(&n); err != nil {
+		t.Fatalf("count versions: %v", err)
+	}
+	if n != 2 {
+		t.Fatalf("schema_migrations has %d rows, want 2", n)
+	}
+
+	// Nothing may still hold the migration lock once every replica has finished.
+	var held bool
+	if err := pool.QueryRow(ctx,
+		`SELECT EXISTS (SELECT 1 FROM pg_locks WHERE locktype = 'advisory' AND objid IS NOT NULL AND granted)`,
+	).Scan(&held); err != nil {
+		t.Fatalf("check locks: %v", err)
+	}
+	if held {
+		t.Error("an advisory lock is still held after every replica finished")
+	}
+}
+
+// DSNFromEnv's output must be something pgx can actually connect with.
+func TestDSNFromEnv_ConnectsToRealPostgres(t *testing.T) {
+	ctx := testCtx(t)
+	dsn := pgtest.MustStartPostgres(ctx, t)
+
+	t.Setenv("DATABASE_URL", "")
+	t.Setenv("APP_DATABASE_URL", "")
+
+	cfg, err := parseDSN(dsn)
+	if err != nil {
+		t.Fatalf("parse container DSN: %v", err)
+	}
+	t.Setenv("APP_DBHOST", cfg.host)
+	t.Setenv("APP_DBPORT", cfg.port)
+	t.Setenv("APP_DBUSER", cfg.user)
+	t.Setenv("APP_DBPASS", cfg.pass)
+	t.Setenv("APP_DBNAME", cfg.name)
+	t.Setenv("APP_DBSSL", "disable")
+
+	built, err := pg.DSNFromEnv("APP_")
+	if err != nil {
+		t.Fatalf("DSNFromEnv: %v", err)
+	}
+	pool, err := pg.New(ctx, built, nil)
+	if err != nil {
+		t.Fatalf("New(%q): %v", built, err)
+	}
+	pool.Close()
+}
+
+// dsnParts is the container DSN split back into the fields DSNFromEnv reads.
+type dsnParts struct{ host, port, user, pass, name string }
+
+func parseDSN(dsn string) (dsnParts, error) {
+	u, err := url.Parse(dsn)
+	if err != nil {
+		return dsnParts{}, err
+	}
+	pass, _ := u.User.Password()
+	return dsnParts{
+		host: u.Hostname(),
+		port: u.Port(),
+		user: u.User.Username(),
+		pass: pass,
+		name: strings.TrimPrefix(u.Path, "/"),
+	}, nil
+}

pg/migrate.go

@@ -0,0 +1,243 @@
+package pg
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"hash/fnv"
+	"io/fs"
+	"log/slog"
+	"sort"
+	"strings"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// schemaMigrationsDDL creates the version table itself, outside the tracked
+// set: it is step zero of every run and is never recorded as a migration.
+const schemaMigrationsDDL = `CREATE TABLE IF NOT EXISTS schema_migrations (
+    version    TEXT PRIMARY KEY,
+    applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
+)`
+
+// MigrateOptions configures one migration run.
+type MigrateOptions struct {
+	// LockName names the cluster-wide advisory lock replicas contend for.
+	// Every process migrating the same database must pass the same name, and
+	// two databases sharing a Postgres cluster should not: the lock is per
+	// cluster, not per database. Required.
+	LockName string
+
+	// Logger receives one line per applied migration. Nil discards them.
+	Logger *slog.Logger
+}
+
+// LockKey derives the pg_advisory_lock key for a lock name as FNV-1a/64 of the
+// name, reinterpreted as int64. It is exported so a service migrating off a
+// hardcoded key can assert the two agree before switching over.
+func LockKey(name string) int64 {
+	h := fnv.New64a()
+	// hash.Hash.Write never returns an error.
+	_, _ = h.Write([]byte(name))
+	return int64(h.Sum64())
+}
+
+// migrator is the slice of Postgres the migration runner drives. It keeps the
+// ordering, locking and bookkeeping logic testable without a live database.
+type migrator interface {
+	// Lock blocks until this process holds the cluster-wide migration lock.
+	Lock(ctx context.Context) error
+	// Unlock releases it.
+	Unlock(ctx context.Context) error
+	// Discard throws away the underlying session so a lock that could not be
+	// released dies with the connection instead of being returned to the pool.
+	Discard(ctx context.Context)
+	// EnsureVersionTable creates schema_migrations if it is missing.
+	EnsureVersionTable(ctx context.Context) error
+	// AppliedVersions returns the versions already recorded.
+	AppliedVersions(ctx context.Context) (map[string]bool, error)
+	// Apply runs one migration's SQL and records its version in a single
+	// transaction, so a failure leaves neither behind.
+	Apply(ctx context.Context, version, sql string) error
+}
+
+// migrationNames returns the .sql files in fsys in version (lexical) order.
+func migrationNames(fsys fs.FS) ([]string, error) {
+	entries, err := fs.ReadDir(fsys, ".")
+	if err != nil {
+		return nil, fmt.Errorf("read migrations: %w", err)
+	}
+	var names []string
+	for _, e := range entries {
+		if !e.IsDir() && strings.HasSuffix(e.Name(), ".sql") {
+			names = append(names, e.Name())
+		}
+	}
+	if len(names) == 0 {
+		return nil, errors.New("no migrations found")
+	}
+	sort.Strings(names)
+	return names, nil
+}
+
+// runMigrations applies every migration in fsys that schema_migrations does not
+// already record, in version order, while holding the advisory lock. Replicas
+// starting at the same time queue on the lock and then find nothing to do.
+//
+// A file absent from schema_migrations is re-run even if the live database
+// already has it, which is how a schema applied out of band before adopting
+// this runner is picked up. Migrations are therefore expected to be
+// IF NOT EXISTS-guarded, so such a re-run is a no-op that only lands the
+// missing tracking row.
+func runMigrations(ctx context.Context, m migrator, fsys fs.FS, log *slog.Logger) error {
+	names, err := migrationNames(fsys)
+	if err != nil {
+		return err
+	}
+	if err := m.Lock(ctx); err != nil {
+		return fmt.Errorf("acquire migration lock: %w", err)
+	}
+	defer func() {
+		if err := m.Unlock(ctx); err != nil {
+			// The lock is session-scoped: if the unlock did not land we cannot
+			// know the session dropped it, so kill the session rather than let a
+			// still-locked connection back into the pool.
+			log.Warn("release migration lock, discarding connection", "err", err)
+			m.Discard(ctx)
+		}
+	}()
+
+	if err := m.EnsureVersionTable(ctx); err != nil {
+		return fmt.Errorf("ensure schema_migrations: %w", err)
+	}
+	applied, err := m.AppliedVersions(ctx)
+	if err != nil {
+		return fmt.Errorf("read applied migrations: %w", err)
+	}
+
+	var n int
+	for _, name := range names {
+		if applied[name] {
+			continue
+		}
+		body, err := fs.ReadFile(fsys, name)
+		if err != nil {
+			return fmt.Errorf("read migration %s: %w", name, err)
+		}
+		if err := m.Apply(ctx, name, string(body)); err != nil {
+			return fmt.Errorf("apply migration %s: %w", name, err)
+		}
+		log.Info("applied migration", "version", name)
+		n++
+	}
+	if n == 0 {
+		log.Info("schema up to date")
+	}
+	return nil
+}
+
+// session is the slice of a pgx connection the migrator drives. *pgxpool.Conn
+// satisfies it; naming it keeps the SQL testable without a live database.
+type session interface {
+	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
+	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
+	Begin(ctx context.Context) (pgx.Tx, error)
+}
+
+// pgMigrator runs migrations on one dedicated pooled connection: the advisory
+// lock is session-scoped, so lock, apply and unlock must share a connection.
+type pgMigrator struct {
+	conn session
+	key  int64
+	// discard closes the physical connection; the pool destroys a closed
+	// connection on Release instead of reusing it.
+	discard func(context.Context)
+}
+
+func (p *pgMigrator) Lock(ctx context.Context) error {
+	_, err := p.conn.Exec(ctx, "SELECT pg_advisory_lock($1)", p.key)
+	return err
+}
+
+func (p *pgMigrator) Unlock(ctx context.Context) error {
+	_, err := p.conn.Exec(ctx, "SELECT pg_advisory_unlock($1)", p.key)
+	return err
+}
+
+func (p *pgMigrator) Discard(ctx context.Context) { p.discard(ctx) }
+
+func (p *pgMigrator) EnsureVersionTable(ctx context.Context) error {
+	_, err := p.conn.Exec(ctx, schemaMigrationsDDL)
+	return err
+}
+
+func (p *pgMigrator) AppliedVersions(ctx context.Context) (map[string]bool, error) {
+	rows, err := p.conn.Query(ctx, "SELECT version FROM schema_migrations")
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	applied := map[string]bool{}
+	for rows.Next() {
+		var v string
+		if err := rows.Scan(&v); err != nil {
+			return nil, err
+		}
+		applied[v] = true
+	}
+	if err := rows.Err(); err != nil {
+		// A read that died part-way must not look like a complete set, or the
+		// caller would skip migrations it has not actually applied.
+		return nil, err
+	}
+	return applied, nil
+}
+
+func (p *pgMigrator) Apply(ctx context.Context, version, sql string) error {
+	tx, err := p.conn.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	// Zero-arg Exec uses the simple protocol, so a multi-statement file runs.
+	if _, err := tx.Exec(ctx, sql); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(ctx, "INSERT INTO schema_migrations (version) VALUES ($1)", version); err != nil {
+		return err
+	}
+	return tx.Commit(ctx)
+}
+
+// Migrate brings the database behind pool up to date with the .sql files at the
+// root of fsys, applied in lexical filename order. It is safe to call from
+// every replica at once: the run holds a cluster-wide advisory lock derived
+// from opts.LockName, and replicas that queue behind the winner find the set
+// already recorded and do nothing.
+//
+// Migrations run on one dedicated pooled connection, because the advisory lock
+// is session-scoped. Each file is applied together with its schema_migrations
+// row in a single transaction, so a failure part-way through leaves neither the
+// half-applied file nor a tracking row that would skip it next time.
+func Migrate(ctx context.Context, pool *pgxpool.Pool, fsys fs.FS, opts MigrateOptions) error {
+	if opts.LockName == "" {
+		return errors.New("pg: MigrateOptions.LockName is required")
+	}
+	conn, err := pool.Acquire(ctx)
+	if err != nil {
+		return fmt.Errorf("acquire migration connection: %w", err)
+	}
+	defer conn.Release()
+	return migrateSession(ctx, conn, func(ctx context.Context) { _ = conn.Conn().Close(ctx) }, fsys, opts)
+}
+
+// migrateSession runs the set on an already-acquired connection.
+func migrateSession(ctx context.Context, conn session, discard func(context.Context), fsys fs.FS, opts MigrateOptions) error {
+	m := &pgMigrator{conn: conn, key: LockKey(opts.LockName), discard: discard}
+	if err := runMigrations(ctx, m, fsys, logger(opts.Logger)); err != nil {
+		return fmt.Errorf("migrate schema: %w", err)
+	}
+	return nil
+}

pg/migrate_test.go

@@ -0,0 +1,428 @@
+package pg
+
+import (
+	"context"
+	"errors"
+	"hash/fnv"
+	"io"
+	"log/slog"
+	"reflect"
+	"strings"
+	"sync"
+	"testing"
+	"testing/fstest"
+)
+
+func testLogger() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+// testFS is a three-file migration set in deliberately unsorted map order.
+func testFS() fstest.MapFS {
+	return fstest.MapFS{
+		"0002_second.sql": {Data: []byte("CREATE TABLE IF NOT EXISTS b ();")},
+		"0001_first.sql":  {Data: []byte("CREATE TABLE IF NOT EXISTS a ();")},
+		"0003_third.sql":  {Data: []byte("CREATE TABLE IF NOT EXISTS c ();")},
+		"README.md":       {Data: []byte("not a migration")},
+	}
+}
+
+var testNames = []string{"0001_first.sql", "0002_second.sql", "0003_third.sql"}
+
+// fakeDB stands in for Postgres: the advisory lock is a mutex, the version
+// table an in-memory set, and a migration is "run" by recording its version.
+type fakeDB struct {
+	lock sync.Mutex // the advisory lock: one holder at a time, cluster-wide
+
+	mu           sync.Mutex
+	held         bool
+	tableCreated bool
+	applied      []string
+	applyCalls   []string
+	bodies       []string
+	failOn       string
+	unlockErr    error
+	discards     int
+	violations   []string
+}
+
+type fakeMigrator struct{ db *fakeDB }
+
+func (f *fakeMigrator) Lock(ctx context.Context) error {
+	f.db.lock.Lock()
+	f.db.mu.Lock()
+	defer f.db.mu.Unlock()
+	f.db.held = true
+	return nil
+}
+
+func (f *fakeMigrator) Unlock(ctx context.Context) error {
+	f.db.mu.Lock()
+	f.db.held = false
+	err := f.db.unlockErr
+	f.db.mu.Unlock()
+	f.db.lock.Unlock()
+	return err
+}
+
+func (f *fakeMigrator) Discard(ctx context.Context) {
+	f.db.mu.Lock()
+	defer f.db.mu.Unlock()
+	f.db.discards++
+}
+
+// requireHeld records any access made without the advisory lock; every schema
+// read or write must happen inside the locked section.
+func (f *fakeMigrator) requireHeld(op string) {
+	if !f.db.held {
+		f.db.violations = append(f.db.violations, op)
+	}
+}
+
+func (f *fakeMigrator) EnsureVersionTable(ctx context.Context) error {
+	f.db.mu.Lock()
+	defer f.db.mu.Unlock()
+	f.requireHeld("EnsureVersionTable")
+	f.db.tableCreated = true
+	return nil
+}
+
+func (f *fakeMigrator) AppliedVersions(ctx context.Context) (map[string]bool, error) {
+	f.db.mu.Lock()
+	defer f.db.mu.Unlock()
+	f.requireHeld("AppliedVersions")
+	if !f.db.tableCreated {
+		return nil, errors.New("schema_migrations does not exist")
+	}
+	out := map[string]bool{}
+	for _, v := range f.db.applied {
+		out[v] = true
+	}
+	return out, nil
+}
+
+func (f *fakeMigrator) Apply(ctx context.Context, version, sql string) error {
+	f.db.mu.Lock()
+	defer f.db.mu.Unlock()
+	f.requireHeld("Apply")
+	f.db.applyCalls = append(f.db.applyCalls, version)
+	f.db.bodies = append(f.db.bodies, sql)
+	if sql == "" {
+		return errors.New("empty migration body")
+	}
+	// A failing migration rolls back, so neither the SQL nor the version row lands.
+	if version == f.db.failOn {
+		return errors.New("boom")
+	}
+	f.db.applied = append(f.db.applied, version)
+	return nil
+}
+
+func (f *fakeDB) snapshot() (applied, calls, violations []string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]string(nil), f.applied...),
+		append([]string(nil), f.applyCalls...),
+		append([]string(nil), f.violations...)
+}
+
+func (f *fakeDB) discardCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.discards
+}
+
+// arrproxy pinned its advisory lock key as a literal derived from
+// FNV-1a/64("arrproxy-migrations"). LockKey must reproduce it exactly, or
+// adopting this package would silently stop excluding the old deployment.
+func TestLockKey_MatchesTheEstatesPinnedKey(t *testing.T) {
+	const arrproxyKey int64 = 7816645656172167846
+	if got := LockKey("arrproxy-migrations"); got != arrproxyKey {
+		t.Fatalf("LockKey(%q) = %d, want %d", "arrproxy-migrations", got, arrproxyKey)
+	}
+}
+
+func TestLockKey_IsFNV1a64(t *testing.T) {
+	for _, name := range []string{"", "encapi-migrations", "a much longer lock name"} {
+		h := fnv.New64a()
+		if _, err := h.Write([]byte(name)); err != nil {
+			t.Fatalf("hash: %v", err)
+		}
+		if got, want := LockKey(name), int64(h.Sum64()); got != want {
+			t.Errorf("LockKey(%q) = %d, want %d", name, got, want)
+		}
+	}
+}
+
+// Distinct names must not collide, or two services sharing a cluster would
+// serialise against each other by accident.
+func TestLockKey_DistinctNamesDistinctKeys(t *testing.T) {
+	if LockKey("encapi-migrations") == LockKey("artifactapi-migrations") {
+		t.Fatal("two different lock names produced the same key")
+	}
+}
+
+func TestMigrationNames_SortedSQLOnly(t *testing.T) {
+	got, err := migrationNames(testFS())
+	if err != nil {
+		t.Fatalf("migrationNames: %v", err)
+	}
+	if !reflect.DeepEqual(got, testNames) {
+		t.Fatalf("got %v, want %v", got, testNames)
+	}
+}
+
+func TestMigrationNames_IgnoresDirectories(t *testing.T) {
+	fsys := fstest.MapFS{
+		"0001_first.sql":     {Data: []byte("SELECT 1;")},
+		"sub.sql/keep.sql":   {Data: []byte("SELECT 1;")},
+		"0002_second.sql.gz": {Data: []byte("SELECT 1;")},
+	}
+	got, err := migrationNames(fsys)
+	if err != nil {
+		t.Fatalf("migrationNames: %v", err)
+	}
+	if want := []string{"0001_first.sql"}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestMigrationNames_EmptySetIsAnError(t *testing.T) {
+	if _, err := migrationNames(fstest.MapFS{}); err == nil {
+		t.Fatal("expected an error for an empty migration set")
+	}
+}
+
+func TestMigrationNames_UnreadableFSIsAnError(t *testing.T) {
+	if _, err := migrationNames(brokenFS{}); err == nil {
+		t.Fatal("expected an error when the migration directory cannot be read")
+	}
+}
+
+func TestRunMigrations_FreshDatabaseAppliesAllInOrder(t *testing.T) {
+	db := &fakeDB{}
+	if err := runMigrations(context.Background(), &fakeMigrator{db: db}, testFS(), testLogger()); err != nil {
+		t.Fatalf("runMigrations: %v", err)
+	}
+	applied, _, violations := db.snapshot()
+	if !reflect.DeepEqual(applied, testNames) {
+		t.Fatalf("applied %v, want %v", applied, testNames)
+	}
+	if !db.tableCreated {
+		t.Error("schema_migrations was not created")
+	}
+	if len(violations) != 0 {
+		t.Errorf("schema accessed without the advisory lock: %v", violations)
+	}
+}
+
+// The runner must hand Apply the file's bytes, not just its name.
+func TestRunMigrations_PassesFileBodies(t *testing.T) {
+	db := &fakeDB{}
+	fsys := testFS()
+	if err := runMigrations(context.Background(), &fakeMigrator{db: db}, fsys, testLogger()); err != nil {
+		t.Fatalf("runMigrations: %v", err)
+	}
+	db.mu.Lock()
+	defer db.mu.Unlock()
+	for i, name := range testNames {
+		if want := string(fsys[name].Data); db.bodies[i] != want {
+			t.Errorf("body %d = %q, want %q", i, db.bodies[i], want)
+		}
+	}
+}
+
+func TestRunMigrations_SecondRunIsANoOp(t *testing.T) {
+	db := &fakeDB{}
+	ctx := context.Background()
+	for i := range 2 {
+		if err := runMigrations(ctx, &fakeMigrator{db: db}, testFS(), testLogger()); err != nil {
+			t.Fatalf("run %d: %v", i, err)
+		}
+	}
+	applied, calls, _ := db.snapshot()
+	if !reflect.DeepEqual(applied, testNames) {
+		t.Fatalf("applied %v, want %v", applied, testNames)
+	}
+	if len(calls) != len(testNames) {
+		t.Fatalf("Apply called %d times across two runs, want %d", len(calls), len(testNames))
+	}
+}
+
+// A file missing from schema_migrations is re-applied even when the rest of the
+// set is recorded: that is how a schema applied out of band is adopted.
+func TestRunMigrations_AppliesOnlyTheMissingVersion(t *testing.T) {
+	db := &fakeDB{tableCreated: true, applied: []string{testNames[0], testNames[2]}}
+	if err := runMigrations(context.Background(), &fakeMigrator{db: db}, testFS(), testLogger()); err != nil {
+		t.Fatalf("runMigrations: %v", err)
+	}
+	_, calls, _ := db.snapshot()
+	if want := testNames[1:2]; !reflect.DeepEqual(calls, want) {
+		t.Fatalf("Apply calls %v, want %v", calls, want)
+	}
+}
+
+// Replicas starting together queue on the advisory lock: the first applies the
+// set, the rest find it already recorded and do nothing.
+func TestRunMigrations_ConcurrentStartersSerialize(t *testing.T) {
+	db := &fakeDB{}
+	ctx := context.Background()
+	var wg sync.WaitGroup
+	errs := make([]error, 4)
+	for i := range errs {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			errs[i] = runMigrations(ctx, &fakeMigrator{db: db}, testFS(), testLogger())
+		}()
+	}
+	wg.Wait()
+	for i, err := range errs {
+		if err != nil {
+			t.Fatalf("starter %d: %v", i, err)
+		}
+	}
+	applied, calls, violations := db.snapshot()
+	if !reflect.DeepEqual(applied, testNames) {
+		t.Fatalf("applied %v, want %v", applied, testNames)
+	}
+	if len(calls) != len(testNames) {
+		t.Fatalf("Apply called %d times, want %d (one starter should have applied)", len(calls), len(testNames))
+	}
+	if len(violations) != 0 {
+		t.Errorf("schema accessed without the advisory lock: %v", violations)
+	}
+}
+
+// A migration that fails mid-set aborts the run, leaves earlier versions
+// recorded, and never reaches later ones.
+func TestRunMigrations_FailureStopsAndKeepsEarlierVersions(t *testing.T) {
+	db := &fakeDB{failOn: testNames[1]}
+	err := runMigrations(context.Background(), &fakeMigrator{db: db}, testFS(), testLogger())
+	if err == nil {
+		t.Fatal("expected the failing migration to abort the run")
+	}
+	if !strings.Contains(err.Error(), testNames[1]) {
+		t.Errorf("error %q does not name the failing migration", err)
+	}
+	applied, calls, _ := db.snapshot()
+	if !reflect.DeepEqual(applied, testNames[:1]) {
+		t.Fatalf("applied %v, want %v", applied, testNames[:1])
+	}
+	if !reflect.DeepEqual(calls, testNames[:2]) {
+		t.Fatalf("Apply calls %v, want %v (later migrations must not run)", calls, testNames[:2])
+	}
+	// The lock must be released even on failure, or every later pod deadlocks.
+	if !db.lock.TryLock() {
+		t.Fatal("advisory lock still held after a failed run")
+	}
+	db.lock.Unlock()
+}
+
+// A run that cannot take the lock must not touch the schema.
+func TestRunMigrations_LockFailureAborts(t *testing.T) {
+	db := &fakeDB{}
+	m := &failingLockMigrator{fakeMigrator{db: db}}
+	if err := runMigrations(context.Background(), m, testFS(), testLogger()); err == nil {
+		t.Fatal("expected a lock failure to abort the run")
+	}
+	applied, calls, _ := db.snapshot()
+	if len(applied) != 0 || len(calls) != 0 {
+		t.Fatalf("migrations ran without the lock: applied %v, calls %v", applied, calls)
+	}
+	if db.tableCreated {
+		t.Error("schema_migrations was created without the lock")
+	}
+}
+
+// A run that cannot see its bookkeeping table fails instead of re-applying blind.
+func TestRunMigrations_ReadAppliedFailureAborts(t *testing.T) {
+	// tableCreated stays false because EnsureVersionTable is a no-op here, so
+	// the fake's AppliedVersions reports the table as missing.
+	db := &fakeDB{}
+	m := &noTableMigrator{fakeMigrator{db: db}}
+	err := runMigrations(context.Background(), m, testFS(), testLogger())
+	if err == nil {
+		t.Fatal("expected the run to fail when applied versions cannot be read")
+	}
+	if !strings.Contains(err.Error(), "read applied migrations") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+	_, calls, _ := db.snapshot()
+	if len(calls) != 0 {
+		t.Fatalf("migrations applied without reading the version table: %v", calls)
+	}
+}
+
+func TestRunMigrations_EnsureVersionTableFailureAborts(t *testing.T) {
+	db := &fakeDB{}
+	m := &failingEnsureMigrator{fakeMigrator{db: db}}
+	err := runMigrations(context.Background(), m, testFS(), testLogger())
+	if err == nil {
+		t.Fatal("expected the run to fail when schema_migrations cannot be created")
+	}
+	if !strings.Contains(err.Error(), "ensure schema_migrations") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+	// Even this early failure must release the lock.
+	if !db.lock.TryLock() {
+		t.Fatal("advisory lock still held")
+	}
+	db.lock.Unlock()
+}
+
+func TestRunMigrations_UnreadableMigrationAborts(t *testing.T) {
+	db := &fakeDB{}
+	// A directory entry that ReadDir reports as a file but ReadFile cannot open.
+	err := runMigrations(context.Background(), &fakeMigrator{db: db}, missingFileFS{}, testLogger())
+	if err == nil {
+		t.Fatal("expected an unreadable migration to abort the run")
+	}
+	if !strings.Contains(err.Error(), "read migration") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+}
+
+// An unlock that does not land leaves a session that may still hold the lock, so
+// the connection is discarded rather than returned to the pool. The migrations
+// themselves already succeeded, so the run still reports success.
+func TestRunMigrations_UnlockFailureDiscardsConnection(t *testing.T) {
+	db := &fakeDB{unlockErr: errors.New("connection reset")}
+	if err := runMigrations(context.Background(), &fakeMigrator{db: db}, testFS(), testLogger()); err != nil {
+		t.Fatalf("runMigrations: %v", err)
+	}
+	if got := db.discardCount(); got != 1 {
+		t.Fatalf("Discard called %d times after a failed unlock, want 1", got)
+	}
+	applied, _, _ := db.snapshot()
+	if !reflect.DeepEqual(applied, testNames) {
+		t.Fatalf("applied %v, want %v", applied, testNames)
+	}
+}
+
+// A clean unlock keeps the connection: discarding every migration connection
+// would churn the pool on every start.
+func TestRunMigrations_CleanUnlockKeepsConnection(t *testing.T) {
+	db := &fakeDB{}
+	if err := runMigrations(context.Background(), &fakeMigrator{db: db}, testFS(), testLogger()); err != nil {
+		t.Fatalf("runMigrations: %v", err)
+	}
+	if got := db.discardCount(); got != 0 {
+		t.Fatalf("Discard called %d times after a clean unlock, want 0", got)
+	}
+}
+
+type failingLockMigrator struct{ fakeMigrator }
+
+func (f *failingLockMigrator) Lock(ctx context.Context) error { return errors.New("lock unavailable") }
+
+type failingEnsureMigrator struct{ fakeMigrator }
+
+func (f *failingEnsureMigrator) EnsureVersionTable(ctx context.Context) error {
+	return errors.New("permission denied")
+}
+
+type noTableMigrator struct{ fakeMigrator }
+
+// EnsureVersionTable silently does nothing, so AppliedVersions then fails.
+func (f *noTableMigrator) EnsureVersionTable(ctx context.Context) error { return nil }

pg/pg.go

@@ -0,0 +1,18 @@
+// Package pg holds the estate's shared Postgres plumbing: environment-driven
+// DSN construction, pgxpool construction, and the migration runner every
+// service uses to bring its own schema up to date at startup.
+package pg
+
+import (
+	"io"
+	"log/slog"
+)
+
+// logger returns log, or a logger that discards everything when log is nil, so
+// callers may leave the option unset.
+func logger(log *slog.Logger) *slog.Logger {
+	if log != nil {
+		return log
+	}
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}

pg/pgmigrator_test.go

@@ -0,0 +1,302 @@
+package pg
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+const testLockName = "golib-test-migrations"
+
+func newTestMigrator(s *fakeSession) (*pgMigrator, *int) {
+	discards := 0
+	m := &pgMigrator{
+		conn:    s,
+		key:     LockKey(testLockName),
+		discard: func(context.Context) { discards++ },
+	}
+	return m, &discards
+}
+
+// The lock and unlock must name the same key, and it must be the derived one:
+// a mismatch here is a deadlock or a lock nobody else respects.
+func TestPgMigrator_LockUnlockUseTheDerivedKey(t *testing.T) {
+	s := &fakeSession{}
+	m, _ := newTestMigrator(s)
+	ctx := context.Background()
+
+	if err := m.Lock(ctx); err != nil {
+		t.Fatalf("Lock: %v", err)
+	}
+	if err := m.Unlock(ctx); err != nil {
+		t.Fatalf("Unlock: %v", err)
+	}
+
+	want := []string{"SELECT pg_advisory_lock($1)", "SELECT pg_advisory_unlock($1)"}
+	if got := s.execSQL(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("statements %v, want %v", got, want)
+	}
+	key := LockKey(testLockName)
+	for _, c := range s.execs {
+		if len(c.args) != 1 || c.args[0] != key {
+			t.Fatalf("%q got args %v, want [%d]", c.sql, c.args, key)
+		}
+	}
+}
+
+func TestPgMigrator_LockAndUnlockPropagateErrors(t *testing.T) {
+	lockErr := errors.New("lock failed")
+	unlockErr := errors.New("unlock failed")
+	s := &fakeSession{execErrs: map[string]error{
+		"SELECT pg_advisory_lock($1)":   lockErr,
+		"SELECT pg_advisory_unlock($1)": unlockErr,
+	}}
+	m, _ := newTestMigrator(s)
+	if err := m.Lock(context.Background()); !errors.Is(err, lockErr) {
+		t.Errorf("Lock error = %v, want %v", err, lockErr)
+	}
+	if err := m.Unlock(context.Background()); !errors.Is(err, unlockErr) {
+		t.Errorf("Unlock error = %v, want %v", err, unlockErr)
+	}
+}
+
+func TestPgMigrator_DiscardClosesTheSession(t *testing.T) {
+	m, discards := newTestMigrator(&fakeSession{})
+	m.Discard(context.Background())
+	if *discards != 1 {
+		t.Fatalf("discard called %d times, want 1", *discards)
+	}
+}
+
+func TestPgMigrator_EnsureVersionTableRunsTheDDL(t *testing.T) {
+	s := &fakeSession{}
+	m, _ := newTestMigrator(s)
+	if err := m.EnsureVersionTable(context.Background()); err != nil {
+		t.Fatalf("EnsureVersionTable: %v", err)
+	}
+	if got := s.execSQL(); !reflect.DeepEqual(got, []string{schemaMigrationsDDL}) {
+		t.Fatalf("statements %v, want the schema_migrations DDL", got)
+	}
+	// Re-running the DDL must be harmless, so it has to be IF NOT EXISTS.
+	if !strings.Contains(schemaMigrationsDDL, "IF NOT EXISTS") {
+		t.Error("schema_migrations DDL is not IF NOT EXISTS-guarded")
+	}
+}
+
+func TestPgMigrator_EnsureVersionTablePropagatesErrors(t *testing.T) {
+	want := errors.New("permission denied")
+	s := &fakeSession{execErrs: map[string]error{schemaMigrationsDDL: want}}
+	m, _ := newTestMigrator(s)
+	if err := m.EnsureVersionTable(context.Background()); !errors.Is(err, want) {
+		t.Fatalf("error = %v, want %v", err, want)
+	}
+}
+
+func TestPgMigrator_AppliedVersionsReadsAndClosesRows(t *testing.T) {
+	rows := &fakeRows{values: []string{"0001_first.sql", "0002_second.sql"}}
+	s := &fakeSession{rows: rows}
+	m, _ := newTestMigrator(s)
+
+	got, err := m.AppliedVersions(context.Background())
+	if err != nil {
+		t.Fatalf("AppliedVersions: %v", err)
+	}
+	want := map[string]bool{"0001_first.sql": true, "0002_second.sql": true}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("versions %v, want %v", got, want)
+	}
+	if !rows.closed {
+		t.Error("rows were not closed")
+	}
+	if wantQ := []string{"SELECT version FROM schema_migrations"}; !reflect.DeepEqual(s.queries, wantQ) {
+		t.Fatalf("queries %v, want %v", s.queries, wantQ)
+	}
+}
+
+func TestPgMigrator_AppliedVersionsEmptyTable(t *testing.T) {
+	s := &fakeSession{rows: &fakeRows{}}
+	m, _ := newTestMigrator(s)
+	got, err := m.AppliedVersions(context.Background())
+	if err != nil {
+		t.Fatalf("AppliedVersions: %v", err)
+	}
+	if len(got) != 0 {
+		t.Fatalf("versions %v, want an empty set", got)
+	}
+}
+
+func TestPgMigrator_AppliedVersionsErrors(t *testing.T) {
+	queryErr := errors.New("relation does not exist")
+	scanErr := errors.New("bad column type")
+	rowsErr := errors.New("connection reset mid-read")
+
+	tests := []struct {
+		name string
+		sess *fakeSession
+		want error
+	}{
+		{"query fails", &fakeSession{queryErr: queryErr}, queryErr},
+		{"scan fails", &fakeSession{rows: &fakeRows{values: []string{"x"}, scanErr: scanErr}}, scanErr},
+		// A read that dies part-way must not be reported as a complete set, or
+		// already-applied migrations would be re-run.
+		{"rows.Err after iteration", &fakeSession{rows: &fakeRows{values: []string{"x"}, err: rowsErr}}, rowsErr},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			m, _ := newTestMigrator(tc.sess)
+			got, err := m.AppliedVersions(context.Background())
+			if !errors.Is(err, tc.want) {
+				t.Fatalf("error = %v, want %v", err, tc.want)
+			}
+			if got != nil {
+				t.Fatalf("versions = %v, want nil on error", got)
+			}
+		})
+	}
+}
+
+// One transaction carries both the migration body and its tracking row, so a
+// crash can never leave the schema changed but unrecorded.
+func TestPgMigrator_ApplyRunsBodyAndVersionRowInOneTx(t *testing.T) {
+	const body = "CREATE TABLE IF NOT EXISTS a ();"
+	tx := &fakeTx{}
+	s := &fakeSession{tx: tx}
+	m, _ := newTestMigrator(s)
+
+	if err := m.Apply(context.Background(), "0001_first.sql", body); err != nil {
+		t.Fatalf("Apply: %v", err)
+	}
+	want := []string{body, "INSERT INTO schema_migrations (version) VALUES ($1)"}
+	if got := tx.execSQL(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("statements %v, want %v", got, want)
+	}
+	// The body must be sent with no arguments so pgx uses the simple protocol
+	// and a multi-statement file runs.
+	if len(tx.execs[0].args) != 0 {
+		t.Errorf("migration body sent with args %v, want none", tx.execs[0].args)
+	}
+	if got := tx.execs[1].args; len(got) != 1 || got[0] != "0001_first.sql" {
+		t.Errorf("version row args %v, want [0001_first.sql]", got)
+	}
+	if !tx.committed {
+		t.Error("transaction was not committed")
+	}
+	if tx.rolled != 1 {
+		t.Errorf("Rollback called %d times, want 1 (the no-op deferred rollback)", tx.rolled)
+	}
+}
+
+func TestPgMigrator_ApplyRollsBackOnFailure(t *testing.T) {
+	const body = "CREATE TABLE oops ();"
+	bodyErr := errors.New("syntax error")
+	insertErr := errors.New("duplicate key")
+	const insert = "INSERT INTO schema_migrations (version) VALUES ($1)"
+
+	tests := []struct {
+		name      string
+		execErrs  map[string]error
+		commitErr error
+		want      error
+		wantExecs int
+	}{
+		{"body fails", map[string]error{body: bodyErr}, nil, bodyErr, 1},
+		{"version row fails", map[string]error{insert: insertErr}, nil, insertErr, 2},
+		{"commit fails", nil, errors.New("commit failed"), nil, 2},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			tx := &fakeTx{execErrs: tc.execErrs, commitErr: tc.commitErr}
+			s := &fakeSession{tx: tx}
+			m, _ := newTestMigrator(s)
+
+			err := m.Apply(context.Background(), "0001_first.sql", body)
+			if err == nil {
+				t.Fatal("expected Apply to fail")
+			}
+			if tc.want != nil && !errors.Is(err, tc.want) {
+				t.Fatalf("error = %v, want %v", err, tc.want)
+			}
+			if len(tx.execs) != tc.wantExecs {
+				t.Errorf("%d statements ran, want %d", len(tx.execs), tc.wantExecs)
+			}
+			if tx.committed {
+				t.Error("transaction was committed despite the failure")
+			}
+			if tx.rolled == 0 {
+				t.Error("transaction was not rolled back")
+			}
+		})
+	}
+}
+
+func TestPgMigrator_ApplyPropagatesBeginFailure(t *testing.T) {
+	want := errors.New("cannot start transaction")
+	m, _ := newTestMigrator(&fakeSession{beginErr: want})
+	if err := m.Apply(context.Background(), "0001_first.sql", "SELECT 1;"); !errors.Is(err, want) {
+		t.Fatalf("error = %v, want %v", err, want)
+	}
+}
+
+// migrateSession is the whole runner over a real pgMigrator: lock, DDL, read,
+// apply each file, unlock — in that order, on one session.
+func TestMigrateSession_FullRunOnOneSession(t *testing.T) {
+	tx := &fakeTx{}
+	s := &fakeSession{rows: &fakeRows{}, tx: tx}
+	discards := 0
+	err := migrateSession(context.Background(), s,
+		func(context.Context) { discards++ },
+		testFS(),
+		MigrateOptions{LockName: testLockName})
+	if err != nil {
+		t.Fatalf("migrateSession: %v", err)
+	}
+
+	want := []string{
+		"SELECT pg_advisory_lock($1)",
+		schemaMigrationsDDL,
+		"SELECT pg_advisory_unlock($1)",
+	}
+	if got := s.execSQL(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("session statements %v, want %v", got, want)
+	}
+	// Every file's body plus its version row, all through the one fake tx.
+	if got, wantN := len(tx.execs), 2*len(testNames); got != wantN {
+		t.Fatalf("%d statements in transactions, want %d", got, wantN)
+	}
+	if discards != 0 {
+		t.Errorf("connection discarded %d times after a clean run, want 0", discards)
+	}
+}
+
+func TestMigrateSession_WrapsRunnerErrors(t *testing.T) {
+	s := &fakeSession{execErrs: map[string]error{"SELECT pg_advisory_lock($1)": errors.New("no lock")}}
+	err := migrateSession(context.Background(), s, func(context.Context) {}, testFS(),
+		MigrateOptions{LockName: testLockName})
+	if err == nil {
+		t.Fatal("expected an error")
+	}
+	if !strings.Contains(err.Error(), "migrate schema") {
+		t.Fatalf("error %q is not wrapped with the migration context", err)
+	}
+}
+
+// An unlock that fails must reach the discard func through the real migrator,
+// not just the fake one the runner tests use.
+func TestMigrateSession_UnlockFailureDiscards(t *testing.T) {
+	s := &fakeSession{
+		rows:     &fakeRows{},
+		tx:       &fakeTx{},
+		execErrs: map[string]error{"SELECT pg_advisory_unlock($1)": errors.New("gone")},
+	}
+	discards := 0
+	err := migrateSession(context.Background(), s, func(context.Context) { discards++ },
+		testFS(), MigrateOptions{LockName: testLockName})
+	if err != nil {
+		t.Fatalf("migrateSession: %v", err)
+	}
+	if discards != 1 {
+		t.Fatalf("connection discarded %d times, want 1", discards)
+	}
+}

pg/pgtest/pgtest.go

@@ -0,0 +1,100 @@
+// Package pgtest starts a throwaway Postgres container for integration-style
+// tests. Import it only from _test.go files so testcontainers never reaches a
+// production binary.
+//
+// The estate's CI runs on Kubernetes with no Docker socket, so tests that need
+// a container must skip themselves under -short. SkipIfShort does exactly that,
+// and StartPostgres reports a plain error when no container runtime is
+// reachable, leaving it to the caller whether that is a skip or a failure.
+package pgtest
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/testcontainers/testcontainers-go"
+	tcpostgres "github.com/testcontainers/testcontainers-go/modules/postgres"
+	"github.com/testcontainers/testcontainers-go/wait"
+)
+
+// Image is the Postgres the estate targets; every service runs the same major.
+const Image = "postgres:17-alpine"
+
+const (
+	database = "pgtest"
+	username = "pgtest"
+	password = "pgtest123"
+)
+
+// startTimeout bounds the container's readiness wait.
+const startTimeout = 60 * time.Second
+
+func init() {
+	// The Ryuk reaper container cannot start in every environment (rootless
+	// podman, restricted CI). Callers get an explicit terminate func instead,
+	// so disable Ryuk unless the environment has deliberately enabled it.
+	if _, ok := os.LookupEnv("TESTCONTAINERS_RYUK_DISABLED"); !ok {
+		_ = os.Setenv("TESTCONTAINERS_RYUK_DISABLED", "true")
+	}
+}
+
+// SkipIfShort skips the test under -short. Container-backed tests call it
+// first, which is what keeps `go test -short ./...` green with no Docker.
+func SkipIfShort(t *testing.T) {
+	t.Helper()
+	if testing.Short() {
+		t.Skip("skipping container-backed test in short mode")
+	}
+}
+
+// StartPostgres launches a Postgres container and returns its DSN plus a
+// terminate func the caller must run, even on failure paths.
+func StartPostgres(ctx context.Context) (dsn string, terminate func(), err error) {
+	c, err := tcpostgres.Run(ctx,
+		Image,
+		tcpostgres.WithDatabase(database),
+		tcpostgres.WithUsername(username),
+		tcpostgres.WithPassword(password),
+		testcontainers.WithWaitStrategy(
+			// Postgres opens the port, runs its init scripts, then restarts, so
+			// wait for the readiness log twice to avoid connection resets.
+			wait.ForLog("database system is ready to accept connections").
+				WithOccurrence(2).
+				WithStartupTimeout(startTimeout),
+		),
+	)
+	if err != nil {
+		return "", nil, fmt.Errorf("start postgres container: %w", err)
+	}
+	terminate = func() { _ = c.Terminate(ctx) }
+
+	host, err := c.Host(ctx)
+	if err != nil {
+		terminate()
+		return "", nil, fmt.Errorf("container host: %w", err)
+	}
+	port, err := c.MappedPort(ctx, "5432/tcp")
+	if err != nil {
+		terminate()
+		return "", nil, fmt.Errorf("container port: %w", err)
+	}
+	dsn = fmt.Sprintf("postgres://%s:%s@%s:%s/%s?sslmode=disable",
+		username, password, host, port.Port(), database)
+	return dsn, terminate, nil
+}
+
+// MustStartPostgres is StartPostgres for a TestMain-less test: it skips under
+// -short, fails the test if the container cannot start, and registers cleanup.
+func MustStartPostgres(ctx context.Context, t *testing.T) string {
+	t.Helper()
+	SkipIfShort(t)
+	dsn, terminate, err := StartPostgres(ctx)
+	if err != nil {
+		t.Fatalf("start postgres: %v", err)
+	}
+	t.Cleanup(terminate)
+	return dsn
+}

pg/pool.go

@@ -0,0 +1,44 @@
+package pg
+
+import (
+	"context"
+	"fmt"
+	"io/fs"
+	"log/slog"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// New opens a pgxpool against dsn and verifies it can reach the server before
+// returning. pgxpool connects lazily, so without the ping a bad address only
+// surfaces on the first query, long after startup has reported success.
+//
+// The caller owns the pool and must Close it. log may be nil.
+func New(ctx context.Context, dsn string, log *slog.Logger) (*pgxpool.Pool, error) {
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		return nil, fmt.Errorf("connect postgres: %w", err)
+	}
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("ping postgres: %w", err)
+	}
+	logger(log).Debug("postgres pool ready", "host", pool.Config().ConnConfig.Host)
+	return pool, nil
+}
+
+// NewMigrated opens a pool and brings the schema up to date before returning
+// it, so a service cannot start serving against a half-migrated database. A
+// migration failure closes the pool and returns the error; callers treat it as
+// fatal.
+func NewMigrated(ctx context.Context, dsn string, fsys fs.FS, opts MigrateOptions) (*pgxpool.Pool, error) {
+	pool, err := New(ctx, dsn, opts.Logger)
+	if err != nil {
+		return nil, err
+	}
+	if err := Migrate(ctx, pool, fsys, opts); err != nil {
+		pool.Close()
+		return nil, err
+	}
+	return pool, nil
+}

pg/pool_test.go

@@ -0,0 +1,114 @@
+package pg
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"testing/fstest"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// deadDSN points at a port nothing listens on, so connecting fails immediately
+// and locally: no container, no network, no waiting.
+const deadDSN = "postgres://u:p@127.0.0.1:1/d?sslmode=disable&connect_timeout=2"
+
+func shortCtx(t *testing.T) context.Context {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	t.Cleanup(cancel)
+	return ctx
+}
+
+func TestNew_RejectsAnUnparseableDSN(t *testing.T) {
+	pool, err := New(shortCtx(t), "://not a dsn", nil)
+	if err == nil {
+		pool.Close()
+		t.Fatal("expected an error for an unparseable DSN")
+	}
+	if !strings.Contains(err.Error(), "connect postgres") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+}
+
+// pgxpool connects lazily, so New must ping: without it a wrong address only
+// surfaces on the first query, long after startup reported success.
+func TestNew_PingsBeforeReturning(t *testing.T) {
+	pool, err := New(shortCtx(t), deadDSN, nil)
+	if err == nil {
+		pool.Close()
+		t.Fatal("expected New to fail against an unreachable server")
+	}
+	if !strings.Contains(err.Error(), "ping postgres") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+}
+
+func TestNewMigrated_PropagatesConnectFailure(t *testing.T) {
+	pool, err := NewMigrated(shortCtx(t), deadDSN, testFS(), MigrateOptions{LockName: testLockName})
+	if err == nil {
+		pool.Close()
+		t.Fatal("expected NewMigrated to fail against an unreachable server")
+	}
+	if !strings.Contains(err.Error(), "ping postgres") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+}
+
+// LockName is what makes replicas exclude each other; defaulting it would let a
+// caller silently share a key with an unrelated service, so it is required.
+func TestMigrate_RequiresALockName(t *testing.T) {
+	pool, err := pgxpool.New(shortCtx(t), deadDSN)
+	if err != nil {
+		t.Fatalf("pgxpool.New: %v", err)
+	}
+	t.Cleanup(pool.Close)
+
+	err = Migrate(shortCtx(t), pool, testFS(), MigrateOptions{})
+	if err == nil {
+		t.Fatal("expected Migrate to reject an empty LockName")
+	}
+	if !strings.Contains(err.Error(), "LockName") {
+		t.Fatalf("error %q does not name the missing option", err)
+	}
+}
+
+func TestMigrate_ReportsAcquireFailure(t *testing.T) {
+	pool, err := pgxpool.New(shortCtx(t), deadDSN)
+	if err != nil {
+		t.Fatalf("pgxpool.New: %v", err)
+	}
+	t.Cleanup(pool.Close)
+
+	err = Migrate(shortCtx(t), pool, testFS(), MigrateOptions{LockName: testLockName})
+	if err == nil {
+		t.Fatal("expected Migrate to fail when no connection can be acquired")
+	}
+	if !strings.Contains(err.Error(), "acquire migration connection") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+}
+
+// An empty migration set is a build mistake, not an empty schema; it must fail
+// before any connection work rather than reporting a successful no-op run.
+func TestMigrateSession_EmptySetIsAnError(t *testing.T) {
+	err := migrateSession(context.Background(), &fakeSession{}, func(context.Context) {},
+		fstest.MapFS{}, MigrateOptions{LockName: testLockName})
+	if err == nil {
+		t.Fatal("expected an empty migration set to be an error")
+	}
+}
+
+// A nil Logger is the common case for a library caller; it must not panic.
+func TestLogger_NilIsDiscarding(t *testing.T) {
+	if logger(nil) == nil {
+		t.Fatal("logger(nil) returned nil")
+	}
+	logger(nil).Info("this must not panic")
+
+	custom := testLogger()
+	if logger(custom) != custom {
+		t.Fatal("logger replaced the caller's logger")
+	}
+}