carlos / platform Public

registry: the import journal joins the snapshot digest; swap and close guards (Plan 7 Task 1 fix round 1)

SnapshotFingerprint hashes import_journal, every column in host order, only
when it has rows, so a deleted cleanup-pending row can no longer linger in
the last snapshot while an empty journal digests exactly as before (pinned by
a golden from f20d2c5e). SwapForImport refuses a claim-owned, held or gated
original and one its JSON round trip would change; CloseImportGate keeps the
stored fence generation when already rolling-back. Atomicity is proven with a
mid-transaction fault, and the tests check every read's error.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 2 weeks ago 6955bbaecc58677a77503856a14c9de0f47e35cb parent f20d2c5

3 files changed, +303 −45

internal/registry/importjournal.go +52 −6
internal/registry/importjournal_test.go +248 −39
internal/registry/registry.go +3 −0

CI cancelled — superseded by a newer push — run details

diff --git a/internal/registry/importjournal.go b/internal/registry/importjournal.go

index 446f0130b..cec53db5f 100644

--- a/internal/registry/importjournal.go

+++ b/internal/registry/importjournal.go

@@ -6,6 +6,7 @@ import (

"encoding/json"

"errors"

"fmt"

+ "io"

"reflect"

)

@@ -108,6 +109,36 @@ func (r *Registry) ImportJournals() ([]ImportJournal, error) {

return out, rows.Err()

}

+// hashImportJournal folds the journal into SnapshotFingerprint's digest:

+// the journal is box-local state nothing in the bucket can rebuild, and a

+// snapshot that kept a deleted cleanup-pending row would, restored cold,

+// resume a finished rollback's bucket cleanup over a later instance on the

+// same host. Every column, in host order, under its own section marker. An

+// empty journal writes nothing, so a box with no import in flight digests

+// exactly as it did before the journal existed.

