diff --git a/internal/storage/schema/aux_row_id_backfill.go b/internal/storage/schema/aux_row_id_backfill.go index 3315b1e68f..9c8f8c9418 100644 --- a/internal/storage/schema/aux_row_id_backfill.go +++ b/internal/storage/schema/aux_row_id_backfill.go @@ -4,6 +4,8 @@ import ( "context" "database/sql" "fmt" + "log" + "slices" "sort" "strings" @@ -34,31 +36,97 @@ const auxRowRekeyShippedMainVersion = 51 // auxRowRekeyShippedMainVersion in the crashed pass — without it the // fresh-clone skip above reads the crashed lineage as already converged and // strands the partially re-keyed rows forever (bd-578h9.16). +// +// It means "a rewrite is in flight", nothing else. MigrateUp reads it to exempt +// the aux tables from the pre-existing-dirty guards, because a crashed pass's +// own partial UPDATEs are sitting in the working set — so it must not outlive +// an actual in-flight rewrite. A drift skip (#4380) therefore clears it like +// any completed pass and records auxRowRekeyDriftedKey instead. const auxRowRekeyInProgressKey = "aux_row_rekey_in_progress" -// auxRekeyResumePending reports whether a previous re-key pass on this clone -// recorded the in-progress sentinel and never cleared it. A missing -// local_metadata table means no sentinel: the table is dolt-ignored and -// therefore clone-local, so a fresh clone lacks it until something recreates -// it — setAuxRekeyInProgress creates it on demand. -func auxRekeyResumePending(ctx context.Context, db DBConn) (bool, error) { +// auxRowRekeyDriftedKey is the clone-local record (local_metadata, +// dolt-ignored) of tables the re-key had to skip because their storage carries +// dolthub/dolt#11131 encoding drift (#4380): a comma-separated table list, +// absent when there is nothing skipped. +// +// It exists because the skip completes the pass, so MigrateUp records the +// clone-local marker in that same pass and the marker gate can never re-admit +// the re-key again. This record is what re-admits it — and it names the +// affected tables so the resumed pass touches only those, leaving tables that +// already converged alone. +// +// The record is clone-local because each clone has to discover, retry and +// retire its own skip; the drift itself is not necessarily clone-local, since +// both halves of it (the column's storage-encoding tag and the row chunks it +// disagrees with) are committed data that travels on push/pull — migration 0057 +// documents bd's own 0048 as one way a lineage acquires it. +const auxRowRekeyDriftedKey = "aux_row_rekey_drifted" + +// auxRekeyState is what this clone still owes the re-key: a rewrite that was in +// flight and never finished, and tables skipped for storage drift. Both live in +// local_metadata, which is dolt-ignored and therefore clone-local — and, per +// migration 0030, ephemeral: it is recreated empty by a working-set reset, so +// "no record" is always a legitimate reading. +type auxRekeyState struct { + resume bool + drifted []string +} + +func (s auxRekeyState) pending() bool { return s.resume || len(s.drifted) > 0 } + +// readAuxRekeyState reads both records in one round trip. A missing +// local_metadata table means neither is set: a fresh clone lacks the table +// until something recreates it — setAuxRekeyInProgress creates it on demand. +func readAuxRekeyState(ctx context.Context, db DBConn) (auxRekeyState, error) { + var state auxRekeyState var tableCount int if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'local_metadata'`, ).Scan(&tableCount); err != nil { - return false, err + return state, err } if tableCount == 0 { - return false, nil + return state, nil } - var n int - if err := db.QueryRowContext(ctx, - "SELECT COUNT(*) FROM local_metadata WHERE `key` = ?", - auxRowRekeyInProgressKey).Scan(&n); err != nil { - return false, err + rows, err := db.QueryContext(ctx, + "SELECT `key`, value FROM local_metadata WHERE `key` IN (?, ?)", + auxRowRekeyInProgressKey, auxRowRekeyDriftedKey) + if err != nil { + // This is the one place the re-key reads a TEXT cell of local_metadata, + // so it is the one place a drifted local_metadata could panic the very + // function meant to survive drift. No migration re-encodes that table + // today, but degrade to "nothing recorded" rather than re-break the + // open path if that ever changes. + if isSchemaEncodingDriftErr(err) { + return state, nil + } + return state, err } - return n > 0, nil + defer rows.Close() + for rows.Next() { + var key, value string + if err := rows.Scan(&key, &value); err != nil { + return auxRekeyState{}, err + } + switch key { + case auxRowRekeyInProgressKey: + state.resume = true + case auxRowRekeyDriftedKey: + for _, name := range strings.Split(value, ",") { + if name = strings.TrimSpace(name); name != "" { + state.drifted = append(state.drifted, name) + } + } + } + } + if err := rows.Err(); err != nil { + if isSchemaEncodingDriftErr(err) { + return auxRekeyState{}, nil + } + return auxRekeyState{}, err + } + return state, nil } func setAuxRekeyInProgress(ctx context.Context, db DBConn) error { @@ -83,6 +151,25 @@ func clearAuxRekeyInProgress(ctx context.Context, db DBConn) error { return err } +func setAuxRekeyDrifted(ctx context.Context, db DBConn, tables []string) error { + // Same on-demand create as the sentinel: local_metadata is dolt-ignored, so + // a clone that arrived past 0029 may not have it yet. + if _, err := db.ExecContext(ctx, "CREATE TABLE IF NOT EXISTS local_metadata (`key` VARCHAR(255) PRIMARY KEY, value TEXT NOT NULL DEFAULT '')"); err != nil { + return fmt.Errorf("ensuring local_metadata: %w", err) + } + _, err := db.ExecContext(ctx, + "REPLACE INTO local_metadata (`key`, value) VALUES (?, ?)", + auxRowRekeyDriftedKey, strings.Join(tables, ",")) + return err +} + +func clearAuxRekeyDrifted(ctx context.Context, db DBConn) error { + _, err := db.ExecContext(ctx, + "DELETE FROM local_metadata WHERE `key` = ?", + auxRowRekeyDriftedKey) + return err +} + // auxRekeyTable describes one table covered by the re-key. columns is the // frozen SELECT list of every non-id column, in creation order, with datetime // columns CAST to CHAR server-side so the scanned text is identical across @@ -139,6 +226,14 @@ var auxRekeyTables = []auxRekeyTable{ // mainVersionBefore is the main-source cursor as it stood before this pass's // migrations ran; at or past auxRowRekeyShippedMainVersion the rewrite is // skipped (see that constant). +// +// A table whose storage carries dolthub/dolt#11131 encoding drift cannot be +// scanned at all, so it is skipped and recorded (auxRowRekeyDriftedKey) rather +// than failing the pass; a later pass re-keys just that table, and succeeds +// once the drift is repaired. Like the crash-resume path it re-keys the whole +// table, including rows minted since the skip, so the sooner the retry lands +// the fewer of those there are — which is why it is re-attempted on every +// subsequent migration pass rather than waiting to be asked. func rekeyAuxRowIDs(ctx context.Context, db DBConn, mainVersionBefore int) (bool, error) { pending, err := ignoredSource.pendingVersions(ctx, db) if err != nil { @@ -151,21 +246,47 @@ func rekeyAuxRowIDs(ctx context.Context, db DBConn, mainVersionBefore int) (bool break } } - if !markerPending { - return false, nil - } - resume, err := auxRekeyResumePending(ctx, db) + state, err := readAuxRekeyState(ctx, db) if err != nil { - return false, fmt.Errorf("reading aux rekey sentinel: %w", err) + return false, fmt.Errorf("reading aux rekey state: %w", err) + } + // The marker gate is "once per clone" only for a pass that actually + // converged every table. A pass that skipped one for #11131 drift completed + // — so the marker was recorded in that same MigrateUp — and the marker + // alone would then bar re-entry forever, making the skip permanent. The + // drift record re-admits the pass so it can finish once the storage is + // repaired. + if !markerPending && !state.pending() { + return false, nil } // The fresh-clone skip must not fire on a lineage whose previous pass // crashed mid-rekey: that pass already advanced the main cursor past the // shipped version, but its sentinel proves the rewrite never finished - // (bd-578h9.16). - if mainVersionBefore >= auxRowRekeyShippedMainVersion && !resume { + // (bd-578h9.16). A recorded drift skip says the same thing. + if mainVersionBefore >= auxRowRekeyShippedMainVersion && !state.pending() { return false, nil } + // Once the marker is recorded, the one-time pass has completed: MigrateUp + // records it (ignoredSource.migrate) only after this function returns + // without error, so a clone that still owes work past that point owes it for + // exactly the tables the drift record names. Re-keying any other table there + // would be wrong — those converged in the completed pass, and rows minted + // since carry app-minted ids that are already consistent across clones (see + // above), so rewriting them on this clone alone manufactures the divergence + // the re-key exists to remove. Note this covers a drift-resume pass that + // itself died on some unrelated error and left the sentinel set: it is still + // a post-marker pass, so it still touches only the recorded tables. + tables := auxRekeyTables + if !markerPending { + tables = nil + for _, t := range auxRekeyTables { + if slices.Contains(state.drifted, t.name) { + tables = append(tables, t) + } + } + } + // Sentinel before the first UPDATE: a crash anywhere in the rewrite // leaves it set, so the next pass resumes (the rewrite is idempotent) // instead of recording the marker over partially re-keyed rows. @@ -174,19 +295,103 @@ func rekeyAuxRowIDs(ctx context.Context, db DBConn, mainVersionBefore int) (bool } wrote := false - for _, t := range auxRekeyTables { + var skipped []string + for _, t := range tables { w, err := rekeyAuxRowTable(ctx, db, t) wrote = wrote || w if err != nil { + // #4380: a table carrying dolthub/dolt#11131 schema-encoding drift + // cannot be re-keyed at all — every read of an affected cell panics + // server-side, and no SQL write can repair or remove the row. Before + // this, that aborted the whole MigrateUp pass, which also made the + // database unopenable by any rekey-aware binary: the migration is + // re-attempted on open, so the panic recurred on every start and the + // only build that could read the data was the pre-re-key one. Skip + // the table loudly and let the rest of the pass complete instead. + if isSchemaEncodingDriftErr(err) { + skipped = append(skipped, t.name) + // log, not the TTY-gated progress writer: a piped or CI caller + // discards that writer, and these three lines are the only + // notice that a table's ids stayed divergent. + log.Printf("schema migration: aux row id re-key skipped %q — the table holds rows Dolt cannot decode (schema-encoding drift, gastownhall/beads#4380): %v", + t.name, err) + continue + } return wrote, fmt.Errorf("%s: %w", t.name, err) } } + // The tables this pass covered are exactly the ones the record should now + // name: a full pass covered all four, a post-marker pass covered precisely + // the previously-recorded ones, so in both cases what is still skipped is + // what failed just now. + switch { + case len(skipped) > 0: + if err := setAuxRekeyDrifted(ctx, db, skipped); err != nil { + return wrote, fmt.Errorf("recording aux rekey drift: %w", err) + } + log.Printf("schema migration: %d table(s) kept their old row ids: %s — merges with other clones of this database may duplicate those rows", + len(skipped), strings.Join(skipped, ", ")) + log.Printf("schema migration: repair the storage drift (see gastownhall/beads#4380); the re-key is re-attempted on each later pass that applies schema migrations") + case len(state.drifted) > 0: + if err := clearAuxRekeyDrifted(ctx, db); err != nil { + return wrote, fmt.Errorf("clearing aux rekey drift record: %w", err) + } + // Names what this pass actually covered, which is the recorded set + // minus any entry no longer in auxRekeyTables — a stale name is dropped + // by clearing the record, not by claiming it converged. + if len(tables) > 0 { + covered := make([]string, 0, len(tables)) + for _, t := range tables { + covered = append(covered, t.name) + } + log.Printf("schema migration: aux row id re-key completed for previously skipped table(s): %s", + strings.Join(covered, ", ")) + } + } if err := clearAuxRekeyInProgress(ctx, db); err != nil { return wrote, fmt.Errorf("clearing aux rekey sentinel: %w", err) } return wrote, nil } +// isSchemaEncodingDriftErr reports whether err carries the dolthub/dolt#11131 +// schema-encoding-drift signature: a TEXT/LONGTEXT column whose on-disk +// storage-encoding tag was re-derived without rewriting rows, so decoding a +// cell reads an address of the wrong width and panics inside the storage +// engine. The engine recovers it per query and surfaces +// "Error 1105 (HY000): panic recovered: invalid hash length: 19". Only this +// class is tolerated by the re-key; every other failure still aborts the pass. +// +// Deliberately matched on the bare hash-length phrase, not on ": 19" and not +// on the "panic recovered" wrapper: +// +// - the width varies with the direction of the tag flip — migration 0057 +// documents this same drift from bd's own side (0048 re-widening an +// already-LONGTEXT column under a pre-#11126 embedded engine) and records +// both "invalid hash length: 19" on read and "invalid hash length: 1" on +// insert/merge; +// - the wrapper is added by the SQL engine, so a path that surfaced the +// panic unwrapped would fall through to the abort that leaves affected +// databases unopenable — the failure this whole change exists to end. +// +// The cost of the wider match is bounded: a non-#11131 "invalid hash length" +// (dolt raises it from hash.New generally) means storage this re-key cannot +// read either, so skipping the table and recording it is the same correct +// answer. The pass still aborts on every error that does not carry the phrase. +// +// What it cannot catch: the same drift also produces cells that decode to +// garbage instead of panicking (#4380 reports ~10% of affected cells returning +// a zero payload). Those scan without error, so their rows are re-keyed from +// corrupt content and converge to an id no healthy clone derives. Nothing here +// can detect that — it is a reason the storage repair is still required, not a +// reason to widen the match further. +func isSchemaEncodingDriftErr(err error) bool { + if err == nil { + return false + } + return strings.Contains(strings.ToLower(err.Error()), "invalid hash length") +} + // rekeyAuxRowTable re-derives the ids of one table. The whole table is grouped // by content digest; each digest's rows take the deterministic ids for // ordinals 0..n-1. A row already holding one of its group's target ids keeps diff --git a/internal/storage/schema/aux_row_id_backfill_test.go b/internal/storage/schema/aux_row_id_backfill_test.go index 4f204f8b4f..2204fa2ee5 100644 --- a/internal/storage/schema/aux_row_id_backfill_test.go +++ b/internal/storage/schema/aux_row_id_backfill_test.go @@ -5,6 +5,7 @@ import ( "database/sql" "errors" "regexp" + "strings" "testing" "github.com/DATA-DOG/go-sqlmock" @@ -190,8 +191,9 @@ func TestRekeyAuxRowTableSkipsMissingTable(t *testing.T) { } // TestRekeyAuxRowIDsSkipsWhenMarkerRecorded verifies the clone-local gate: once -// the ignored marker migration is recorded, the re-key never scans a table -// again, so steady-state opens do not churn synced rows. +// the ignored marker migration is recorded and the pass that recorded it left +// no sentinel behind, the re-key never scans a table again, so steady-state +// opens do not churn synced rows. func TestRekeyAuxRowIDsSkipsWhenMarkerRecorded(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { @@ -201,6 +203,7 @@ func TestRekeyAuxRowIDsSkipsWhenMarkerRecorded(t *testing.T) { expectScalar(mock, "SELECT COALESCE(MAX(version), 0) FROM ignored_schema_migrations", "version", auxRowRekeyMarkerVersion) + expectAuxRekeyState(mock, false) // No further expectations: no table may be probed or scanned. wrote, err := rekeyAuxRowIDs(context.Background(), db, auxRowRekeyShippedMainVersion-1) @@ -226,7 +229,7 @@ func TestRekeyAuxRowIDsRunsAllTablesWhenMarkerPending(t *testing.T) { expectScalar(mock, "SELECT COALESCE(MAX(version), 0) FROM ignored_schema_migrations", "version", auxRowRekeyMarkerVersion-1) - expectAuxRekeySentinel(mock, false) + expectAuxRekeyState(mock, false) expectSetAuxRekeySentinel(mock) // Each of the four tables is probed; this mocked world has none of them, // so each probe returns 0 and the loop completes without scanning. @@ -247,18 +250,36 @@ func TestRekeyAuxRowIDsRunsAllTablesWhenMarkerPending(t *testing.T) { } } -// expectAuxRekeySentinel mocks auxRekeyResumePending: the local_metadata -// table-existence probe, then (when the table exists) the sentinel-row count. -func expectAuxRekeySentinel(mock sqlmock.Sqlmock, pending bool) { +// expectAuxRekeyState mocks readAuxRekeyState: the local_metadata +// table-existence probe, then (when the table exists) the one row-read that +// returns both the in-flight sentinel and the #4380 drift record. +func expectAuxRekeyState(mock sqlmock.Sqlmock, resume bool, drifted ...string) { mock.ExpectQuery(`SELECT COUNT\(\*\) FROM INFORMATION_SCHEMA\.TABLES`). WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(1)) - n := 0 - if pending { - n = 1 + rows := sqlmock.NewRows([]string{"key", "value"}) + if resume { + rows.AddRow(auxRowRekeyInProgressKey, "1") } - mock.ExpectQuery(regexp.QuoteMeta("SELECT COUNT(*) FROM local_metadata WHERE `key` = ?")). - WithArgs(auxRowRekeyInProgressKey). - WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(n)) + if len(drifted) > 0 { + rows.AddRow(auxRowRekeyDriftedKey, strings.Join(drifted, ",")) + } + mock.ExpectQuery(regexp.QuoteMeta("SELECT `key`, value FROM local_metadata WHERE `key` IN (?, ?)")). + WithArgs(auxRowRekeyInProgressKey, auxRowRekeyDriftedKey). + WillReturnRows(rows) +} + +func expectSetAuxRekeyDrifted(mock sqlmock.Sqlmock, tables ...string) { + mock.ExpectExec(`CREATE TABLE IF NOT EXISTS local_metadata`). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec(regexp.QuoteMeta("REPLACE INTO local_metadata (`key`, value) VALUES (?, ?)")). + WithArgs(auxRowRekeyDriftedKey, strings.Join(tables, ",")). + WillReturnResult(sqlmock.NewResult(0, 1)) +} + +func expectClearAuxRekeyDrifted(mock sqlmock.Sqlmock) { + mock.ExpectExec(regexp.QuoteMeta("DELETE FROM local_metadata WHERE `key` = ?")). + WithArgs(auxRowRekeyDriftedKey). + WillReturnResult(sqlmock.NewResult(0, 1)) } func expectSetAuxRekeySentinel(mock sqlmock.Sqlmock) { @@ -291,7 +312,7 @@ func TestRekeyAuxRowIDsSkipsConvergedLineage(t *testing.T) { expectScalar(mock, "SELECT COALESCE(MAX(version), 0) FROM ignored_schema_migrations", "version", auxRowRekeyMarkerVersion-1) - expectAuxRekeySentinel(mock, false) + expectAuxRekeyState(mock, false) // No further expectations: marker is pending, but the pre-pass main // cursor at the watershed and no crash sentinel means no table may be // probed or scanned. @@ -323,7 +344,7 @@ func TestRekeyAuxRowIDsResumesAfterCrash(t *testing.T) { expectScalar(mock, "SELECT COALESCE(MAX(version), 0) FROM ignored_schema_migrations", "version", auxRowRekeyMarkerVersion-1) - expectAuxRekeySentinel(mock, true) + expectAuxRekeyState(mock, true) expectSetAuxRekeySentinel(mock) for range auxRekeyTables { expectColumnExists(mock, false) @@ -352,7 +373,7 @@ func TestRekeyAuxRowIDsKeepsSentinelOnFailure(t *testing.T) { expectScalar(mock, "SELECT COALESCE(MAX(version), 0) FROM ignored_schema_migrations", "version", auxRowRekeyMarkerVersion-1) - expectAuxRekeySentinel(mock, false) + expectAuxRekeyState(mock, false) expectSetAuxRekeySentinel(mock) // First table probe fails; no DELETE of the sentinel may follow. mock.ExpectQuery(`SELECT COUNT\(\*\) FROM INFORMATION_SCHEMA\.COLUMNS`). diff --git a/internal/storage/schema/aux_row_id_drift_test.go b/internal/storage/schema/aux_row_id_drift_test.go new file mode 100644 index 0000000000..27eb6ce464 --- /dev/null +++ b/internal/storage/schema/aux_row_id_drift_test.go @@ -0,0 +1,294 @@ +package schema + +import ( + "bytes" + "context" + "errors" + "fmt" + "log" + "regexp" + "strings" + "testing" + + "github.com/DATA-DOG/go-sqlmock" +) + +var eventsTable = auxRekeyTables[0] + +// driftErr is the error a drifted table returns: dolt recovers the decode panic +// per query and surfaces it as a 1105. +var driftErr = errors.New("Error 1105 (HY000): panic recovered: invalid hash length: 19") + +func expectEventsSelect(mock sqlmock.Sqlmock) *sqlmock.ExpectedQuery { + return mock.ExpectQuery(regexp.QuoteMeta( + fmt.Sprintf("SELECT id, %s FROM %s", eventsTable.columns, eventsTable.name))) +} + +// captureLog redirects the standard logger, which is where the drift notices +// go: the package's own stderr writer is TTY-gated to io.Discard, and these are +// data-integrity warnings that a piped or CI caller must still see. +func captureLog(t *testing.T) *bytes.Buffer { + t.Helper() + var buf bytes.Buffer + origOut, origFlags := log.Writer(), log.Flags() + log.SetOutput(&buf) + log.SetFlags(0) + t.Cleanup(func() { log.SetOutput(origOut); log.SetFlags(origFlags) }) + return &buf +} + +// The re-key tolerates exactly the dolthub/dolt#11131 schema-encoding-drift +// signature and nothing else: every other failure must still abort the pass. +func TestIsSchemaEncodingDriftErr(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"server 1105 drift panic", driftErr, true}, + { + "wrapped drift panic", + fmt.Errorf("scan rows of events: %w", driftErr), + true, + }, + { + // Migration 0057 records this width for the opposite tag flip. + "insert/merge side of the flip", + errors.New("Error 1105 (HY000): panic recovered: invalid hash length: 1"), + true, + }, + { + "unwrapped panic text", + errors.New("panic: invalid hash length: 19"), + true, + }, + {"plain sql error", errors.New("Error 1062 (23000): duplicate entry"), false}, + {"connection drop", errors.New("invalid connection"), false}, + {"unrelated panic", errors.New("Error 1105 (HY000): panic recovered: index out of range"), false}, + {"lock timeout", errors.New("Error 1205 (HY000): lock wait timeout exceeded"), false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := isSchemaEncodingDriftErr(c.err); got != c.want { + t.Fatalf("isSchemaEncodingDriftErr(%v) = %v, want %v", c.err, got, c.want) + } + }) + } +} + +// TestRekeyAuxRowIDsSkipsDriftedTable covers the #4380 unblock: a table whose +// rows Dolt cannot decode is skipped so the rest of the migration pass +// completes, instead of aborting it — which left affected databases unopenable, +// because the pass is re-attempted on every open. The skipped table is recorded +// so a later pass can finish the job, and the in-progress sentinel is cleared +// like any completed pass: it means "a rewrite is in flight", and MigrateUp +// relaxes its dirty-table guards while it is set. +func TestRekeyAuxRowIDsSkipsDriftedTable(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + logged := captureLog(t) + + expectScalar(mock, "SELECT COALESCE(MAX(version), 0) FROM ignored_schema_migrations", + "version", auxRowRekeyMarkerVersion-1) + expectAuxRekeyState(mock, false) + expectSetAuxRekeySentinel(mock) + // events carries the drift: its scan panics server-side. + expectColumnExists(mock, true) + expectEventsSelect(mock).WillReturnError(driftErr) + // The remaining three tables still run. + for range auxRekeyTables[1:] { + expectColumnExists(mock, false) + } + expectSetAuxRekeyDrifted(mock, "events") + expectClearAuxRekeySentinel(mock) + + if _, err := rekeyAuxRowIDs(context.Background(), db, auxRowRekeyShippedMainVersion-1); err != nil { + t.Fatalf("drift must not abort the pass: %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sql expectations: %v", err) + } + if got := logged.String(); !strings.Contains(got, `skipped "events"`) || + !strings.Contains(got, "may duplicate those rows") { + t.Errorf("log = %q\nwant a skip warning naming events and its merge consequence", got) + } +} + +// TestRekeyAuxRowIDsSkipsDriftDuringRewrite covers the other half of the drift +// surface: #4380 reports that UPDATE of an affected row panics just like a read, +// so the drift can land mid-rewrite, after some ids have already changed. The +// half-re-keyed table must be recorded and committed rather than failing the +// pass — it stays resumable because a row already holding one of its group's +// target ids keeps it, so a later attempt never swaps ids within a group. +func TestRekeyAuxRowIDsSkipsDriftDuringRewrite(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + captureLog(t) + + expectScalar(mock, "SELECT COALESCE(MAX(version), 0) FROM ignored_schema_migrations", + "version", auxRowRekeyMarkerVersion-1) + expectAuxRekeyState(mock, false) + expectSetAuxRekeySentinel(mock) + expectColumnExists(mock, true) + // The scan succeeds — these are the readable-but-drifted rows — and the + // rewrite panics instead. + expectEventsSelect(mock). + WillReturnRows(sqlmock.NewRows([]string{ + "id", "issue_id", "event_type", "actor", "old_value", "new_value", "comment", "created_at", + }).AddRow("random-a", "bd-1", "status", "steve", "open", "closed", "", "2026-06-09 12:00:00")) + mock.ExpectExec(regexp.QuoteMeta("UPDATE events SET id = ? WHERE id = ?")). + WithArgs(sqlmock.AnyArg(), "random-a"). + WillReturnError(driftErr) + for range auxRekeyTables[1:] { + expectColumnExists(mock, false) + } + expectSetAuxRekeyDrifted(mock, "events") + expectClearAuxRekeySentinel(mock) + + wrote, err := rekeyAuxRowIDs(context.Background(), db, auxRowRekeyShippedMainVersion-1) + if err != nil { + t.Fatalf("drift mid-rewrite must not abort the pass: %v", err) + } + if !wrote { + t.Error("expected wrote=true so MigrateUp commits the partially re-keyed table") + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sql expectations: %v", err) + } +} + +// TestRekeyAuxRowIDsAbortsOnNonDriftError: the tolerance is narrow. An ordinary +// table failure must still propagate and keep the sentinel, exactly as before +// #4380 — otherwise it would swallow real migration bugs. +func TestRekeyAuxRowIDsAbortsOnNonDriftError(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + captureLog(t) + + expectScalar(mock, "SELECT COALESCE(MAX(version), 0) FROM ignored_schema_migrations", + "version", auxRowRekeyMarkerVersion-1) + expectAuxRekeyState(mock, false) + expectSetAuxRekeySentinel(mock) + expectColumnExists(mock, true) + expectEventsSelect(mock).WillReturnError(errors.New("Error 1062 (23000): duplicate entry")) + + if _, err := rekeyAuxRowIDs(context.Background(), db, auxRowRekeyShippedMainVersion-1); err == nil { + t.Fatal("expected a non-drift table failure to propagate") + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sql expectations: %v", err) + } +} + +// TestRekeyAuxRowIDsResumesDriftedTableOnly is the counterpart to the skip. The +// pass that skipped completed, so MigrateUp recorded the clone-local marker and +// the marker gate can no longer re-admit the re-key; the drift record must, or +// the skipped table would keep its 0037 per-clone-random ids forever, even +// after the storage drift is repaired. +// +// It must also re-key ONLY the recorded table: the other three converged in the +// first pass, and rows minted since then carry app-minted ids that are already +// consistent across clones — re-keying those here would manufacture the +// divergence the whole pass exists to remove. +func TestRekeyAuxRowIDsResumesDriftedTableOnly(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + captureLog(t) + + // Marker recorded, no crash sentinel — the post-skip steady state. + expectScalar(mock, "SELECT COALESCE(MAX(version), 0) FROM ignored_schema_migrations", + "version", auxRowRekeyMarkerVersion) + expectAuxRekeyState(mock, false, "events") + expectSetAuxRekeySentinel(mock) + // Exactly one table probe: any second one is an unexpected call, which is + // what proves the other three are left alone. + expectColumnExists(mock, false) + expectClearAuxRekeyDrifted(mock) + expectClearAuxRekeySentinel(mock) + + // Pre-pass cursor past the watershed too, so the drift record is overriding + // both the marker gate and the fresh-clone gate. + wrote, err := rekeyAuxRowIDs(context.Background(), db, auxRowRekeyShippedMainVersion) + if err != nil { + t.Fatalf("rekeyAuxRowIDs: %v", err) + } + if wrote { + t.Error("expected wrote=false when the resumed table has nothing to re-key") + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sql expectations: %v", err) + } +} + +// TestRekeyAuxRowIDsResumeStaysScopedAfterCrash: a drift-resume pass sets the +// in-flight sentinel like any other, so an unrelated failure (Ctrl-C, lock +// timeout, dropped connection) can leave the sentinel set on a clone whose +// marker is already recorded. That state must NOT be read as the bd-578h9.16 +// crash-resume and widened back to all four tables: the other three converged +// in the completed pass, and re-keying rows minted since — on this clone alone — +// is exactly the cross-clone divergence the scoping exists to prevent. +func TestRekeyAuxRowIDsResumeStaysScopedAfterCrash(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + captureLog(t) + + expectScalar(mock, "SELECT COALESCE(MAX(version), 0) FROM ignored_schema_migrations", + "version", auxRowRekeyMarkerVersion) + // Marker recorded AND sentinel set: a drift-resume that died partway. + expectAuxRekeyState(mock, true, "events") + expectSetAuxRekeySentinel(mock) + expectColumnExists(mock, false) + expectClearAuxRekeyDrifted(mock) + expectClearAuxRekeySentinel(mock) + + if _, err := rekeyAuxRowIDs(context.Background(), db, auxRowRekeyShippedMainVersion); err != nil { + t.Fatalf("rekeyAuxRowIDs: %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sql expectations: %v", err) + } +} + +// TestRekeyAuxRowIDsKeepsDriftRecordWhileUnrepaired: a resumed pass that finds +// the storage still drifted must re-record the table rather than lose it, so +// the retry survives arbitrarily many passes before the repair happens. +func TestRekeyAuxRowIDsKeepsDriftRecordWhileUnrepaired(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + captureLog(t) + + expectScalar(mock, "SELECT COALESCE(MAX(version), 0) FROM ignored_schema_migrations", + "version", auxRowRekeyMarkerVersion) + expectAuxRekeyState(mock, false, "events") + expectSetAuxRekeySentinel(mock) + expectColumnExists(mock, true) + expectEventsSelect(mock).WillReturnError(driftErr) + expectSetAuxRekeyDrifted(mock, "events") + expectClearAuxRekeySentinel(mock) + + if _, err := rekeyAuxRowIDs(context.Background(), db, auxRowRekeyShippedMainVersion); err != nil { + t.Fatalf("rekeyAuxRowIDs: %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sql expectations: %v", err) + } +} diff --git a/internal/storage/schema/lock_test.go b/internal/storage/schema/lock_test.go index ffa22bd28c..6577d72ff8 100644 --- a/internal/storage/schema/lock_test.go +++ b/internal/storage/schema/lock_test.go @@ -256,8 +256,14 @@ func expectOnePendingMigration(t *testing.T, mock sqlmock.Sqlmock) { expectColumnExists(mock, false) expectColumnExists(mock, false) // rekeyAuxRowIDs reads the ignored cursor to see whether its clone-local - // marker is pending; at latest it is not, so the re-key no-ops. + // marker is pending; at latest it is not. It then reads the clone-local + // re-key state — the crash sentinel and the #4380 drift record, either of + // which would re-admit the pass; this mocked world has no local_metadata + // table, so the read stops at the table-existence probe and the re-key + // no-ops. expectScalar(mock, "SELECT COALESCE(MAX(version), 0) FROM ignored_schema_migrations", "version", latestIgnored) + mock.ExpectQuery(`SELECT COUNT\(\*\) FROM INFORMATION_SCHEMA\.TABLES`). + WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0)) mock.ExpectExec("(?s)^CREATE TABLE IF NOT EXISTS ignored_schema_migrations"). WillReturnResult(sqlmock.NewResult(0, 0)) expectContentHashColumnExists(mock) diff --git a/internal/storage/schema/schema.go b/internal/storage/schema/schema.go index feda2f8a70..f0f3eafdf8 100644 --- a/internal/storage/schema/schema.go +++ b/internal/storage/schema/schema.go @@ -459,13 +459,26 @@ func MigrateUp(ctx context.Context, db DBConn) (int, error) { // pre-existing user writes: dropping them from dirtyBefore exempts them // from the changed-signature guard (the resumed rekey is about to change // them) and lets stageSchemaTables commit them with the rest of the pass. - if resuming, err := auxRekeyResumePending(ctx, db); err != nil { - return 0, fmt.Errorf("reading aux rekey sentinel: %w", err) - } else if resuming { + // + // A table skipped for #11131 encoding drift (#4380) needs the same + // exemption on the pass that retries it, and needs it more: the + // changed-signature guard below reads dirty tables through dolt_diff, which + // decodes exactly the cells that panic — so leaving a drifted table in + // dirtyBefore would fail the pass on the read, re-creating the unopenable + // database this exemption path exists to rescue. Only the recorded tables + // are exempted there, since only those will be touched. + auxRekeyOwed, err := readAuxRekeyState(ctx, db) + if err != nil { + return 0, fmt.Errorf("reading aux rekey state: %w", err) + } + if auxRekeyOwed.resume { for _, t := range auxRekeyTables { delete(dirtyBefore, t.name) } } + for _, name := range auxRekeyOwed.drifted { + delete(dirtyBefore, name) + } touchedDirtyTables, err := mainSource.pendingMigrationDirtyTables(ctx, db, dirtyBefore) if err != nil { return 0, fmt.Errorf("checking dirty tables against pending migrations: %w", err)