amadan
provision: the import journal drives its hosts before the records (Plan 7 Task 4)
Every reconcile pass reads the import journal first, whatever the seed hooks are. Journaled hosts are never removed, an unreadable journal is a failed read (nothing removed, no import acted on), and a journal row at swapped is decided by its record: abandoned imports (gone, no seed, another seed, aborted, refused, rolled back) go to the rollback stub, ready redoes the swap's CAS, committed clears the gate and the journal. A record missing from the pass's in-scope set is read by the journal's key before it counts as gone. seedAction takes the row and journalOK: an uncommitted import's ungated row is neither adopted nor converged (m-d). The converge sets the seed source before the seed command, and leaves the command if the source write fails (M10). A committed seed restores its own replica and never boots fresh (M11).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
7 files changed,
+822
−24
internal/activator/backend_exec.go+5 −5internal/activator/seedplace.go+18 −2internal/activator/seedplace_linux_test.go+47 −1internal/provision/reconcile.go+37 −13internal/provision/seedjournal_test.go+556 −0internal/provision/seedreconcile.go+153 −3internal/provision/seedswap.go+6 −0
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go| index c58d647ab..2c6a48317 100644 |
| --- a/internal/activator/backend_exec.go |
| +++ b/internal/activator/backend_exec.go |
| @@ -632,16 +632,16 @@ var ( |
| func (l *execLease) Release(ctx context.Context) error { return l.held.Release(ctx) } |
| // Restore puts the row's database in place for a wake. A row whose seed is |
| -// placeable (ready, swapped or placed) goes through the seed fallback |
| -// (seedplace.go), which never reports fresh; every other row restores |
| -// plainly. A seed column that does not parse fails the wake: reading it as |
| -// "no seed" would boot a seeded instance empty. |
| +// placeable (ready, swapped or placed) or committed goes through the seed |
| +// fallback (seedplace.go), which never reports fresh; every other row |
| +// restores plainly. A seed column that does not parse fails the wake: |
| +// reading it as "no seed" would boot a seeded instance empty. |
| func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, error) { |
| src, err := seeds.ParseRowSeed(rt.SeedSource) |
| if err != nil { |
| return false, err |
| } |
| - if src != nil && src.Placeable() { |
| + if seedRestores(src) { |
| return b.seededRestore(ctx, rt, src) |
| } |
| return b.restorePlain(ctx, rt) |
diff --git a/internal/activator/seedplace.go b/internal/activator/seedplace.go| index e998ef61b..818213ed2 100644 |
| --- a/internal/activator/seedplace.go |
| +++ b/internal/activator/seedplace.go |
| @@ -74,6 +74,21 @@ func seedPlaceFailed(host string, err error) error { |
| return &SeedPlacementError{Sentence: fmt.Sprintf("%s could not start from its seed: it could not be put in place. The next wake tries again.", host), Err: err} |
| } |
| +// ownsReplica reports a seed whose host already has a replica of its own: |
| +// placed is set only after the first upload, and committed only from |
| +// placed (spec §5.4). Such a row restores its own lineage and never boots |
| +// fresh: after commit the app retires its old process, so an empty boot |
| +// would read as the import's data lost (preflight M11). |
| +func ownsReplica(src *seeds.RowSeed) bool { |
| + return src.State == seeds.StatePlaced || src.State == seeds.StateCommitted |
| +} |
| + |
| +// seedRestores reports a row Restore sends through seededRestore: a seed |
| +// still to place, or one whose own replica exists. |
| +func seedRestores(src *seeds.RowSeed) bool { |
| + return src != nil && (src.Placeable() || ownsReplica(src)) |
| +} |
| + |
| // seededRestore is the plain-path seed fallback (spec §3.4). It never |
| // reports fresh: a seeded row never runs the manifest's seed command, and |
| // every failure fails the wake instead of falling through to an empty |
| @@ -111,8 +126,9 @@ func (b *ExecBackend) seededRestore(ctx context.Context, rt registry.Route, src |
| // the host's own replica exists. So a placed row's wake is a plain |
| // restore on any box: no layout or claim refusal, no listing on every |
| // wake, and the staging fallback still promotes a moved host's replica |
| - // (the snapshot was deleted at placement; it is never read again). |
| - if src.State == seeds.StatePlaced { |
| + // (the snapshot was deleted at placement; it is never read again). A |
| + // committed import was placed first, so it is restored the same way. |
| + if ownsReplica(src) { |
| return restoreOwn() |
| } |
| if b.ClaimArmed { |
diff --git a/internal/activator/seedplace_linux_test.go b/internal/activator/seedplace_linux_test.go| index 41282ac8e..afc2fbb0f 100644 |
| --- a/internal/activator/seedplace_linux_test.go |
| +++ b/internal/activator/seedplace_linux_test.go |
| @@ -400,6 +400,52 @@ func TestSeededRestoreOfAPlacedSeedWithNothingToRestoreFails(t *testing.T) { |
| } |
| } |
| +// committedImport is a row whose import is committed: placed first, so |
| +// the host has a replica of its own (preflight M11). |
| +func committedImport(t *testing.T, b *ExecBackend, rt registry.Route) (registry.Route, *fakeSeedObjects) { |
| + t.Helper() |
| + rt, objs, _ := seededRoute(t, b, rt, seeds.StateCommitted) |
| + col, err := seeds.ParseRowSeed(rt.SeedSource) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + col.Kind = seeds.KindImport |
| + rt.SeedSource = col.Encode() |
| + return rt, objs |
| +} |
| + |
| +// M11: a committed import restores its own replica, as placed does, and a |
| +// wake that finds none fails. It never boots empty: after commit the app |
| +// retires its old process, so an empty boot would lose the import. |
| +func TestSeededRestoreOfACommittedImportWithNothingToRestoreFails(t *testing.T) { |
| + lists := countOwnReplicaLists(t, false) |
| + b, rt := nestedBackend(t, &fakeControl{}, &fakeUnit{}) |
| + rt, objs := committedImport(t, b, rt) |
| + oldRestore := restoreReplica |
| + restored := 0 |
| + restoreReplica = func(context.Context, string, string, string, string) (bool, error) { restored++; return false, nil } |
| + defer func() { restoreReplica = oldRestore }() |
| + if fresh, err := b.Restore(context.Background(), rt); err == nil || fresh || restored != 1 || objs.opens != 0 || *lists != 0 { |
| + t.Fatalf("fresh %v err %v restored %d opens %d lists %d; want a failed wake, never fresh", fresh, err, restored, objs.opens, *lists) |
| + } |
| +} |
| + |
| +func TestSeededRestoreOfACommittedImportRestoresItsOwnReplica(t *testing.T) { |
| + lists := countOwnReplicaLists(t, false) |
| + b, rt := nestedBackend(t, &fakeControl{}, &fakeUnit{}) |
| + rt, objs := committedImport(t, b, rt) |
| + b.ClaimArmed = true // as placed: no claim refusal for a restore of its own |
| + oldRestore := restoreReplica |
| + restoreReplica = func(_ context.Context, dbPath, _, _, _ string) (bool, error) { |
| + return true, writeMinimalSQLite(dbPath) |
| + } |
| + defer func() { restoreReplica = oldRestore }() |
| + os.MkdirAll(filepath.Dir(rt.DBPath), 0o700) |
| + if fresh, err := b.Restore(context.Background(), rt); err != nil || fresh || objs.opens != 0 || *lists != 0 { |
| + t.Fatalf("fresh %v err %v opens %d lists %d; want its own replica, no fetch", fresh, err, objs.opens, *lists) |
| + } |
| +} |
| + |
| // A crash after the marker and before the rename leaves a marker naming |
| // this seed and no database: the snapshot is fetched again. |
| func TestSeededRestoreRefetchesWhenOnlyTheMarkerSurvived(t *testing.T) { |
| @@ -417,7 +463,7 @@ func TestSeededRestoreRefetchesWhenOnlyTheMarkerSurvived(t *testing.T) { |
| } |
| func TestSeededRestoreLeavesAnUnplaceableSeedToThePlainPath(t *testing.T) { |
| - for _, state := range []string{seeds.StateAborted, seeds.StateAwaitingUpload, seeds.StateCommitted} { |
| + for _, state := range []string{seeds.StateAborted, seeds.StateAwaitingUpload, seeds.StateRolledBack} { |
| t.Run(state, func(t *testing.T) { |
| lists := countOwnReplicaLists(t, false) |
| b, rt := nestedBackend(t, &fakeControl{}, &fakeUnit{}) |
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go| index cd0a4ccf8..e3e5730b1 100644 |
| --- a/internal/provision/reconcile.go |
| +++ b/internal/provision/reconcile.go |
| @@ -1080,6 +1080,15 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| byHost[rt.Host] = rt |
| } |
| mintLayout := mintLayoutReader(ctx, s, cfg) |
| + // The import journal is read before the records are acted on, every |
| + // pass (seeded-instances spec §5.7, seedreconcile.go). A host it handled |
| + // is skipped by the loop below, and every journaled host by the removal |
| + // phase: the journal owns the row until its row is gone. An unreadable |
| + // journal could hide any host, so the removal phase does not run. |
| + journalHandled, journaled, journalOK := journalPass(ctx, s, reg, cfg, recs, readFailed, protect) |
| + if !journalOK { |
| + readFailed = true |
| + } |
| for _, rec := range recs { |
| if trashedFor(rec.Account, rec.App) { |
| @@ -1094,19 +1103,22 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| // clobber, and a later pass converges it once the manifest reads. |
| seed, seedIsKnown := seedFor(rec.Account, rec.App) |
| existing, ok := byHost[rec.Host] |
| + if journalHandled[rec.Host] { |
| + continue |
| + } |
| // A seed is dispatched before anything else (seeded-instances spec |
| // §3.2, codex 13): an aborted one is finished, an unready one is |
| // skipped, and a seeded row never carries the manifest's seed |
| // command. |
| if rec.Seed != nil { |
| - switch seedAction(rec, ok, cfg, mintLayout) { |
| + var row *registry.Route |
| + if ok { |
| + row = &existing |
| + } |
| + switch seedAction(rec, row, cfg, mintLayout, journalOK) { |
| case seedSkip: |
| continue |
| case seedSwap: |
| - var row *registry.Route |
| - if ok { |
| - row = &existing |
| - } |
| swapImport(ctx, s, reg, cfg, rec, row, mintLayout) |
| continue |
| case seedFinishAbort: |
| @@ -1270,22 +1282,31 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| log.Printf("provision: converged %s gate %s -> %s", rec.Host, gateWord(existing.Gate), gateWord(rec.Gate)) |
| } |
| } |
| + // The seed source goes first, and the seed command follows only |
| + // once it has landed (seeded-instances preflight M10). A row |
| + // whose record has a seed gets its seed column before its seed |
| + // command is cleared, so between the two writes it is placeable |
| + // (seededRestore, never fresh). In the other order, or with the |
| + // source write failed and the command cleared anyway, the row |
| + // would hold neither, and a wake then could boot it empty. |
| + sourceConverged := true |
| + if col := SeedColumn(rec); existing.SeedSource != col { |
| + if cfg.DryRun { |
| + log.Printf("provision: dry-run: would converge %s seed source", rec.Host) |
| + } else if err := reg.SetSeedSource(rec.Host, col); err != nil { |
| + log.Printf("provision: converge seed source on %s: %v", rec.Host, err) |
| + sourceConverged = false |
| + } |
| + } |
| // seedIsKnown guards against stamping a transient manifest-read |
| // failure's "" fallback over a working seed — see seedFor. |
| - if seedIsKnown && existing.Seed != seed { |
| + if sourceConverged && seedIsKnown && existing.Seed != seed { |
| if cfg.DryRun { |
| log.Printf("provision: dry-run: would converge %s seed", rec.Host) |
| } else if err := reg.SetSeed(rec.Host, seed); err != nil { |
| log.Printf("provision: converge seed on %s: %v", rec.Host, err) |
| } |
| } |
| - if col := SeedColumn(rec); existing.SeedSource != col { |
| - if cfg.DryRun { |
| - log.Printf("provision: dry-run: would converge %s seed source", rec.Host) |
| - } else if err := reg.SetSeedSource(rec.Host, col); err != nil { |
| - log.Printf("provision: converge seed source on %s: %v", rec.Host, err) |
| - } |
| - } |
| // Unit is compared only when the RECORD states one: `carlos |
| // route -unit` (issue #110) is the supported way to stamp the |
| // unit owning a route's process out of band, and a record that |
| @@ -1453,6 +1474,9 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| } |
| why = "app in the trash" |
| } |
| + if journaled[rt.Host] { |
| + continue // the import journal owns the row (seedreconcile.go) |
| + } |
| if protect[rt.Host] { |
| log.Printf("provision: not removing %s: its record exists but could not be read cleanly this pass", rt.Host) |
| continue |
diff --git a/internal/provision/seedjournal_test.go b/internal/provision/seedjournal_test.go| new file mode 100644 |
| index 000000000..d82d4d35a |
| --- /dev/null |
| +++ b/internal/provision/seedjournal_test.go |
| @@ -0,0 +1,556 @@ |
| +package provision |
| + |
| +import ( |
| + "bytes" |
| + "database/sql" |
| + "log" |
| + "os" |
| + "path/filepath" |
| + "reflect" |
| + "strings" |
| + "testing" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/seeds" |
| +) |
| + |
| +// captureLog collects the package's log lines for the rest of the test. |
| +func captureLog(t *testing.T) *bytes.Buffer { |
| + t.Helper() |
| + var b bytes.Buffer |
| + log.SetOutput(&b) |
| + t.Cleanup(func() { log.SetOutput(os.Stderr) }) |
| + return &b |
| +} |
| + |
| +// The rollback is Task 5's; until then journalPass hands it to a stub |
| +// that logs this. |
| +const rollbackStubLine = "needs a rollback (not built yet)" |
| + |
| +// journaled is an import this box has swapped: the gated exec row in |
| +// place of the addr row, and the journal at swapped under the record's |
| +// key. The record is whatever state was asked for. |
| +type journaled struct { |
| + s *release.DirStore |
| + reg *registry.Registry |
| + rec Record |
| + row registry.Route |
| + j registry.ImportJournal |
| + cfg ReconcileConfig |
| +} |
| + |
| +func journaledImport(t *testing.T, state string) journaled { |
| + t.Helper() |
| + s, sqid := shopSetup(t) |
| + rec := newImport(sqid) |
| + rec.Seed.State = state |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + reg := regWith(t, addrRow(rec, sqid)) |
| + original := mustGet(t, reg, rec.Host) |
| + if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, Original: *original, |
| + RecordKey: RecordKey(rec.Account, rec.App, rec.Host), At: 1}, importRow(rec, "/data")); err != nil { |
| + t.Fatal(err) |
| + } |
| + j, err := reg.ImportJournal(rec.Host) |
| + if err != nil || j == nil { |
| + t.Fatalf("journal = %+v, %v", j, err) |
| + } |
| + return journaled{s: s, reg: reg, rec: rec, row: *mustGet(t, reg, rec.Host), j: *j, cfg: importCfg(t, nil)} |
| +} |
| + |
| +// untouched asserts the journal owned the host this pass: the swapped row |
| +// and the journal row exactly as they were. |
| +func (x journaled) untouched(t *testing.T) { |
| + t.Helper() |
| + if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, x.row) { |
| + t.Fatalf("row changed:\n got %+v\nwant %+v", *got, x.row) |
| + } |
| + if j, _ := x.reg.ImportJournal(x.rec.Host); j == nil || !reflect.DeepEqual(*j, x.j) { |
| + t.Fatalf("journal changed: %+v", j) |
| + } |
| +} |
| + |
| +func TestJournalResumesTheRecordCAS(t *testing.T) { |
| + s, sqid := shopSetup(t) |
| + rec := importRecord(t, s, sqid) |
| + reg := regWith(t, addrRow(rec, sqid)) |
| + stored, _ := reg.Get(rec.Host) |
| + // The box crashed after the transaction and before the record CAS. |
| + next := routeFor(rec, "/data", "", true) |
| + next.ImportGate = rec.Seed.VerifyHash |
| + if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, Original: *stored, |
| + RecordKey: RecordKey(rec.Account, rec.App, rec.Host)}, next); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil { |
| + t.Fatal(err) |
| + } |
| + r, _ := GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| + if r.Seed.State != seeds.StateSwapped { |
| + t.Fatalf("state = %s, want the CAS redone", r.Seed.State) |
| + } |
| + if got := mustGet(t, reg, rec.Host); got.Kind != "instance" { |
| + t.Fatalf("the swapped row was refused or changed: %+v", got) |
| + } |
| +} |
| + |
| +func TestCommitClearsTheGateAndTheJournal(t *testing.T) { |
| + s, sqid := shopSetup(t) |
| + rec := importRecord(t, s, sqid) |
| + reg := regWith(t, addrRow(rec, sqid)) |
| + cfg := importCfg(t, nil) |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, step := range []func(string) func(*Record) error{SeedPlace, SeedCommit} { |
| + if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, step(rec.Seed.ID)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + got := mustGet(t, reg, rec.Host) |
| + if got.ImportGate != "" { |
| + t.Fatalf("gate after commit = %q", got.ImportGate) |
| + } |
| + if j, _ := reg.ImportJournal(rec.Host); j != nil { |
| + t.Fatalf("journal after commit = %+v", j) |
| + } |
| + // The same pass converged the row: its seed column says committed. |
| + if col, _ := seeds.ParseRowSeed(got.SeedSource); col == nil || col.State != seeds.StateCommitted { |
| + t.Fatalf("seed column after commit = %+v", col) |
| + } |
| +} |
| + |
| +// Each record state a journal row at swapped can meet (the Task 4 table, |
| +// spec §5.4 "Crashes" and §5.6). A rollback case must leave the host to |
| +// the rollback alone: no ordinary handling, no removal, and none of the |
| +// no-journal paths. |
| +func TestJournalDecidesEachRecordState(t *testing.T) { |
| + type want int |
| + const ( |
| + rollback want = iota |
| + nothing |
| + ordinary |
| + ) |
| + for _, c := range []struct { |
| + name string |
| + state string |
| + edit func(t *testing.T, x *journaled) |
| + want want |
| + }{ |
| + {name: "record gone", state: seeds.StateReady, want: rollback, edit: func(t *testing.T, x *journaled) { |
| + if err := x.s.Delete(ctx, RecordKey(x.rec.Account, x.rec.App, x.rec.Host)); err != nil { |
| + t.Fatal(err) |
| + } |
| + }}, |
| + {name: "record with no seed", state: seeds.StateReady, want: rollback, edit: func(t *testing.T, x *journaled) { |
| + plain := x.rec |
| + plain.Seed = nil |
| + putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, plain)) |
| + }}, |
| + {name: "another seed", state: seeds.StateReady, want: rollback, edit: func(t *testing.T, x *journaled) { |
| + other := x.rec |
| + seed := *x.rec.Seed |
| + seed.ID, _ = seeds.NewID() |
| + other.Seed = &seed |
| + putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, other)) |
| + }}, |
| + {name: "aborted", state: seeds.StateAborted, want: rollback}, |
| + // m-c: a refusal that lost the race to this box's swap. The |
| + // journal wins. |
| + {name: "refused", state: seeds.StateRefused, want: rollback}, |
| + {name: "rolled back", state: seeds.StateRolledBack, want: rollback}, |
| + // m12: states that come before any swap. Unreachable, so the |
| + // host is held and logged. |
| + {name: "reserving", state: seeds.StateReserving, want: nothing}, |
| + {name: "awaiting upload", state: seeds.StateAwaitingUpload, want: nothing}, |
| + {name: "uploaded", state: seeds.StateUploaded, want: nothing}, |
| + {name: "swapped", state: seeds.StateSwapped, want: ordinary}, |
| + {name: "placed", state: seeds.StatePlaced, want: ordinary}, |
| + } { |
| + t.Run(c.name, func(t *testing.T) { |
| + x := journaledImport(t, c.state) |
| + if c.edit != nil { |
| + c.edit(t, &x) |
| + } |
| + before, _ := GetRecord(ctx, x.s, x.rec.Account, x.rec.App, x.rec.Host) |
| + // Converging the row as an ordinary owned row would rewrite its |
| + // seed column: make it stale so ordinary handling shows. |
| + if err := x.reg.SetSeedSource(x.rec.Host, "{}"); err != nil { |
| + t.Fatal(err) |
| + } |
| + x.row.SeedSource = "{}" |
| + logged := captureLog(t) |
| + if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + rolledBack := strings.Contains(logged.String(), x.rec.Host+": import "+x.rec.Seed.ID+" "+rollbackStubLine) |
| + switch c.want { |
| + case rollback, nothing: |
| + x.untouched(t) |
| + if rolledBack != (c.want == rollback) { |
| + t.Fatalf("rollback started = %v, want %v\n%s", rolledBack, c.want == rollback, logged) |
| + } |
| + if c.want == nothing && !strings.Contains(logged.String(), "is journaled but its record says "+c.state) { |
| + t.Fatalf("the host was not held with a log line:\n%s", logged) |
| + } |
| + after, _ := GetRecord(ctx, x.s, x.rec.Account, x.rec.App, x.rec.Host) |
| + if !reflect.DeepEqual(after, before) { |
| + t.Fatalf("record changed: %+v -> %+v", before, after) |
| + } |
| + case ordinary: |
| + if rolledBack { |
| + t.Fatalf("rolled back a live import:\n%s", logged) |
| + } |
| + got := mustGet(t, x.reg, x.rec.Host) |
| + if got.SeedSource != SeedColumn(x.rec) || got.ImportGate != x.rec.Seed.VerifyHash { |
| + t.Fatalf("row = %+v, want it converged and still gated", got) |
| + } |
| + if j, _ := x.reg.ImportJournal(x.rec.Host); j == nil || !reflect.DeepEqual(*j, x.j) { |
| + t.Fatalf("journal changed: %+v", j) |
| + } |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// The redo and the commit act only for the record the journal names: a |
| +// record holding this seed ID under another key moves nothing. |
| +func TestJournalNeedsItsRecordKeyToMoveForward(t *testing.T) { |
| + for _, state := range []string{seeds.StateReady, seeds.StateCommitted} { |
| + t.Run(state, func(t *testing.T) { |
| + s, sqid := shopSetup(t) |
| + rec := newImport(sqid) |
| + rec.Seed.State = state |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + reg := regWith(t, addrRow(rec, sqid)) |
| + original := mustGet(t, reg, rec.Host) |
| + if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, Original: *original, |
| + RecordKey: RecordKey(sqid, "other", rec.Host), At: 1}, importRow(rec, "/data")); err != nil { |
| + t.Fatal(err) |
| + } |
| + row := mustGet(t, reg, rec.Host) |
| + if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if st := seedState(t, s, rec); st.State != state { |
| + t.Fatalf("record = %s, want %s", st.State, state) |
| + } |
| + if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, *row) { |
| + t.Fatalf("row changed: %+v", got) |
| + } |
| + if j, _ := reg.ImportJournal(rec.Host); j == nil { |
| + t.Fatal("the journal row went") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// m3: a dry run redoes no CAS and clears no gate. |
| +func TestJournalDryRunWritesNothing(t *testing.T) { |
| + for _, state := range []string{seeds.StateReady, seeds.StateCommitted} { |
| + t.Run(state, func(t *testing.T) { |
| + x := journaledImport(t, state) |
| + x.cfg.DryRun = true |
| + if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if st := seedState(t, x.s, x.rec); st.State != state { |
| + t.Fatalf("record = %s, want %s", st.State, state) |
| + } |
| + x.untouched(t) |
| + }) |
| + } |
| +} |
| + |
| +// The removal phase never removes a journaled host's row, whatever the |
| +// journal's state: the journal owns it until the row is gone (M1, M13). |
| +// The gated row is Provisioned and its record is gone, which without the |
| +// journal is exactly what the removal phase deletes. |
| +func TestRemovalSkipsJournaledHosts(t *testing.T) { |
| + for _, state := range []string{registry.JournalSwapped, registry.JournalRollingBack} { |
| + t.Run(state, func(t *testing.T) { |
| + x := journaledImport(t, seeds.StateSwapped) |
| + if state == registry.JournalRollingBack { |
| + if err := x.reg.CloseImportGate(x.rec.Host, x.rec.Seed.ID, 7); err != nil { |
| + t.Fatal(err) |
| + } |
| + x.row = *mustGet(t, x.reg, x.rec.Host) |
| + j, _ := x.reg.ImportJournal(x.rec.Host) |
| + x.j = *j |
| + } |
| + if err := x.s.Delete(ctx, RecordKey(x.rec.Account, x.rec.App, x.rec.Host)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + x.untouched(t) |
| + }) |
| + } |
| +} |
| + |
| +// M1: the journal is read even with seed hooks unarmed. Its hosts are |
| +// protected from removal and from ordinary handling, and nothing is done |
| +// for them, a commit included. |
| +func TestJournalIsReadWithSeedHooksUnarmed(t *testing.T) { |
| + t.Run("record gone", func(t *testing.T) { |
| + x := journaledImport(t, seeds.StateSwapped) |
| + x.cfg.Seeds = nil |
| + if err := x.s.Delete(ctx, RecordKey(x.rec.Account, x.rec.App, x.rec.Host)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + x.untouched(t) |
| + }) |
| + t.Run("record replaced by one with no seed", func(t *testing.T) { |
| + x := journaledImport(t, seeds.StateSwapped) |
| + x.cfg.Seeds = nil |
| + plain := x.rec |
| + plain.Seed = nil |
| + putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, plain)) |
| + if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + x.untouched(t) |
| + }) |
| + t.Run("committed", func(t *testing.T) { |
| + x := journaledImport(t, seeds.StateCommitted) |
| + x.cfg.Seeds = nil |
| + logged := captureLog(t) |
| + if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + x.untouched(t) |
| + if strings.Contains(logged.String(), rollbackStubLine) { |
| + t.Fatalf("a rollback started with hooks unarmed:\n%s", logged) |
| + } |
| + }) |
| +} |
| + |
| +// corruptJournal makes ImportJournals fail: the row's original no longer |
| +// parses. |
| +func corruptJournal(t *testing.T, path, host string) { |
| + t.Helper() |
| + db, err := sql.Open("sqlite", path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer db.Close() |
| + if _, err := db.Exec(`UPDATE import_journal SET original = '{' WHERE host = ?`, host); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// regAt is regWith at a path the test can open beside the registry. |
| +func regAt(t *testing.T, routes ...registry.Route) (*registry.Registry, string) { |
| + t.Helper() |
| + path := filepath.Join(t.TempDir(), "registry.db") |
| + reg, err := registry.Open(path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, rt := range routes { |
| + if err := reg.Upsert(rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + return reg, path |
| +} |
| + |
| +// M1: a journal that cannot be read is a failed read. Nothing is removed |
| +// that pass, and no import is acted on: no swap, and no no-journal finish |
| +// of an aborted import (it might have a journal row). |
| +func TestJournalReadFailureRemovesNothingAndActsOnNoImport(t *testing.T) { |
| + s, sqid := shopSetup(t) |
| + // A swapped import, whose journal row will not parse. |
| + swapped := importRecord(t, s, sqid) |
| + // A ready import on another host, with its addr row. |
| + ready := newImport(sqid) |
| + ready.Host, ready.Slug = "b.shop.example.com", "b" |
| + if err := CreateRecord(ctx, s, ready); err != nil { |
| + t.Fatal(err) |
| + } |
| + // An aborted import on a third host. |
| + aborted := newImport(sqid) |
| + aborted.Host, aborted.Slug = "c.shop.example.com", "c" |
| + aborted.Seed.State = seeds.StateAborted |
| + if err := CreateRecord(ctx, s, aborted); err != nil { |
| + t.Fatal(err) |
| + } |
| + // And a provisioned row whose record is gone. |
| + orphan := routeFor(shopRecord(sqid), "/data", "", true) |
| + orphan.Host, orphan.Slug = "d.shop.example.com", "d" |
| + reg, path := regAt(t, addrRow(swapped, sqid), addrRow(ready, sqid), orphan) |
| + original := mustGet(t, reg, swapped.Host) |
| + if err := reg.SwapForImport(registry.ImportJournal{Host: swapped.Host, SeedID: swapped.Seed.ID, Original: *original, |
| + RecordKey: RecordKey(swapped.Account, swapped.App, swapped.Host), At: 1}, importRow(swapped, "/data")); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := s.Delete(ctx, RecordKey(swapped.Account, swapped.App, swapped.Host)); err != nil { |
| + t.Fatal(err) |
| + } |
| + gated := mustGet(t, reg, swapped.Host) |
| + readyRow := mustGet(t, reg, ready.Host) |
| + corruptJournal(t, path, swapped.Host) |
| + |
| + if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := mustGet(t, reg, swapped.Host); !reflect.DeepEqual(*got, *gated) { |
| + t.Fatalf("the journaled row changed: %+v", got) |
| + } |
| + mustGet(t, reg, orphan.Host) // not removed: the pass could not read the journal |
| + if got := mustGet(t, reg, ready.Host); !reflect.DeepEqual(*got, *readyRow) { |
| + t.Fatalf("the ready import was swapped: %+v", got) |
| + } |
| + if st := seedState(t, s, ready); st.State != seeds.StateReady || st.Reason != "" { |
| + t.Fatalf("ready import = %s %q", st.State, st.Reason) |
| + } |
| + if st := seedState(t, s, aborted); st.State != seeds.StateAborted { |
| + t.Fatalf("aborted import = %s, want it left aborted", st.State) |
| + } |
| +} |
| + |
| +// A journal row's record that this box's scope excludes is still there: |
| +// it is read by its key, never taken as gone, and a committed import is |
| +// committed, never rolled back. |
| +func TestJournalReadsAnOutOfScopeRecordByItsKey(t *testing.T) { |
| + x := journaledImport(t, seeds.StateCommitted) |
| + moved := x.rec |
| + moved.Region = "us" |
| + putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, moved)) |
| + logged := captureLog(t) |
| + if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if strings.Contains(logged.String(), rollbackStubLine) { |
| + t.Fatalf("rolled back a committed import:\n%s", logged) |
| + } |
| + if got := mustGet(t, x.reg, x.rec.Host); got.ImportGate != "" { |
| + t.Fatalf("gate = %q, want it cleared", got.ImportGate) |
| + } |
| + if j, _ := x.reg.ImportJournal(x.rec.Host); j != nil { |
| + t.Fatalf("journal = %+v", j) |
| + } |
| +} |
| + |
| +// m-d: before commit, only the swap's gated row is the import's row. An |
| +// ungated row for a swapped or placed import, hand-added or with its gate |
| +// cleared by hand, is neither adopted nor converged. |
| +func TestReconcileLeavesAnUngatedRowOfAnUncommittedImport(t *testing.T) { |
| + for _, state := range []string{seeds.StateSwapped, seeds.StatePlaced} { |
| + for _, provisioned := range []bool{false, true} { |
| + name := state + "/hand-added" |
| + if provisioned { |
| + name = state + "/gate cleared" |
| + } |
| + t.Run(name, func(t *testing.T) { |
| + s, sqid := shopSetup(t) |
| + rec := newImport(sqid) |
| + rec.Seed.State = state |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + row := importRow(rec, "/data") |
| + row.ImportGate = "" |
| + row.Provisioned = provisioned |
| + row.SeedSource = "{}" // stale: a converge would rewrite it |
| + reg := regWith(t, row) |
| + stored := mustGet(t, reg, rec.Host) |
| + if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, *stored) { |
| + t.Fatalf("row changed:\n got %+v\nwant %+v", *got, *stored) |
| + } |
| + }) |
| + } |
| + } |
| +} |
| + |
| +// The journal check inside swapImport stays as a backstop (Task 3's |
| +// departure 1): called with this seed's journal row at swapped, it only |
| +// redoes the record CAS, even with no journal pass ahead of it. |
| +func TestSwapImportBackstopRedoesOnlyTheCAS(t *testing.T) { |
| + x := journaledImport(t, seeds.StateReady) |
| + swapImport(ctx, x.s, x.reg, x.cfg, x.rec, &x.row, func() (bool, error) { return true, nil }) |
| + if st := seedState(t, x.s, x.rec); st.State != seeds.StateSwapped || st.Reason != "" { |
| + t.Fatalf("record = %s %q, want swapped", st.State, st.Reason) |
| + } |
| + x.untouched(t) |
| +} |
| + |
| +// convergeScene is a --from seed whose record is placed and whose row on |
| +// this box still holds the manifest's seed command and no seed column: |
| +// the converge must set the column and clear the command. |
| +func convergeScene(t *testing.T) (*release.DirStore, *registry.Registry, string, Record) { |
| + t.Helper() |
| + s := dirStore(t) |
| + sqid := writeAccountManifest(t, s, acctID, "Shop") |
| + if err := PutManifest(ctx, s, sqid, "shop", Manifest{Schema: 1, App: "shop", Domain: "shop.example.com", Seed: []string{"seed"}}); err != nil { |
| + t.Fatal(err) |
| + } |
| + rec := shopRecord(sqid) |
| + id, _ := seeds.NewID() |
| + rec.Seed = &SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StatePlaced, SHA256: strings.Repeat("a", 64), Bytes: 10} |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + row := routeFor(rec, "/data", `["seed"]`, true) |
| + row.SeedSource = "" |
| + reg, path := regAt(t, row) |
| + return s, reg, path, rec |
| +} |
| + |
| +func addTrigger(t *testing.T, path, ddl string) { |
| + t.Helper() |
| + db, err := sql.Open("sqlite", path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer db.Close() |
| + if _, err := db.Exec(ddl); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// M10: the converge sets the seed column before it clears the seed |
| +// command, so the row is never left holding neither (which could boot |
| +// empty). The trigger refuses that state; in the other order the seed |
| +// write is refused and the command stays. |
| +func TestConvergeSetsTheSeedSourceBeforeClearingTheSeed(t *testing.T) { |
| + s, reg, path, rec := convergeScene(t) |
| + addTrigger(t, path, `CREATE TRIGGER neither BEFORE UPDATE ON routes |
| +WHEN NEW.seed = '' AND NEW.seed_source = '' |
| +BEGIN SELECT RAISE(ABORT, 'a row with neither a seed command nor a seed source'); END`) |
| + if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := mustGet(t, reg, rec.Host); got.Seed != "" || got.SeedSource != SeedColumn(rec) { |
| + t.Fatalf("row = seed %q source %q, want no seed command and this seed's column", got.Seed, got.SeedSource) |
| + } |
| +} |
| + |
| +// M10: a seed column that cannot be written leaves the seed command in |
| +// place, so the row keeps the state it had. |
| +func TestConvergeKeepsTheSeedWhenTheSeedSourceFails(t *testing.T) { |
| + s, reg, path, rec := convergeScene(t) |
| + addTrigger(t, path, `CREATE TRIGGER nosource BEFORE UPDATE OF seed_source ON routes |
| +BEGIN SELECT RAISE(ABORT, 'seed_source is read-only here'); END`) |
| + if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := mustGet(t, reg, rec.Host); got.Seed != `["seed"]` || got.SeedSource != "" { |
| + t.Fatalf("row = seed %q source %q, want it as it was", got.Seed, got.SeedSource) |
| + } |
| +} |
diff --git a/internal/provision/seedreconcile.go b/internal/provision/seedreconcile.go| index abdb6688d..ee054d462 100644 |
| --- a/internal/provision/seedreconcile.go |
| +++ b/internal/provision/seedreconcile.go |
| @@ -39,9 +39,13 @@ const ( |
| ) |
| // seedAction is §3.2's dispatch, run before any other per-record handling. |
| -// hasRow is whether this box already holds a row for the host. nested is |
| -// asked only for a mint, through the pass's lazy layout reader. |
| -func seedAction(rec Record, hasRow bool, cfg ReconcileConfig, nested func() (bool, error)) seedVerdict { |
| +// row is this box's row for the host, nil when it holds none. nested is |
| +// asked only for a mint, through the pass's lazy layout reader. journalOK |
| +// is whether this pass read the import journal: when it could not, no |
| +// import is acted on, since the journal decides every import it holds a |
| +// row for (spec §5.7) and the no-journal paths must never run while a |
| +// journal row might exist. |
| +func seedAction(rec Record, row *registry.Route, cfg ReconcileConfig, nested func() (bool, error), journalOK bool) seedVerdict { |
| src := rec.Seed |
| if src == nil { |
| return seedOrdinary |
| @@ -49,6 +53,11 @@ func seedAction(rec Record, hasRow bool, cfg ReconcileConfig, nested func() (boo |
| if cfg.Seeds == nil { |
| return seedSkip |
| } |
| + if src.Import() && !journalOK { |
| + log.Printf("provision: %s: import %s left alone this pass: the import journal could not be read", rec.Host, src.ID) |
| + return seedSkip |
| + } |
| + hasRow := row != nil |
| switch src.State { |
| case seeds.StateAborted: |
| return seedFinishAbort |
| @@ -87,6 +96,13 @@ func seedAction(rec Record, hasRow bool, cfg ReconcileConfig, nested func() (boo |
| log.Printf("provision: %s: not minting import %s: it is %s but not committed, and only its swap makes its row", rec.Host, src.ID, src.State) |
| return seedSkip |
| } |
| + // The swap's row is gated until commit. An ungated row here was |
| + // put there by hand, and adopting or converging it would serve |
| + // the unverified import to ordinary traffic. |
| + if src.Import() && row.ImportGate == "" { |
| + log.Printf("provision: %s: not converging import %s: it is %s but not committed, and the row on this box has no import gate", rec.Host, src.ID, src.State) |
| + return seedSkip |
| + } |
| return seedOrdinary |
| case seeds.StateCommitted: |
| return seedOrdinary |
| @@ -161,3 +177,137 @@ func finishAbortedSeed(ctx context.Context, s release.Store, cfg ReconcileConfig |
| } |
| log.Printf("provision: %s: aborted seed %s rolled back (no box had swapped it)", rec.Host, rec.Seed.ID) |
| } |
| + |
| +// journalPass is spec §5.7: the box's import journal is read before the |
| +// records are acted on, every pass, and each row drives its host (§5.4 |
| +// "Crashes", §5.6). handled hosts are skipped by the record loop; |
| +// journaled hosts (every journal row) by the removal phase, since the |
| +// journal owns the row until the row is gone. |
| +// |
| +// The journal is read whatever cfg.Seeds is (preflight M1): with seed |
| +// hooks unarmed every journaled host is still protected and handled, |
| +// with no action. ok is false when the journal cannot be read. Reconcile |
| +// then treats the pass as a failed read for removal, and seedAction acts |
| +// on no import. |
| +// |
| +// A journal row at swapped is decided by the record that answers to it: |
| +// |
| +// - gone, no seed, another seed, or this seed aborted, refused or rolled |
| +// back: the import is abandoned, so it is rolled back. A refused |
| +// record with a journal row is a refusal that lost the race to this |
| +// box's swap: the journal wins; |
| +// - this seed at ready: the swap's record CAS is redone; |
| +// - this seed at committed: the gate is cleared and the journal row |
| +// dropped (CommitImport), and the loop converges the row; |
| +// - this seed at swapped or placed: the loop handles it as usual. |
| +// |
| +// The redo and the commit also need the journal's record key, so a |
| +// record that names this seed under another key moves nothing. Every |
| +// other case leaves the host alone this pass, logged. |
| +func journalPass(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, |
| + recs map[string]Record, readFailed bool, protect map[string]bool) (handled, journaled map[string]bool, ok bool) { |
| + handled, journaled = map[string]bool{}, map[string]bool{} |
| + rows, err := reg.ImportJournals() |
| + if err != nil { |
| + log.Printf("provision: the import journal could not be read: %v (no import is acted on and no route is removed this pass)", err) |
| + return handled, journaled, false |
| + } |
| + for _, j := range rows { |
| + journaled[j.Host] = true |
| + handled[j.Host] = true // cleared below for the two cases the loop converges |
| + if cfg.Seeds == nil { |
| + log.Printf("provision: %s: import %s is journaled, but this box cannot act on imports; its row is left alone", j.Host, j.SeedID) |
| + continue |
| + } |
| + rec, known := journalRecord(ctx, s, j, recs, readFailed, protect) |
| + if !known { |
| + log.Printf("provision: %s: import %s left alone this pass: its record could not be read", j.Host, j.SeedID) |
| + continue |
| + } |
| + switch j.State { |
| + case registry.JournalCleanupPending, registry.JournalRollingBack: |
| + rollbackImport(ctx, s, reg, cfg, j, rec) |
| + continue |
| + case registry.JournalSwapped: |
| + default: |
| + log.Printf("provision: %s: import %s is journaled in an unknown state %q; leaving it alone", j.Host, j.SeedID, j.State) |
| + continue |
| + } |
| + var seed *SeedSource |
| + if rec != nil { |
| + seed = rec.Seed |
| + } |
| + switch { |
| + case seed == nil || seed.ID != j.SeedID: |
| + rollbackImport(ctx, s, reg, cfg, j, rec) |
| + case seed.State == seeds.StateAborted || seed.State == seeds.StateRefused || seed.State == seeds.StateRolledBack: |
| + rollbackImport(ctx, s, reg, cfg, j, rec) |
| + case RecordKey(rec.Account, rec.App, rec.Host) != j.RecordKey: |
| + log.Printf("provision: %s: import %s left alone: the journal holds it under %s, not %s", j.Host, j.SeedID, |
| + j.RecordKey, RecordKey(rec.Account, rec.App, rec.Host)) |
| + case seed.State == seeds.StateReady: |
| + if cfg.DryRun { |
| + log.Printf("provision: dry-run: would redo %s's swap record CAS for import %s", j.Host, j.SeedID) |
| + } else { |
| + markSwapped(ctx, s, *rec) |
| + } |
| + case seed.State == seeds.StateCommitted: |
| + if cfg.DryRun { |
| + log.Printf("provision: dry-run: would clear %s's import gate", j.Host) |
| + } else if err := reg.CommitImport(j.Host, j.SeedID); err != nil { |
| + log.Printf("provision: %s: clear the import gate: %v", j.Host, err) |
| + } else { |
| + log.Printf("provision: %s: import %s committed: its gate is cleared and its journal row dropped", j.Host, j.SeedID) |
| + } |
| + handled[j.Host] = false |
| + case seed.State == seeds.StateSwapped || seed.State == seeds.StatePlaced: |
| + handled[j.Host] = false |
| + default: |
| + // reserving, awaiting-upload and uploaded come before any swap. |
| + log.Printf("provision: %s: import %s is journaled but its record says %s; leaving it alone", j.Host, j.SeedID, seed.State) |
| + } |
| + } |
| + return handled, journaled, true |
| +} |
| + |
| +// journalRecord is the record a journal row answers to, nil when it is |
| +// gone, and known false when this pass cannot tell. recs holds only |
| +// in-scope records that read cleanly, so a host missing from it is gone |
| +// only if the pass read every record, the host is not protected, and the |
| +// journal's record key is absent from the bucket: a record out of this |
| +// box's scope, or skipped as a duplicate host, is still there, and a |
| +// committed import must never be rolled back for either. |
| +func journalRecord(ctx context.Context, s release.Store, j registry.ImportJournal, recs map[string]Record, |
| + readFailed bool, protect map[string]bool) (rec *Record, known bool) { |
| + if protect[j.Host] { |
| + return nil, false |
| + } |
| + if r, ok := recs[j.Host]; ok { |
| + return &r, true |
| + } |
| + if readFailed { |
| + return nil, false |
| + } |
| + if j.RecordKey == "" { |
| + return nil, true |
| + } |
| + raw, _, err := s.Get(ctx, j.RecordKey) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return nil, true |
| + } |
| + if err != nil { |
| + log.Printf("provision: %s: read the import's record %s: %v", j.Host, j.RecordKey, err) |
| + return nil, false |
| + } |
| + r, err := DecodeRecord(raw) |
| + if err != nil || r.Host != j.Host || RecordKey(r.Account, r.App, r.Host) != j.RecordKey { |
| + log.Printf("provision: %s: the import's record %s is not one this pass can use (%v)", j.Host, j.RecordKey, err) |
| + return nil, false |
| + } |
| + return &r, true |
| +} |
| + |
| +// rollbackImport is Task 5's. Until then, a stub that logs. |
| +func rollbackImport(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, j registry.ImportJournal, rec *Record) { |
| + log.Printf("provision: %s: import %s needs a rollback (not built yet)", j.Host, j.SeedID) |
| +} |
diff --git a/internal/provision/seedswap.go b/internal/provision/seedswap.go| index c197b5866..e827a5538 100644 |
| --- a/internal/provision/seedswap.go |
| +++ b/internal/provision/seedswap.go |
| @@ -163,6 +163,12 @@ func swapImport(ctx context.Context, s release.Store, reg *registry.Registry, cf |
| // already swapped, so the legacy-row predicates never judge the row it |
| // made. This seed's own row at swapped is a crash between the swap's |
| // transaction and the record's CAS: only the CAS is redone. |
| + // |
| + // Reconcile's journalPass decides every journaled host before the swap |
| + // is reached, so in a pass this check finds nothing. It is kept as a |
| + // backstop: without it, a journal row this function did not expect |
| + // (another caller, or a row written after the pass read the journal) |
| + // would be judged as a legacy row and refused, terminally. |
| j, err := reg.ImportJournal(rec.Host) |
| if err != nil { |
| log.Printf("provision: %s: import deferred: the import journal could not be read: %v", rec.Host, err) |