feat(store): version migration ledger entries

docs/internal/accounts-identities-collections-dedup/spec.md

@@ -645,13 +645,19 @@ predicate combines them.
 ```sql
 CREATE TABLE IF NOT EXISTS applied_migrations (
     name       TEXT PRIMARY KEY,
+    version    INTEGER NOT NULL DEFAULT 1,
     applied_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
 );
 ```
 
 DDL changes use `IF NOT EXISTS`. This table records *data*
-migrations that must run exactly once (e.g.
-`legacy_identity_to_per_account`).
+migrations by name and the highest successfully applied implementation
+version (e.g. `legacy_identity_to_per_account`). Existing rows upgraded
+from the legacy schema receive version 1. Callers that omit a version
+request version 1; a migration is applied when the recorded version is
+below the requested minimum, and successful runs advance the recorded
+version monotonically. A failed or cancelled run leaves the previous
+version unchanged.
 
 ## CLI surface
 
@@ -867,16 +873,18 @@ runs the one-time data migration `legacy_identity_to_per_account`:
      identifiers, insert a confirmed identity record with
      `source_signal = manual` if the address is not already
      confirmed.
-2. Insert a row into `applied_migrations` with
-   `name = 'legacy_identity_to_per_account'`.
+2. Insert or advance the `applied_migrations` row with
+   `name = 'legacy_identity_to_per_account'` and `version = 1`.
 3. Log a warning naming the migration and the number of records
    inserted.
 4. Print a one-time CLI notice asking the user to review per-account
    identities via `msgvault identity list`.
 
-After migration, the `[identity]` block is no longer read. The
-migration runs exactly once: subsequent startups see the
-`applied_migrations` row and skip.
+After migration, the `[identity]` block is no longer read. Once the
+version-1 run succeeds, subsequent startups see the
+`applied_migrations` row and skip it. A future reshaped implementation
+can request a higher minimum version and rerun the migration; the ledger
+retains the highest successful version.
 
 If startup happens before any source exists, the migration defers
 until the first source is created, then runs against that source

internal/store/derived_data_revision.go

