amadan
provision: a journal row is decided only by its own record (Plan 7 Task 4 fix round 1)
journalRecord used whichever record held the host in the pass. Another app's record winning a duplicate host, or holding the host while the import's own record is out of scope, could send a committed import to rollback. The pass's record now answers only when its key is the journal's; otherwise the record is read by the journal's key, and only not-found means gone. A protected host, a failed read, a row with no key and a body naming another host or key decide nothing. The key check precedes every rollback, and a host whose pass record is not the journal's stays handled, so no other record touches the row. Minors: a dry run starts no rollback; the handled skip comes first in the record loop; the fail-closed branches are tested.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3 files changed,
+224
−38
internal/provision/reconcile.go+3 −3internal/provision/seedjournal_test.go+167 −2internal/provision/seedreconcile.go+54 −33
CI passed — run details
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go| index e3e5730b1..fa89b5e6c 100644 |
| --- a/internal/provision/reconcile.go |
| +++ b/internal/provision/reconcile.go |
| @@ -1091,6 +1091,9 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| } |
| for _, rec := range recs { |
| + if journalHandled[rec.Host] { |
| + continue |
| + } |
| if trashedFor(rec.Account, rec.App) { |
| continue |
| } |
| @@ -1103,9 +1106,6 @@ 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 |
diff --git a/internal/provision/seedjournal_test.go b/internal/provision/seedjournal_test.go| index d82d4d35a..2df8fdb97 100644 |
| --- a/internal/provision/seedjournal_test.go |
| +++ b/internal/provision/seedjournal_test.go |
| @@ -2,13 +2,16 @@ package provision |
| import ( |
| "bytes" |
| + "context" |
| "database/sql" |
| + "errors" |
| "log" |
| "os" |
| "path/filepath" |
| "reflect" |
| "strings" |
| "testing" |
| + "time" |
| "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| @@ -219,8 +222,10 @@ func TestJournalDecidesEachRecordState(t *testing.T) { |
| } |
| } |
| -// The redo and the commit act only for the record the journal names: a |
| -// record holding this seed ID under another key moves nothing. |
| +// The redo and the commit act only for the record the journal names. A |
| +// record holding this seed ID under another key moves nothing forward; |
| +// the journal's own record is the one that decides, and here it is gone, |
| +// so the import goes to the rollback. |
| func TestJournalNeedsItsRecordKeyToMoveForward(t *testing.T) { |
| for _, state := range []string{seeds.StateReady, seeds.StateCommitted} { |
| t.Run(state, func(t *testing.T) { |
| @@ -237,9 +242,13 @@ func TestJournalNeedsItsRecordKeyToMoveForward(t *testing.T) { |
| t.Fatal(err) |
| } |
| row := mustGet(t, reg, rec.Host) |
| + logged := captureLog(t) |
| if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil { |
| t.Fatal(err) |
| } |
| + if !strings.Contains(logged.String(), rec.Host+": import "+rec.Seed.ID+" "+rollbackStubLine) { |
| + t.Fatalf("the journal's own record is gone, but no rollback started:\n%s", logged) |
| + } |
| if st := seedState(t, s, rec); st.State != state { |
| t.Fatalf("record = %s, want %s", st.State, state) |
| } |
| @@ -270,6 +279,162 @@ func TestJournalDryRunWritesNothing(t *testing.T) { |
| } |
| } |
| +// m4: a dry run starts no rollback either. |
| +func TestJournalDryRunStartsNoRollback(t *testing.T) { |
| + x := journaledImport(t, seeds.StateSwapped) |
| + x.cfg.DryRun = true |
| + if err := x.s.Delete(ctx, RecordKey(x.rec.Account, x.rec.App, x.rec.Host)); err != nil { |
| + t.Fatal(err) |
| + } |
| + logged := captureLog(t) |
| + if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if strings.Contains(logged.String(), rollbackStubLine) || !strings.Contains(logged.String(), "dry-run: would roll back "+x.rec.Host) { |
| + t.Fatalf("a dry run started a rollback, or did not say it would:\n%s", logged) |
| + } |
| + x.untouched(t) |
| +} |
| + |
| +// I1 (Task 4 review): another record for the journal's host never decides |
| +// the journal's import. Here an app named other, under the same account, |
| +// holds the host in the pass, while the import's own record is committed. |
| +// The import is committed, never rolled back, and the other record never |
| +// touches the row. |
| +func TestJournalIgnoresAnotherRecordForItsHost(t *testing.T) { |
| + for _, c := range []struct { |
| + name string |
| + state string |
| + edit func(t *testing.T, x *journaled) |
| + dup bool // the pass logs the duplicate host, keeping the other record |
| + }{ |
| + {name: "another record wins the duplicate pick", state: seeds.StateCommitted, dup: true}, |
| + {name: "another record wins the duplicate pick, swapped", state: seeds.StateSwapped, dup: true}, |
| + {name: "the import's record is out of scope", state: seeds.StateCommitted, edit: func(t *testing.T, x *journaled) { |
| + moved := x.rec |
| + moved.Region = "us" |
| + putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, moved)) |
| + }}, |
| + } { |
| + t.Run(c.name, func(t *testing.T) { |
| + x := journaledImport(t, c.state) |
| + if c.edit != nil { |
| + c.edit(t, &x) |
| + } |
| + other := shopRecord(x.rec.Account) |
| + other.App = "other" // listed before shop, so it wins a duplicate host |
| + // A run on the other record makes the loop's handling of it |
| + // visible (RunSeen): the loop must take up no other record for |
| + // a host the journal owns. |
| + other.ActiveRun = &RunState{ID: "run_abcdefghijklmnopqrstuvwxyz", State: RunQueued, Deadline: time.Now().Add(time.Minute)} |
| + var seen []string |
| + x.cfg.RunSeen = func(r Record) { seen = append(seen, RecordKey(r.Account, r.App, r.Host)) } |
| + otherKey := RecordKey(other.Account, other.App, other.Host) |
| + if _, err := x.s.Put(ctx, otherKey, mustJSON(t, other), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + logged := captureLog(t) |
| + if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + out := logged.String() |
| + if c.dup && !strings.Contains(out, "duplicate host "+x.rec.Host+": keeping "+otherKey) { |
| + t.Fatalf("the scene did not put the other record in the pass:\n%s", out) |
| + } |
| + if strings.Contains(out, rollbackStubLine) { |
| + t.Fatalf("another record sent the import to rollback:\n%s", out) |
| + } |
| + if len(seen) != 0 { |
| + t.Fatalf("the loop took up %v for a host the journal owns", seen) |
| + } |
| + if c.state == seeds.StateSwapped { |
| + x.untouched(t) |
| + return |
| + } |
| + got := mustGet(t, x.reg, x.rec.Host) |
| + if got.ImportGate != "" || got.App != "shop" || got.Account != x.rec.Account || got.Kind != "instance" || !got.Provisioned { |
| + t.Fatalf("row = %+v, want the import's row, committed", got) |
| + } |
| + if got.SeedSource != x.row.SeedSource || got.Socket != x.row.Socket { |
| + t.Fatalf("the row was converged by another record: %+v", got) |
| + } |
| + if j, _ := x.reg.ImportJournal(x.rec.Host); j != nil { |
| + t.Fatalf("journal = %+v, want it dropped by the commit", j) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// getFails is a store whose reads fail with something other than "not |
| +// found". |
| +type getFails struct { |
| + *release.DirStore |
| +} |
| + |
| +func (getFails) Get(context.Context, string) ([]byte, string, error) { |
| + return nil, "", errors.New("connection reset") |
| +} |
| + |
| +// m1, m2 (Task 4 review): every way journalRecord cannot tell decides |
| +// nothing (known false), and only "not found" under the journal's own key |
| +// means gone. |
| +func TestJournalRecordFailsClosed(t *testing.T) { |
| + s, sqid := shopSetup(t) |
| + rec := newImport(sqid) |
| + key := RecordKey(rec.Account, rec.App, rec.Host) |
| + j := registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, RecordKey: key, State: registry.JournalSwapped} |
| + other := shopRecord(sqid) |
| + other.App = "other" |
| + inPass := map[string]Record{rec.Host: rec} |
| + otherInPass := map[string]Record{rec.Host: other} |
| + none := map[string]bool{} |
| + |
| + for _, c := range []struct { |
| + name string |
| + store release.Store |
| + j registry.ImportJournal |
| + recs map[string]Record |
| + readFailed bool |
| + protect map[string]bool |
| + body []byte // written under the journal's key first |
| + known bool |
| + gone bool |
| + pass bool |
| + }{ |
| + {name: "the pass's own record", store: s, j: j, recs: inPass, protect: none, known: true, pass: true}, |
| + {name: "the pass's own record, another read failed", store: s, j: j, recs: inPass, readFailed: true, protect: none, known: true, pass: true}, |
| + {name: "protected", store: s, j: j, recs: inPass, protect: map[string]bool{rec.Host: true}}, |
| + {name: "no record key", store: s, j: registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, State: registry.JournalSwapped}, recs: map[string]Record{}, protect: none}, |
| + {name: "missing from a pass that failed a read", store: s, j: j, recs: map[string]Record{}, readFailed: true, protect: none}, |
| + {name: "another record in the pass, a read failed", store: s, j: j, recs: otherInPass, readFailed: true, protect: none}, |
| + {name: "the read fails", store: getFails{s}, j: j, recs: otherInPass, protect: none}, |
| + {name: "the body does not decode", store: s, j: j, recs: otherInPass, protect: none, body: []byte("{")}, |
| + {name: "the body names another host", store: s, j: j, recs: otherInPass, protect: none, body: func() []byte { |
| + r := rec |
| + r.Host = "b.shop.example.com" |
| + return mustJSON(t, r) |
| + }()}, |
| + {name: "gone", store: s, j: j, recs: otherInPass, protect: none, known: true, gone: true}, |
| + {name: "out of the pass, read by its key", store: s, j: j, recs: otherInPass, protect: none, body: mustJSON(t, rec), known: true}, |
| + } { |
| + t.Run(c.name, func(t *testing.T) { |
| + _ = s.Delete(ctx, key) |
| + if c.body != nil { |
| + if _, err := s.Put(ctx, key, c.body, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + got, pass, known := journalRecord(ctx, c.store, c.j, c.recs, c.readFailed, c.protect) |
| + if known != c.known || pass != c.pass || (known && (got == nil) != c.gone) { |
| + t.Fatalf("record %v, in pass %v, known %v; want known %v, gone %v, in pass %v", got, pass, known, c.known, c.gone, c.pass) |
| + } |
| + if got != nil && RecordKey(got.Account, got.App, got.Host) != key { |
| + t.Fatalf("answered with %s, not the journal's record", RecordKey(got.Account, got.App, got.Host)) |
| + } |
| + }) |
| + } |
| +} |
| + |
| // 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 |
diff --git a/internal/provision/seedreconcile.go b/internal/provision/seedreconcile.go| index ee054d462..1d33eef12 100644 |
| --- a/internal/provision/seedreconcile.go |
| +++ b/internal/provision/seedreconcile.go |
| @@ -190,7 +190,10 @@ func finishAbortedSeed(ctx context.Context, s release.Store, cfg ReconcileConfig |
| // 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: |
| +// A journal row at swapped is decided by the record that answers to it, |
| +// which is only ever the record under the journal's own key |
| +// (journalRecord): another record for the same host can only vouch for |
| +// the import, never revoke 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 |
| @@ -201,9 +204,10 @@ func finishAbortedSeed(ctx context.Context, s release.Store, cfg ReconcileConfig |
| // 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. |
| +// The loop converges a host only when the pass's own record for it is the |
| +// journal's: when another record won the host, or the journal's record is |
| +// out of scope, the host stays handled, so no other record touches the |
| +// row. 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{} |
| @@ -219,14 +223,21 @@ func journalPass(ctx context.Context, s release.Store, reg *registry.Registry, c |
| 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) |
| + rec, inPass, 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 |
| } |
| + // journalRecord answers only with the journal's own record; this |
| + // holds before any decision below, rollbacks first of all. |
| + if rec != nil && 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)) |
| + continue |
| + } |
| switch j.State { |
| case registry.JournalCleanupPending, registry.JournalRollingBack: |
| - rollbackImport(ctx, s, reg, cfg, j, rec) |
| + startRollback(ctx, s, reg, cfg, j, rec) |
| continue |
| case registry.JournalSwapped: |
| default: |
| @@ -239,12 +250,9 @@ func journalPass(ctx context.Context, s release.Store, reg *registry.Registry, c |
| } |
| switch { |
| case seed == nil || seed.ID != j.SeedID: |
| - rollbackImport(ctx, s, reg, cfg, j, rec) |
| + startRollback(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)) |
| + startRollback(ctx, s, reg, cfg, j, rec) |
| 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) |
| @@ -259,9 +267,9 @@ func journalPass(ctx context.Context, s release.Store, reg *registry.Registry, c |
| } 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 |
| + handled[j.Host] = !inPass |
| case seed.State == seeds.StateSwapped || seed.State == seeds.StatePlaced: |
| - handled[j.Host] = false |
| + handled[j.Host] = !inPass |
| 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) |
| @@ -270,41 +278,54 @@ func journalPass(ctx context.Context, s release.Store, reg *registry.Registry, c |
| 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. |
| +// startRollback hands an abandoned import to its rollback, which a dry |
| +// run never starts. |
| +func startRollback(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, j registry.ImportJournal, rec *Record) { |
| + if cfg.DryRun { |
| + log.Printf("provision: dry-run: would roll back %s's import %s", j.Host, j.SeedID) |
| + return |
| + } |
| + rollbackImport(ctx, s, reg, cfg, j, rec) |
| +} |
| + |
| +// journalRecord is the record under the journal's own key, nil when it is |
| +// gone, and known false when this pass cannot tell. inPass says it is the |
| +// record the pass holds for the host, so the loop may converge with it. |
| +// |
| +// recs holds the first in-scope record per host that read cleanly, which |
| +// need not be the journal's: another app's or account's record can win a |
| +// duplicate host, or hold the host while the journal's record is out of |
| +// this box's scope. So recs answers only when its record's key is the |
| +// journal's; otherwise the record is read by the journal's key, and only |
| +// os.ErrNotExist means gone. A protected host, a failed read, a journal |
| +// row with no key, and a body under the key that names another host or |
| +// key all decide nothing this pass: a committed import must never be |
| +// rolled back on a guess. |
| 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 |
| + readFailed bool, protect map[string]bool) (rec *Record, inPass, known bool) { |
| + if protect[j.Host] || j.RecordKey == "" { |
| + return nil, false, false |
| } |
| - if r, ok := recs[j.Host]; ok { |
| - return &r, true |
| + if r, ok := recs[j.Host]; ok && RecordKey(r.Account, r.App, r.Host) == j.RecordKey { |
| + return &r, true, true |
| } |
| if readFailed { |
| - return nil, false |
| - } |
| - if j.RecordKey == "" { |
| - return nil, true |
| + return nil, false, false |
| } |
| raw, _, err := s.Get(ctx, j.RecordKey) |
| if errors.Is(err, os.ErrNotExist) { |
| - return nil, true |
| + return nil, false, true |
| } |
| if err != nil { |
| log.Printf("provision: %s: read the import's record %s: %v", j.Host, j.RecordKey, err) |
| - return nil, false |
| + return nil, false, 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 nil, false, false |
| } |
| - return &r, true |
| + return &r, false, true |
| } |
| // rollbackImport is Task 5's. Until then, a stub that logs. |