feat(store): api/reaper/offsite/migrate 按版本集合校验 schema，库比二进制新时拒绝启动；bootstrap 迁移前失败回滚宿主二进制

internal/pgint/schema_test.go

@@ -0,0 +1,75 @@
+//go:build pgint
+
+package pgint
+
+import (
+	"context"
+	"errors"
+	"os"
+	"strings"
+	"testing"
+
+	"felis.lolicon.best/internal/store"
+)
+
+// The schema guard the api, reaper and offsite copy open the database through: the
+// migrated test database passes, a never-migrated one reads as behind (the missing
+// schema_migrations table is an empty set, not an error), and one that records a
+// version this build does not embed reads as newer.
+func TestSchemaGuard(t *testing.T) {
+	ctx := context.Background()
+	main, err := store.Open(ctx, os.Getenv("FELIS_TEST_PG_URL"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer main.Close()
+	if err := store.CheckSchema(ctx, main); err != nil {
+		t.Fatalf("CheckSchema on the migrated database: %v", err)
+	}
+
+	const schema = "pgint_schema_guard"
+	if _, err := db.ExecContext(ctx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE; CREATE SCHEMA "+schema); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _, _ = db.ExecContext(context.Background(), "DROP SCHEMA IF EXISTS "+schema+" CASCADE") })
+	dsn := os.Getenv("FELIS_TEST_PG_URL")
+	sep := "?"
+	if strings.Contains(dsn, "?") {
+		sep = "&"
+	}
+	other, err := store.Open(ctx, dsn+sep+"search_path="+schema)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer other.Close()
+
+	if err := store.CheckSchema(ctx, other); !errors.Is(err, store.ErrSchemaBehind) {
+		t.Fatalf("CheckSchema on an unmigrated schema = %v, want ErrSchemaBehind", err)
+	}
+
+	ms, err := store.LoadMigrations()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := other.EnsureVersionTable(ctx); err != nil {
+		t.Fatal(err)
+	}
+	for _, m := range ms {
+		if _, err := other.DB().ExecContext(ctx, "INSERT INTO schema_migrations (version, name) VALUES ($1, $2)", m.Version, m.Name); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := store.CheckSchema(ctx, other); err != nil {
+		t.Fatalf("CheckSchema with every version recorded: %v", err)
+	}
+	if _, err := other.DB().ExecContext(ctx, "INSERT INTO schema_migrations (version, name) VALUES (9999, 'from_a_newer_release')"); err != nil {
+		t.Fatal(err)
+	}
+	err = store.CheckSchema(ctx, other)
+	if !errors.Is(err, store.ErrSchemaNewer) || !strings.Contains(err.Error(), "9999") {
+		t.Fatalf("CheckSchema with a newer version recorded = %v, want ErrSchemaNewer naming 9999", err)
+	}
+	if _, err := store.Up(ctx, other, ms); !errors.Is(err, store.ErrSchemaNewer) {
+		t.Fatalf("Up on a newer schema = %v, want ErrSchemaNewer", err)
+	}
+}

internal/store/migrate.go

@@ -84,7 +84,8 @@ func LoadMigrations() ([]Migration, error) {
 
 // Up applies every pending migration in ascending order, exactly once, under
 // the advisory lock. It is safe to run concurrently from multiple replicas: the
-// lock serializes them and AppliedVersions makes the work idempotent.
+// lock serializes them and AppliedVersions makes the work idempotent. It refuses
+// (ErrSchemaNewer) a database that records a version this build does not embed.
 func Up(ctx context.Context, d Driver, migrations []Migration) (applied []int, err error) {
 	if err := d.Lock(ctx); err != nil {
 		return nil, fmt.Errorf("acquire migration lock: %w", err)
@@ -102,6 +103,12 @@ func Up(ctx context.Context, d Driver, migrations []Migration) (applied []int, e
 	if err != nil {
 		return nil, fmt.Errorf("read applied versions: %w", err)
 	}
+	// A version this build does not embed means a newer release migrated the database.
+	// Applying the older build's remaining steps on top would be a guess about a schema
+	// it never saw, so nothing runs.
+	if err := CompareSchema(done, migrations).Newer(); err != nil {
+		return nil, err
+	}
 
 	ordered := append([]Migration(nil), migrations...)
 	sort.Slice(ordered, func(i, j int) bool { return ordered[i].Version < ordered[j].Version })

internal/store/migrate_test.go

@@ -3,6 +3,7 @@ package store_test
 import (
 	"context"
 	"errors"
+	"fmt"
 	"strings"
 	"testing"
 
@@ -155,3 +156,50 @@ func itoa(v int) string {
 	}
 	return string(buf[i:])
 }
+
+func TestUpRefusesADatabaseANewerBuildMigrated(t *testing.T) {
+	d := &recordingDriver{already: map[int]struct{}{1: {}, 2: {}, 3: {}}}
+	ms := []store.Migration{
+		{Version: 1, Name: "a", SQL: "y"},
+		{Version: 2, Name: "b", SQL: "z"},
+	}
+	_, err := store.Up(context.Background(), d, ms)
+	if !errors.Is(err, store.ErrSchemaNewer) {
+		t.Fatalf("Up err = %v, want ErrSchemaNewer", err)
+	}
+	if !strings.Contains(err.Error(), "migration 0003") || !strings.Contains(err.Error(), "up to 0002") {
+		t.Errorf("error does not name the versions: %v", err)
+	}
+	if !d.unlocked {
+		t.Error("the advisory lock was not released")
+	}
+}
+
+func TestCompareSchemaSeparatesPendingFromUnknown(t *testing.T) {
+	ms := []store.Migration{{Version: 1}, {Version: 2}, {Version: 4}}
+	cases := []struct {
+		name    string
+		applied map[int]struct{}
+		pending []int
+		unknown []int
+		err     error
+	}{
+		{"current", map[int]struct{}{1: {}, 2: {}, 4: {}}, nil, nil, nil},
+		{"fresh", map[int]struct{}{}, []int{1, 2, 4}, nil, store.ErrSchemaBehind},
+		{"behind", map[int]struct{}{1: {}}, []int{2, 4}, nil, store.ErrSchemaBehind},
+		// Same row count as "current": a count comparison calls this up to date.
+		{"newer", map[int]struct{}{1: {}, 2: {}, 5: {}}, []int{4}, []int{5}, store.ErrSchemaNewer},
+	}
+	for _, c := range cases {
+		s := store.CompareSchema(c.applied, ms)
+		if fmt.Sprint(s.Pending) != fmt.Sprint(c.pending) || fmt.Sprint(s.Unknown) != fmt.Sprint(c.unknown) {
+			t.Errorf("%s: pending=%v unknown=%v, want %v %v", c.name, s.Pending, s.Unknown, c.pending, c.unknown)
+		}
+		if err := s.Err(); !errors.Is(err, c.err) || (c.err == nil && err != nil) {
+			t.Errorf("%s: Err() = %v, want %v", c.name, err, c.err)
+		}
+		if s.Total != 3 || s.Latest != 4 {
+			t.Errorf("%s: Total=%d Latest=%d, want 3 4", c.name, s.Total, s.Latest)
+		}
+	}
+}

internal/store/schema.go

@@ -0,0 +1,109 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+)
+
+// ErrSchemaNewer marks a database that a newer Felis has migrated: it records versions
+// this build does not embed. Up only rolls forward, so running this build against it
+// would read and write tables whose shape it was never written for.
+var ErrSchemaNewer = errors.New("database schema is newer than this felis build")
+
+// ErrSchemaBehind marks a database that still lacks migrations this build embeds.
+var ErrSchemaBehind = errors.New("database schema is behind this felis build")
+
+// Schema is how a database's recorded migrations line up with the ones embedded in
+// this build. Comparing the two sets, rather than counting, is what tells "behind"
+// from "migrated by a newer release": both can have the same number of rows.
+type Schema struct {
+	Applied int   // embedded migrations the database has
+	Total   int   // embedded migrations
+	Latest  int   // highest embedded version
+	Pending []int // embedded, not yet applied, ascending
+	Unknown []int // applied, but not embedded here, ascending
+}
+
+// CompareSchema lines applied up with migrations.
+func CompareSchema(applied map[int]struct{}, migrations []Migration) Schema {
+	s := Schema{Total: len(migrations)}
+	known := make(map[int]struct{}, len(migrations))
+	for _, m := range migrations {
+		known[m.Version] = struct{}{}
+		s.Latest = max(s.Latest, m.Version)
+		if _, ok := applied[m.Version]; ok {
+			s.Applied++
+		} else {
+			s.Pending = append(s.Pending, m.Version)
+		}
+	}
+	for v := range applied {
+		if _, ok := known[v]; !ok {
+			s.Unknown = append(s.Unknown, v)
+		}
+	}
+	sort.Ints(s.Pending)
+	sort.Ints(s.Unknown)
+	return s
+}
+
+// Newer returns ErrSchemaNewer, with the versions and what to do, when a newer Felis
+// migrated this database; nil otherwise.
+func (s Schema) Newer() error {
+	if len(s.Unknown) == 0 {
+		return nil
+	}
+	return fmt.Errorf("%w: it records migration %s, and this build knows %s. Run the Felis release that migrated it, or restore the pre-migrate snapshot it took (felis db restore)",
+		ErrSchemaNewer, versionList(s.Unknown), knownRange(s.Latest))
+}
+
+// Err is Newer, else ErrSchemaBehind when migrations are pending: the check a server
+// makes before it serves anything from the database.
+func (s Schema) Err() error {
+	if err := s.Newer(); err != nil {
+		return err
+	}
+	if len(s.Pending) > 0 {
+		return fmt.Errorf("%w: migration %s not applied yet. Run `felis migrate up` (the installer does), then start this again",
+			ErrSchemaBehind, versionList(s.Pending))
+	}
+	return nil
+}
+
+// ReadSchema compares what the database records with the embedded migrations.
+func ReadSchema(ctx context.Context, d Driver) (Schema, error) {
+	migrations, err := LoadMigrations()
+	if err != nil {
+		return Schema{}, err
+	}
+	applied, err := d.AppliedVersions(ctx)
+	if err != nil {
+		return Schema{}, fmt.Errorf("read applied migrations: %w", err)
+	}
+	return CompareSchema(applied, migrations), nil
+}
+
+// CheckSchema fails unless the database carries exactly the migrations this build
+// embeds.
+func CheckSchema(ctx context.Context, d Driver) error {
+	s, err := ReadSchema(ctx, d)
+	if err != nil {
+		return err
+	}
+	return s.Err()
+}
+
+func versionList(vs []int) string {
+	parts := make([]string, len(vs))
+	for i, v := range vs {
+		parts[i] = fmt.Sprintf("%04d", v)
+	}
+	return strings.Join(parts, ", ")
+}
+
+func knownRange(latest int) string {
+	return fmt.Sprintf("migrations up to %04d", latest)
+}

internal/store/sqldriver.go

@@ -3,8 +3,10 @@ package store
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 
+	"github.com/jackc/pgx/v5/pgconn"
 	_ "github.com/jackc/pgx/v5/stdlib" // register the "pgx" database/sql driver
 )
 
@@ -56,10 +58,15 @@ func (d *PostgresDriver) EnsureVersionTable(ctx context.Context) error {
 	return err
 }
 
-// AppliedVersions reads the set of recorded versions.
+// AppliedVersions reads the set of recorded versions. A database that has never been
+// migrated has no schema_migrations table yet, which is an empty set.
 func (d *PostgresDriver) AppliedVersions(ctx context.Context) (map[int]struct{}, error) {
 	rows, err := d.db.QueryContext(ctx, "SELECT version FROM schema_migrations")
 	if err != nil {
+		var pgErr *pgconn.PgError
+		if errors.As(err, &pgErr) && pgErr.Code == "42P01" { // undefined_table
+			return map[int]struct{}{}, nil
+		}
 		return nil, err
 	}
 	defer rows.Close()