@@ -1,6 +1,7 @@
 package store
 
 import (
+	"context"
 	"database/sql"
 	"errors"
 	"fmt"
@@ -60,18 +61,15 @@ func (s *Store) AdvanceDerivedDataRevision() error {
 // re-derivation and advances the cache-visible revision. The ledger can never
 // claim a repair is complete without also making an older analytics cache
 // stale.
-func (s *Store) MarkMigrationAppliedWithDerivedDataRevision(name string) error {
+func (s *Store) MarkMigrationAppliedWithDerivedDataRevision(name string, version ...int) error {
+	resolved, err := resolveMigrationVersion(version)
+	if err != nil {
+		return err
+	}
 	return s.withTx(func(tx *loggedTx) error {
 		if err := s.bumpDerivedDataRevision(tx); err != nil {
 			return err
 		}
-		_, err := tx.Exec(
-			s.dialect.InsertOrIgnore(`INSERT OR IGNORE INTO applied_migrations (name) VALUES (?)`),
-			name,
-		)
-		if err != nil {
-			return fmt.Errorf("mark migration %q applied: %w", name, err)
-		}
-		return nil
+		return s.markMigrationAppliedContext(context.Background(), tx, name, resolved)
 	})
 }

internal/store/dialect_pg.go

@@ -621,6 +621,7 @@ func (d *PostgreSQLDialect) FTSRebuildSchema(ctx context.Context, q contextQueri
 //	TEXT → TEXT, DATETIME → TIMESTAMPTZ, JSON → JSONB.
 func (d *PostgreSQLDialect) LegacyColumnMigrations() []ColumnMigration {
 	return []ColumnMigration{
+		{`ALTER TABLE applied_migrations ADD COLUMN IF NOT EXISTS version INTEGER NOT NULL DEFAULT 1`, "applied_migrations.version"},
 		{`ALTER TABLE person_sweep_cursors ADD COLUMN IF NOT EXISTS backstop_upper_key TEXT NOT NULL DEFAULT ''`, "person_sweep_cursors.backstop_upper_key"},
 		{`ALTER TABLE person_sweep_cursors ADD COLUMN IF NOT EXISTS backstop_after_key TEXT NOT NULL DEFAULT ''`, "person_sweep_cursors.backstop_after_key"},
 		{`ALTER TABLE person_sweep_cursors ADD COLUMN IF NOT EXISTS optimistic_document_key TEXT NOT NULL DEFAULT ''`, "person_sweep_cursors.optimistic_document_key"},

internal/store/dialect_sqlite.go

@@ -1922,6 +1922,7 @@ func (d *SQLiteDialect) contentChangedAtDefaultStamps(q querier) (bool, error) {
 // silences these when the column already exists (idempotent migrations).
 func (d *SQLiteDialect) LegacyColumnMigrations() []ColumnMigration {
 	return []ColumnMigration{
+		{`ALTER TABLE applied_migrations ADD COLUMN version INTEGER NOT NULL DEFAULT 1`, "applied_migrations.version"},
 		{`ALTER TABLE person_sweep_cursors ADD COLUMN backstop_upper_key TEXT NOT NULL DEFAULT ''`, "person_sweep_cursors.backstop_upper_key"},
 		{`ALTER TABLE person_sweep_cursors ADD COLUMN backstop_after_key TEXT NOT NULL DEFAULT ''`, "person_sweep_cursors.backstop_after_key"},
 		{`ALTER TABLE person_sweep_cursors ADD COLUMN optimistic_document_key TEXT NOT NULL DEFAULT ''`, "person_sweep_cursors.optimistic_document_key"},

internal/store/init_schema_context_internal_test.go

@@ -3,6 +3,7 @@ package store
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"path/filepath"
 	"strings"
 	"testing"
@@ -11,6 +12,116 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestRunOnceMigrationAtVersionRerunsReshapedMigration(t *testing.T) {
+	require := require.New(t)
+	assert := assert.New(t)
+	st, err := Open(filepath.Join(t.TempDir(), "versioned-migration.db"))
+	require.NoError(err, "open store")
+	t.Cleanup(func() { _ = st.Close() })
+	require.NoError(st.InitSchema(), "initialize schema")
+
+	const name = "reshaped_migration"
+	runs := 0
+	body := func(context.Context) error {
+		runs++
+		return nil
+	}
+	require.NoError(st.runOnceMigration(context.Background(), name, false, body, 1))
+	require.NoError(st.runOnceMigration(context.Background(), name, false, body, 2))
+	require.NoError(st.runOnceMigration(context.Background(), name, false, body, 1))
+	require.NoError(st.runOnceMigration(context.Background(), name, true, body, 1))
+	assert.Equal(3, runs, "version 2 must rerun once and force must still run")
+
+	var version int
+	require.NoError(st.db.QueryRow(`SELECT version FROM applied_migrations WHERE name = ?`, name).Scan(&version))
+	assert.Equal(2, version, "forced lower version must not downgrade the ledger")
+}
+
+func TestMigrationLedgerVersionFailureAndCancellation(t *testing.T) {
+	require := require.New(t)
+	assert := assert.New(t)
+	st, err := Open(filepath.Join(t.TempDir(), "versioned-migration-failure.db"))
+	require.NoError(err, "open store")
+	t.Cleanup(func() { _ = st.Close() })
+	require.NoError(st.InitSchema(), "initialize schema")
+
+	const name = "retryable_migration"
+	require.NoError(st.MarkMigrationApplied(name, 1), "seed the prior migration version")
+	failure := errors.New("migration body failed")
+	err = st.runOnceMigration(context.Background(), name, false, func(context.Context) error {
+		return failure
+	}, 2)
+	require.ErrorIs(err, failure)
+	applied, err := st.IsMigrationApplied(name, 2)
+	require.NoError(err)
+	assert.False(applied, "a failed body must not be marked")
+	applied, err = st.IsMigrationApplied(name, 1)
+	require.NoError(err)
+	assert.True(applied, "a failed higher-version body must preserve version 1")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	err = st.runOnceMigration(ctx, name, false, func(ctx context.Context) error {
+		cancel()
+		return ctx.Err()
+	}, 2)
+	cancel()
+	require.ErrorIs(err, context.Canceled)
+	applied, err = st.IsMigrationApplied(name, 2)
+	require.NoError(err)
+	assert.False(applied, "a cancelled body must not be marked")
+	applied, err = st.IsMigrationApplied(name, 1)
+	require.NoError(err)
+	assert.True(applied, "a cancelled higher-version body must preserve version 1")
+
+	require.NoError(st.runOnceMigration(context.Background(), name, false, func(context.Context) error {
+		return nil
+	}, 2))
+	applied, err = st.IsMigrationApplied(name, 2)
+	require.NoError(err)
+	assert.True(applied, "a retry after failure or cancellation must run")
+}
+
+func TestInitSchemaAddsVersionToLegacyLedger(t *testing.T) {
+	require := require.New(t)
+	assert := assert.New(t)
+	st, err := OpenForTest(filepath.Join(t.TempDir(), "legacy-ledger.db"))
+	require.NoError(err, "open store")
+	t.Cleanup(func() { _ = st.Close() })
+	require.NoError(st.InitSchema(), "initialize current schema")
+
+	const appliedAt = "2020-01-02 03:04:05"
+	_, err = st.db.Exec(`
+		CREATE TABLE applied_migrations_legacy (
+			name TEXT PRIMARY KEY,
+			applied_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+		)`)
+	require.NoError(err, "create legacy ledger")
+	_, err = st.db.Exec(
+		`INSERT INTO applied_migrations_legacy (name, applied_at) VALUES (?, ?)`,
+		migrationPersonInferenceProviderV2, appliedAt)
+	require.NoError(err, "seed legacy ledger")
+	_, err = st.db.Exec(`INSERT INTO applied_migrations_legacy (name, applied_at)
+		SELECT name, applied_at FROM applied_migrations
+		WHERE name != ?`, migrationPersonInferenceProviderV2)
+	require.NoError(err, "copy legacy ledger rows")
+	_, err = st.db.Exec(`DROP TABLE applied_migrations`)
+	require.NoError(err, "replace current ledger")
+	_, err = st.db.Exec(`ALTER TABLE applied_migrations_legacy RENAME TO applied_migrations`)
+	require.NoError(err, "install legacy ledger")
+	var legacyAppliedAt string
+	require.NoError(st.db.QueryRow(`SELECT applied_at FROM applied_migrations WHERE name = ?`,
+		migrationPersonInferenceProviderV2).Scan(&legacyAppliedAt), "read legacy timestamp")
+
+	require.NoError(st.InitSchema(), "upgrade legacy ledger")
+	var version int
+	var gotAppliedAt string
+	require.NoError(st.db.QueryRow(`
+		SELECT version, applied_at FROM applied_migrations WHERE name = ?`,
+		migrationPersonInferenceProviderV2).Scan(&version, &gotAppliedAt))
+	assert.Equal(1, version, "legacy ledger rows must default to version 1")
+	assert.Equal(legacyAppliedAt, gotAppliedAt, "legacy applied timestamp must survive")
+}
+
 // cancelAtStatement cancels the initialisation the moment a chosen statement is
 // about to be issued, by intercepting the store's placeholder-rebind step —
 // which every statement passes through on its way to the driver, whichever

internal/store/migrate_init_schema_ledger_pg_test.go

@@ -0,0 +1,36 @@
+package store
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestInitSchemaAddsVersionToLegacyLedgerPostgres(t *testing.T) {
+	dbURL := skipUnlessPostgresInternal(t)
+	assert := assert.New(t)
+	require := require.New(t)
+	st := newPGStoreInternal(t, dbURL)
+
+	_, err := st.DB().Exec(`ALTER TABLE applied_migrations DROP COLUMN version`)
+	require.NoError(err, "remove the version column from the legacy ledger")
+
+	const name = "legacy_postgres_version_probe"
+	const appliedAt = "2020-01-02 03:04:05+00"
+	_, err = st.DB().Exec(
+		`INSERT INTO applied_migrations (name, applied_at) VALUES ($1, $2)`,
+		name, appliedAt)
+	require.NoError(err, "seed legacy ledger row")
+
+	require.NoError(st.InitSchema(), "upgrade the legacy ledger")
+	var version int
+	var gotAppliedAt time.Time
+	require.NoError(st.DB().QueryRow(
+		`SELECT version, applied_at FROM applied_migrations WHERE name = $1`, name).
+		Scan(&version, &gotAppliedAt))
+	assert.Equal(1, version, "legacy PostgreSQL rows must default to version 1")
+	assert.Equal(time.Date(2020, 1, 2, 3, 4, 5, 0, time.UTC), gotAppliedAt.UTC(),
+		"legacy PostgreSQL timestamp must survive")
+}

internal/store/migrations.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"database/sql"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"time"
@@ -510,38 +511,94 @@ func legacyCalendarOrganizerSelf(
 	return event.Organizer.Self, true
 }
 
-// IsMigrationApplied reports whether the named one-time data migration
-// has already run.
-func (s *Store) IsMigrationApplied(name string) (bool, error) {
-	return s.IsMigrationAppliedContext(context.Background(), name)
+const migrationLedgerVersionColumnDesc = "applied_migrations.version"
+
+func resolveMigrationVersion(versions []int) (int, error) {
+	if len(versions) > 1 {
+		return 0, errors.New("migration version must be omitted or specified once")
+	}
+	if len(versions) == 0 {
+		return 1, nil
+	}
+	if versions[0] < 1 {
+		return 0, fmt.Errorf("migration version must be positive, got %d", versions[0])
+	}
+	return versions[0], nil
+}
+
+// ensureMigrationLedgerVersionColumn adds the ledger version column before
+// InitSchemaContext issues its first version-aware ledger query.
+func (s *Store) ensureMigrationLedgerVersionColumn(ctx context.Context) error {
+	for _, migration := range s.dialect.LegacyColumnMigrations() {
+		if migration.Desc != migrationLedgerVersionColumnDesc {
+			continue
+		}
+		if _, err := s.db.ExecContext(ctx, migration.SQL); err != nil &&
+			!s.dialect.IsDuplicateColumnError(err) {
+			return fmt.Errorf("migrate schema (%s): %w", migration.Desc, err)
+		}
+		return nil
+	}
+	return fmt.Errorf("migration schema entry %q is missing", migrationLedgerVersionColumnDesc)
+}
+
+const markMigrationAppliedSQL = `
+	INSERT INTO applied_migrations (name, version) VALUES (?, ?)
+	ON CONFLICT (name) DO UPDATE SET
+		version = excluded.version,
+		applied_at = CURRENT_TIMESTAMP
+	WHERE applied_migrations.version < excluded.version`
+
+func (s *Store) markMigrationAppliedContext(
+	ctx context.Context, q contextStatementQuerier, name string, version int,
+) error {
+	_, err := q.ExecContext(ctx, markMigrationAppliedSQL, name, version)
+	if err != nil {
+		return fmt.Errorf("mark migration %q applied: %w", name, err)
+	}
+	return nil
+}
+
+// IsMigrationApplied reports whether the named one-time data migration has
+// reached the requested minimum implementation version. An omitted version
+// means version 1.
+func (s *Store) IsMigrationApplied(name string, minimumVersion ...int) (bool, error) {
+	return s.IsMigrationAppliedContext(context.Background(), name, minimumVersion...)
 }
 
 // IsMigrationAppliedContext is the request-aware form of IsMigrationApplied.
-func (s *Store) IsMigrationAppliedContext(ctx context.Context, name string) (bool, error) {
+func (s *Store) IsMigrationAppliedContext(
+	ctx context.Context, name string, minimumVersion ...int,
+) (bool, error) {
+	version, err := resolveMigrationVersion(minimumVersion)
+	if err != nil {
+		return false, err
+	}
 	var count int
-	err := s.db.QueryRowContext(ctx,
-		`SELECT COUNT(*) FROM applied_migrations WHERE name = ?`, name,
+	err = s.db.QueryRowContext(ctx,
+		`SELECT COUNT(*) FROM applied_migrations WHERE name = ? AND version >= ?`,
+		name, version,
 	).Scan(&count)
 	if err != nil {
 		return false, fmt.Errorf("check migration %q: %w", name, err)
 	}
 	return count > 0, nil
 }
 
-// MarkMigrationApplied records that a migration has run. Idempotent.
-func (s *Store) MarkMigrationApplied(name string) error {
-	return s.MarkMigrationAppliedContext(context.Background(), name)
+// MarkMigrationApplied records that a migration has run. It keeps the highest
+// recorded version. An omitted version means version 1.
+func (s *Store) MarkMigrationApplied(name string, version ...int) error {
+	return s.MarkMigrationAppliedContext(context.Background(), name, version...)
 }
 
 // MarkMigrationAppliedContext is the request-aware form of
 // MarkMigrationApplied.
-func (s *Store) MarkMigrationAppliedContext(ctx context.Context, name string) error {
-	_, err := s.db.ExecContext(ctx,
-		s.dialect.InsertOrIgnore(`INSERT OR IGNORE INTO applied_migrations (name) VALUES (?)`),
-		name,
-	)
+func (s *Store) MarkMigrationAppliedContext(
+	ctx context.Context, name string, version ...int,
+) error {
+	resolved, err := resolveMigrationVersion(version)
 	if err != nil {
-		return fmt.Errorf("mark migration %q applied: %w", name, err)
+		return err
 	}
-	return nil
+	return s.markMigrationAppliedContext(ctx, s.db, name, resolved)
 }