+func hashImportJournal(db *sql.DB, h io.Writer) error {

+ rows, err := db.Query(`SELECT ` + journalCols + ` FROM import_journal ORDER BY host`)

+ if err != nil {

+ return err

+ }

+ defer rows.Close()

+ first := true

+ for rows.Next() {

+ var host, seedID, orig, recordKey, state string

+ var fenceGen uint64

+ var at int64

+ if err := rows.Scan(&host, &seedID, &orig, &recordKey, &state, &fenceGen, &at); err != nil {

+ return err

+ }

+ if first {

+ fmt.Fprint(h, "\x1dimport_journal\x1e")

+ first = false

+ }

+ fmt.Fprintf(h, "%q\x1f%q\x1f%q\x1f%q\x1f%q\x1f%d\x1f%d\x1e", host, seedID, orig, recordKey, state, fenceGen, at)

+ }

+ return rows.Err()

// inTx runs fn in one BEGIN IMMEDIATE transaction at synchronous=FULL (the

// claim transactions' form). IMMEDIATE takes the write lock before the first

// read, so another process's write waits on busy_timeout instead of failing

@@ -147,15 +178,24 @@ func insertRouteTx(tx txer, t Route) error {

// no journal row for the host and the stored route still exactly

// j.Original, it replaces the route with next (Upsert's defaults applied)

// and writes the journal row at swapped. next must be the same host and

-// carry its gate.

+// carry its gate. The original must be one the journal can put back: not

+// claim-owned (claims are off for imports, spec §3.4), not held by an

+// operation, not itself gated, and unchanged by its JSON round trip.

func (r *Registry) SwapForImport(j ImportJournal, next Route) error {

if j.Host == "" || j.SeedID == "" || next.Host != j.Host || j.Original.Host != j.Host || next.ImportGate == "" {

return fmt.Errorf("%w: the swap for %s is not well formed", ErrImportJournal, j.Host)

}

+ if j.Original.ClaimOwnership != (ClaimOwnership{}) || j.Original.HoldOp != "" || j.Original.ImportGate != "" {

+ return fmt.Errorf("%w: the route at %s is not one an import replaces", ErrImportJournal, j.Host)

+ }

orig, err := json.Marshal(j.Original)

if err != nil {

return err

}

+ var back Route

+ if err := json.Unmarshal(orig, &back); err != nil || !reflect.DeepEqual(back, j.Original) {

+ return fmt.Errorf("%w: the route at %s would not restore exactly", ErrImportJournal, j.Host)

+ }

return r.inTx(func(tx txer) error {

if cur, err := scanJournal(tx, j.Host); err != nil {

return err

@@ -201,16 +241,22 @@ func journalFor(tx txer, host, seedID string, states ...string) (*ImportJournal,

// CloseImportGate is the rollback's step 1 (spec §5.6): the journal says

// rolling-back with the fence generation, and the gate refuses even verify

-// requests, in one transaction.

+// requests, in one transaction. Run again on a row already rolling-back, it

+// re-closes the gate and keeps the stored generation (fenceGen is ignored):

+// step 2 recognises this rollback's own fence by that generation, so the

+// caller reads FenceGen back from the journal rather than trusting its own.

func (r *Registry) CloseImportGate(host, seedID string, fenceGen uint64) error {

return r.inTx(func(tx txer) error {

- if _, err := journalFor(tx, host, seedID, JournalSwapped, JournalRollingBack); err != nil {

+ j, err := journalFor(tx, host, seedID, JournalSwapped, JournalRollingBack)

+ if err != nil {

return err

}

- if _, err := tx.Exec(`UPDATE import_journal SET state = ?, fence_gen = ? WHERE host = ?`, JournalRollingBack, fenceGen, host); err != nil {

- return err

+ if j.State == JournalSwapped {

+ if _, err := tx.Exec(`UPDATE import_journal SET state = ?, fence_gen = ? WHERE host = ?`, JournalRollingBack, fenceGen, host); err != nil {

+ return err

+ }

}

- _, err := tx.Exec(`UPDATE routes SET import_gate = ? WHERE host = ?`, ImportGateClosed, host)

+ _, err = tx.Exec(`UPDATE routes SET import_gate = ? WHERE host = ?`, ImportGateClosed, host)

return err

})

}

diff --git a/internal/registry/importjournal_test.go b/internal/registry/importjournal_test.go

index 0ef0d9206..72bdf3800 100644

--- a/internal/registry/importjournal_test.go

+++ b/internal/registry/importjournal_test.go

@@ -26,23 +26,41 @@ func execRow() Route {

Channel: "stable", Hibernate: true, Provisioned: true, ImportGate: "abc123", CreatedAt: 1700000001}

}

+func mustJournal(t *testing.T, reg *Registry, host string) *ImportJournal {

+ t.Helper()

+ j, err := reg.ImportJournal(host)

+ if err != nil {

+ t.Fatal(err)

+ }

+ return j

+func mustDigest(t *testing.T, f func() (string, error)) string {

+ t.Helper()

+ d, err := f()

+ if err != nil {

+ t.Fatal(err)

+ }

+ return d

func TestSwapCommitAndRestore(t *testing.T) {

reg := openTestRegistry(t)

old := legacyRow()

if err := reg.Upsert(old); err != nil {

t.Fatal(err)

}

- stored, _ := reg.Get(old.Host)

- j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: "apps/bbw/shop/instances/old.shop.example.com.json", Original: *stored, At: 1}

+ stored := mustRoute(t, reg, old.Host)

+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: "apps/bbw/shop/instances/old.shop.example.com.json", Original: stored, At: 1}

if err := reg.SwapForImport(j, execRow()); err != nil {

t.Fatal(err)

}

- got, _ := reg.Get(old.Host)

+ got := mustRoute(t, reg, old.Host)

if got.Kind != "instance" || got.ImportGate != "abc123" || got.Account != "bbw" {

t.Fatalf("after swap: %+v", got)

}

- jr, _ := reg.ImportJournal(old.Host)

- if jr == nil || jr.State != JournalSwapped || !reflect.DeepEqual(jr.Original, *stored) || jr.RecordKey != j.RecordKey || jr.At != 1 {

+ jr := mustJournal(t, reg, old.Host)

+ if jr == nil || jr.State != JournalSwapped || !reflect.DeepEqual(jr.Original, stored) || jr.RecordKey != j.RecordKey || jr.At != 1 {

t.Fatalf("journal = %+v", jr)

}

// A second swap for the host is refused: the journal is there.

@@ -53,24 +71,31 @@ func TestSwapCommitAndRestore(t *testing.T) {

if err := reg.CloseImportGate(old.Host, j.SeedID, 77); err != nil {

t.Fatal(err)

}

- if got, _ := reg.Get(old.Host); got.ImportGate != ImportGateClosed {

+ if got := mustRoute(t, reg, old.Host); got.ImportGate != ImportGateClosed {

t.Fatalf("gate = %q, want closed", got.ImportGate)

}

// Commit is only from swapped: a rolling-back import cannot commit.

if err := reg.CommitImport(old.Host, j.SeedID); !errors.Is(err, ErrImportJournal) {

t.Fatalf("commit while rolling back: %v", err)

}

- // Closing again resumes (idempotent) and keeps the latest generation.

- if err := reg.CloseImportGate(old.Host, j.SeedID, 77); err != nil {

+ // Closing again resumes (idempotent) and keeps the stored generation:

+ // step 2 recognises its own fence by it.

+ if err := reg.CloseImportGate(old.Host, j.SeedID, 78); err != nil {

t.Fatal(err)

}

+ if jr := mustJournal(t, reg, old.Host); jr == nil || jr.State != JournalRollingBack || jr.FenceGen != 77 {

+ t.Fatalf("journal after a second close = %+v", jr)

+ }

+ if got := mustRoute(t, reg, old.Host); got.ImportGate != ImportGateClosed {

+ t.Fatalf("gate after a second close = %q", got.ImportGate)

+ }

if err := reg.RestoreImportRoute(old.Host, j.SeedID); err != nil {

t.Fatal(err)

}

- if got, _ := reg.Get(old.Host); !reflect.DeepEqual(*got, *stored) {

+ if got := mustRoute(t, reg, old.Host); !reflect.DeepEqual(got, stored) {

t.Fatalf("restored %+v, want %+v", got, stored)

}

- if jr, _ := reg.ImportJournal(old.Host); jr == nil || jr.State != JournalCleanupPending || jr.FenceGen != 77 {

+ if jr := mustJournal(t, reg, old.Host); jr == nil || jr.State != JournalCleanupPending || jr.FenceGen != 77 {

t.Fatalf("journal after restore = %+v", jr)

}

// cleanup-pending never goes back: no second restore, no gate close.

@@ -86,7 +111,7 @@ func TestSwapCommitAndRestore(t *testing.T) {

if err := reg.DeleteImportJournal(old.Host, j.SeedID); err != nil {

t.Fatal(err)

}

- if jr, _ := reg.ImportJournal(old.Host); jr != nil {

+ if jr := mustJournal(t, reg, old.Host); jr != nil {

t.Fatalf("journal after delete = %+v", jr)

}

if err := reg.DeleteImportJournal(old.Host, j.SeedID); !errors.Is(err, ErrImportJournal) {

@@ -100,29 +125,29 @@ func TestCommitImportClearsTheGateAndTheJournal(t *testing.T) {

if err := reg.Upsert(old); err != nil {

t.Fatal(err)

}

- stored, _ := reg.Get(old.Host)

- j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: *stored}

+ stored := mustRoute(t, reg, old.Host)

+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: stored}

if err := reg.SwapForImport(j, execRow()); err != nil {

t.Fatal(err)

}

if err := reg.CommitImport(old.Host, "sd_bbbbbbbbbbbbbbbbbbbbbbbbbb"); !errors.Is(err, ErrImportJournal) {

t.Fatalf("commit with another id: %v", err)

}

- before, _ := reg.Fingerprint()

+ before := mustDigest(t, reg.Fingerprint)

if err := reg.CommitImport(old.Host, j.SeedID); err != nil {

t.Fatal(err)

}

- got, _ := reg.Get(old.Host)

- jr, _ := reg.ImportJournal(old.Host)

- after, _ := reg.Fingerprint()

+ got := mustRoute(t, reg, old.Host)

+ jr := mustJournal(t, reg, old.Host)

+ after := mustDigest(t, reg.Fingerprint)

if got.ImportGate != "" || jr != nil || before == after {

t.Fatalf("after commit: gate %q journal %+v fingerprint moved %v", got.ImportGate, jr, before != after)

}

// The committed row is the exec row, ungated.

want := execRow()

want.ImportGate = ""

- if !reflect.DeepEqual(*got, want) {

- t.Fatalf("committed row %+v, want %+v", *got, want)

+ if !reflect.DeepEqual(got, want) {

+ t.Fatalf("committed row %+v, want %+v", got, want)

}

@@ -132,20 +157,20 @@ func TestSwapForImportRefusesAChangedRow(t *testing.T) {

if err := reg.Upsert(old); err != nil {

t.Fatal(err)

}

- stored, _ := reg.Get(old.Host)

- changed := *stored

+ stored := mustRoute(t, reg, old.Host)

+ changed := stored

changed.Addr = "127.0.0.1:9999"

if err := reg.Upsert(changed); err != nil {

t.Fatal(err)

}

- j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: *stored}

+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: stored}

if err := reg.SwapForImport(j, execRow()); !errors.Is(err, ErrImportJournal) {

t.Fatalf("swap over a changed row: %v", err)

}

- if got, _ := reg.Get(old.Host); got.Addr != "127.0.0.1:9999" {

+ if got := mustRoute(t, reg, old.Host); got.Addr != "127.0.0.1:9999" {

t.Fatal("the changed row was replaced")

}

- if jr, _ := reg.ImportJournal(old.Host); jr != nil {

+ if jr := mustJournal(t, reg, old.Host); jr != nil {

t.Fatalf("a refused swap left a journal row: %+v", jr)

}

@@ -160,8 +185,8 @@ func TestSwapForImportRefusesAMissingRowAndAMalformedSwap(t *testing.T) {

if err := reg.Upsert(old); err != nil {

t.Fatal(err)

}

- stored, _ := reg.Get(old.Host)

- j.Original = *stored

+ stored := mustRoute(t, reg, old.Host)

+ j.Original = stored

ungated := execRow()

ungated.ImportGate = ""

other := execRow()

@@ -180,7 +205,7 @@ func TestSwapForImportRefusesAMissingRowAndAMalformedSwap(t *testing.T) {

t.Errorf("%s: %v", name, err)

}

- if got, _ := reg.Get(old.Host); !reflect.DeepEqual(*got, *stored) {

+ if got := mustRoute(t, reg, old.Host); !reflect.DeepEqual(got, stored) {

t.Fatalf("a refused swap changed the row: %+v", got)

}

if js, err := reg.ImportJournals(); err != nil || len(js) != 0 {

@@ -198,14 +223,14 @@ func TestUpsertNeverChangesTheImportGate(t *testing.T) {

if err := reg.Upsert(row); err != nil {

t.Fatal(err)

}

- if got, _ := reg.Get(row.Host); got.ImportGate != "abc123" {

+ if got := mustRoute(t, reg, row.Host); got.ImportGate != "abc123" {

t.Fatalf("an Upsert cleared the gate: %q", got.ImportGate)

}

row.ImportGate = "def456"

if err := reg.Upsert(row); err != nil {

t.Fatal(err)

}

- if got, _ := reg.Get(row.Host); got.ImportGate != "abc123" {

+ if got := mustRoute(t, reg, row.Host); got.ImportGate != "abc123" {

t.Fatalf("an Upsert changed the gate: %q", got.ImportGate)

}

@@ -218,10 +243,10 @@ func TestImportJournalsListsEveryRow(t *testing.T) {

if err := reg.Upsert(old); err != nil {

t.Fatal(err)

}

- stored, _ := reg.Get(host)

+ stored := mustRoute(t, reg, host)

next := execRow()

next.Host = host

- j := ImportJournal{Host: host, SeedID: importSeedID, Original: *stored, At: int64(i + 1)}

+ j := ImportJournal{Host: host, SeedID: importSeedID, Original: stored, At: int64(i + 1)}

if err := reg.SwapForImport(j, next); err != nil {

t.Fatal(err)

}

@@ -243,27 +268,27 @@ func TestImportGateMovesBothFingerprints(t *testing.T) {

if err := reg.Upsert(legacyRow()); err != nil {

t.Fatal(err)

}

- fp1, _ := reg.Fingerprint()

- sfp1, _ := reg.SnapshotFingerprint()

+ fp1 := mustDigest(t, reg.Fingerprint)

+ sfp1 := mustDigest(t, reg.SnapshotFingerprint)

if _, err := reg.sql.Exec(`UPDATE routes SET import_gate = 'abc123' WHERE host = ?`, importHost); err != nil {

t.Fatal(err)

}

- fp2, _ := reg.Fingerprint()

- sfp2, _ := reg.SnapshotFingerprint()

+ fp2 := mustDigest(t, reg.Fingerprint)

+ sfp2 := mustDigest(t, reg.SnapshotFingerprint)

if fp1 == fp2 || sfp1 == sfp2 {

t.Fatal("setting the gate did not move both fingerprints")

}

if _, err := reg.sql.Exec(`UPDATE routes SET import_gate = ? WHERE host = ?`, ImportGateClosed, importHost); err != nil {

t.Fatal(err)

}

- if fp3, _ := reg.Fingerprint(); fp3 == fp2 {

+ if fp3 := mustDigest(t, reg.Fingerprint); fp3 == fp2 {

t.Fatal("closing the gate did not move the edge's fingerprint")

}

if _, err := reg.sql.Exec(`UPDATE routes SET import_gate = '' WHERE host = ?`, importHost); err != nil {

t.Fatal(err)

}

- fp4, _ := reg.Fingerprint()

- sfp4, _ := reg.SnapshotFingerprint()

+ fp4 := mustDigest(t, reg.Fingerprint)

+ sfp4 := mustDigest(t, reg.SnapshotFingerprint)

if fp4 != fp1 || sfp4 != sfp1 {

t.Fatal("clearing the gate did not restore the ungated digests")

}

@@ -333,8 +358,192 @@ func TestRestoreClaimRowsCarriesTheImportGate(t *testing.T) {

if *got != next {

t.Fatalf("restored %+v, want %+v", *got, next)

}

- // A replay of the same handover is a success, and reads the gate back.

+ // A replay of the same handover is a success, and the gate is still there.

if err := reg.RestoreClaimRows(ctx, "handover", "handover-old", ColdRows{Route: *old}, ColdRows{Route: next}); err != nil {

t.Fatal(err)

}

+ if got := mustRoute(t, reg, row.Host); got != next || got.ImportGate != "abc123" {

+ t.Fatalf("after replay %+v, want %+v", got, next)

+ }

+// Fix round 1: an empty journal digests byte-for-byte as before the journal

+// was hashed. The golden was produced by this fixture on f20d2c5e, whose

+// SnapshotFingerprint did not read import_journal.

+func TestSnapshotFingerprintUnchangedForAnEmptyJournal(t *testing.T) {

+ reg := openTestRegistry(t)

+ if err := reg.Upsert(legacyRow()); err != nil {

+ t.Fatal(err)

+ }

+ gated := execRow()

+ gated.Host = "gated.shop.example.com"

+ if err := reg.Upsert(gated); err != nil {

+ t.Fatal(err)

+ }

+ const want = "2:28920899b7d5e223b887dadf6924ee774b34d8d412f1c58bebdb214c35d82b07"

+ got, err := reg.SnapshotFingerprint()

+ if err != nil {

+ t.Fatal(err)

+ }

+ if got != want {

+ t.Fatalf("SnapshotFingerprint = %s, want %s", got, want)

+ }

+// Fix round 1 (review Important 1): the journal is in the snapshot digest,

+// so a deleted cleanup-pending row cannot linger in the last snapshot and be

+// restored cold.

+func TestSnapshotFingerprintSeesTheJournal(t *testing.T) {

+ reg := openTestRegistry(t)

+ old := legacyRow()

+ if err := reg.Upsert(old); err != nil {

+ t.Fatal(err)

+ }

+ stored := mustRoute(t, reg, old.Host)

+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: stored, At: 1}

+ if err := reg.SwapForImport(j, execRow()); err != nil {

+ t.Fatal(err)

+ }

+ if err := reg.CloseImportGate(old.Host, j.SeedID, 77); err != nil {

+ t.Fatal(err)

+ }

+ if err := reg.RestoreImportRoute(old.Host, j.SeedID); err != nil {

+ t.Fatal(err)

+ }

+ pending := mustDigest(t, reg.SnapshotFingerprint)

+ // A journal-only change moves it too.

+ if _, err := reg.sql.Exec(`UPDATE import_journal SET fence_gen = 78 WHERE host = ?`, old.Host); err != nil {

+ t.Fatal(err)

+ }

+ if mustDigest(t, reg.SnapshotFingerprint) == pending {

+ t.Fatal("a journal-only change did not move SnapshotFingerprint")

+ }

+ if _, err := reg.sql.Exec(`UPDATE import_journal SET fence_gen = 77 WHERE host = ?`, old.Host); err != nil {

+ t.Fatal(err)

+ }

+ if mustDigest(t, reg.SnapshotFingerprint) != pending {

+ t.Fatal("SnapshotFingerprint is not a function of the journal's contents")

+ }

+ routesBefore := mustDigest(t, reg.Fingerprint)

+ if err := reg.DeleteImportJournal(old.Host, j.SeedID); err != nil {

+ t.Fatal(err)

+ }

+ if mustDigest(t, reg.Fingerprint) != routesBefore {

+ t.Fatal("the journal moved the edge's fingerprint")

+ }

+ if mustDigest(t, reg.SnapshotFingerprint) == pending {

+ t.Fatal("deleting the journal row did not move SnapshotFingerprint")

+ }

+// Review Minor 1: the swap refuses an original it could not put back as an

+// ordinary legacy row.

+func TestSwapForImportRefusesAnOwnedHeldOrGatedOriginal(t *testing.T) {

+ for name, mutate := range map[string]func(*Route){

+ "claim-owned": func(r *Route) { r.ClaimOwnership = ClaimOwnership{"inc", 7, "installation-7", "route-7"} },

+ "held": func(r *Route) { r.HoldOp = "op-1" },

+ "gated": func(r *Route) { r.ImportGate = "abc123" },

+ } {

+ t.Run(name, func(t *testing.T) {

+ reg := openTestRegistry(t)

+ old := legacyRow()

+ mutate(&old)

+ if err := reg.Upsert(old); err != nil {

+ t.Fatal(err)

+ }

+ stored := mustRoute(t, reg, old.Host)

+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: stored}

+ if err := reg.SwapForImport(j, execRow()); !errors.Is(err, ErrImportJournal) {

+ t.Fatalf("swap: %v", err)

+ }

+ if got := mustRoute(t, reg, old.Host); got != stored {

+ t.Fatalf("a refused swap changed the row: %+v", got)

+ }

+ if jr := mustJournal(t, reg, old.Host); jr != nil {

+ t.Fatalf("a refused swap left a journal row: %+v", jr)

+ }

+ })

+ }

+// Review Minor 2: an original the journal's JSON would change (invalid

+// UTF-8 in a free-text column) is refused, so a restore is always exact.

+func TestSwapForImportRefusesAnOriginalJSONWouldChange(t *testing.T) {

+ reg := openTestRegistry(t)

+ old := legacyRow()

+ old.LastError = "bad \xff byte"

+ if err := reg.Upsert(old); err != nil {

+ t.Fatal(err)

+ }

+ stored := mustRoute(t, reg, old.Host)

+ if stored.LastError != old.LastError {

+ t.Skipf("the driver did not keep the raw bytes: %q", stored.LastError)

+ }

+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: stored}

+ if err := reg.SwapForImport(j, execRow()); !errors.Is(err, ErrImportJournal) {

+ t.Fatalf("swap: %v", err)

+ }

+ if got := mustRoute(t, reg, old.Host); got != stored {

+ t.Fatalf("a refused swap changed the row: %+v", got)

+ }

+// Review Minor 6: a failure after the transaction's first write rolls the

+// whole transaction back. A test-only trigger fails the journal write.

+func TestJournalTransactionsAreAtomic(t *testing.T) {

+ reg := openTestRegistry(t)

+ old := legacyRow()

+ if err := reg.Upsert(old); err != nil {

+ t.Fatal(err)

+ }

+ stored := mustRoute(t, reg, old.Host)

+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: stored}

+ // The swap: the route DELETE and the exec row INSERT land before the

+ // journal INSERT fails.

+ if _, err := reg.sql.Exec(`CREATE TRIGGER fail_swap BEFORE INSERT ON import_journal BEGIN SELECT RAISE(ABORT, 'injected'); END`); err != nil {

+ t.Fatal(err)

+ }

+ if err := reg.SwapForImport(j, execRow()); err == nil {

+ t.Fatal("swap with a failing journal write: no error")

+ }

+ if got := mustRoute(t, reg, old.Host); got != stored {

+ t.Fatalf("a failed swap left %+v, want the original", got)

+ }

+ if jr := mustJournal(t, reg, old.Host); jr != nil {

+ t.Fatalf("a failed swap left a journal row: %+v", jr)

+ }

+ if _, err := reg.sql.Exec(`DROP TRIGGER fail_swap`); err != nil {

+ t.Fatal(err)

+ }

+ if err := reg.SwapForImport(j, execRow()); err != nil {

+ t.Fatal(err)

+ }

+ if err := reg.CloseImportGate(old.Host, j.SeedID, 77); err != nil {

+ t.Fatal(err)

+ }

+ closed := mustRoute(t, reg, old.Host)

+ // The restore: the route swap lands before the journal UPDATE fails.

+ if _, err := reg.sql.Exec(`CREATE TRIGGER fail_restore BEFORE UPDATE ON import_journal WHEN NEW.state = 'cleanup-pending' BEGIN SELECT RAISE(ABORT, 'injected'); END`); err != nil {

+ t.Fatal(err)

+ }

+ if err := reg.RestoreImportRoute(old.Host, j.SeedID); err == nil {

+ t.Fatal("restore with a failing journal write: no error")

+ }

+ if got := mustRoute(t, reg, old.Host); got != closed {

+ t.Fatalf("a failed restore left %+v, want the closed exec row", got)

+ }

+ if jr := mustJournal(t, reg, old.Host); jr == nil || jr.State != JournalRollingBack {

+ t.Fatalf("a failed restore left the journal at %+v", jr)

+ }

+ if _, err := reg.sql.Exec(`DROP TRIGGER fail_restore`); err != nil {

+ t.Fatal(err)

+ }

+ if err := reg.RestoreImportRoute(old.Host, j.SeedID); err != nil {

+ t.Fatal(err)

+ }

+ if got := mustRoute(t, reg, old.Host); got != stored {

+ t.Fatalf("restored %+v, want %+v", got, stored)

+ }

}

diff --git a/internal/registry/registry.go b/internal/registry/registry.go

index 32345ef41..e1628cf64 100644

--- a/internal/registry/registry.go

+++ b/internal/registry/registry.go

@@ -2851,6 +2851,9 @@ FROM fleet_boxes ORDER BY fleet, label`)

if err := brows.Err(); err != nil {

return "", err

}

+ if err := hashImportJournal(db, h); err != nil {

+ return "", err

+ }

return fmt.Sprintf("%d:%x", n, h.Sum(nil)), nil

}