console: tests for the three-way reclaim, lost races and lost replies
TestReclaimOnlyOnEvidence expects the 502 for a read failure, a fence failure, a corrupt record and a failed row or cleanup read, adds a fence- time read failure after a good pre-read and HostHasIssuedRequest failing alone; TestReserveLosingTwoRacesIsBusy; TestStagedDoorsConfirmALostReply.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 file changed,
+148
−28
internal/console/offlinereserve_test.go+148 −28
diff --git a/internal/console/offlinereserve_test.go b/internal/console/offlinereserve_test.go| index 35f06105e..e94926638 100644 |
| --- a/internal/console/offlinereserve_test.go |
| +++ b/internal/console/offlinereserve_test.go |
| @@ -169,11 +169,13 @@ func (w *offlineWorld) queueOn(host, id string) { |
| // fakeRows is the self-serve rows' evidence (Task 4's tables): a move id |
| // maps to whether its row is terminal and released; an absent id has no |
| -// row. issued names hosts with a request issued. err fails every read. |
| +// row. issued names hosts with a request issued. err fails every read, |
| +// issuedErr only HostHasIssuedRequest. |
| type fakeRows struct { |
| - terminal map[string]bool |
| - issued map[string]bool |
| - err error |
| + terminal map[string]bool |
| + issued map[string]bool |
| + err error |
| + issuedErr error |
| } |
| func (f fakeRows) MoveTerminalOrAbsent(id string) (bool, error) { |
| @@ -188,6 +190,9 @@ func (f fakeRows) HostHasIssuedRequest(acct, app, host string) (bool, error) { |
| if f.err != nil { |
| return false, f.err |
| } |
| + if f.issuedErr != nil { |
| + return false, f.issuedErr |
| + } |
| return f.issued[host], nil |
| } |
| @@ -473,14 +478,19 @@ func TestReclaimOnlyOnEvidence(t *testing.T) { |
| w.t.Fatalf("after the reclaim: %+v", tb) |
| } |
| } |
| - kept := func(w *offlineWorld) { |
| + // keptAs: the reservation stays, and the reserver is told busy (the |
| + // evidence says the holder is live) or unreadable (it could not be |
| + // read: today's 502 for an unreadable lock holder). |
| + keptAs := func(w *offlineWorld, want *offlineRefusal) { |
| w.t.Helper() |
| before := w.activeRaw() |
| - w.wantReserve(selOf("ie-1", "uk-1"), mres(mvB, betaHost, 0), refBusyAcme) |
| + w.wantReserve(selOf("ie-1", "uk-1"), mres(mvB, betaHost, 0), want) |
| if w.activeRaw() != before { |
| w.t.Fatal("a kept reservation was rewritten") |
| } |
| } |
| + kept := func(w *offlineWorld) { w.t.Helper(); keptAs(w, refBusyAcme) } |
| + unreadable := func(w *offlineWorld) { w.t.Helper(); keptAs(w, refUnread) } |
| fencedRecord := func(w *offlineWorld) bool { |
| w.t.Helper() |
| rec, err := provision.GetRecord(ctx, w.fx.rel, "bab", "shop", offlineHost) |
| @@ -540,30 +550,44 @@ func TestReclaimOnlyOnEvidence(t *testing.T) { |
| t.Fatalf("the delayed queue %+v (seq %d) was not kept and fenced", rec.Move, rec.RecordSeq) |
| } |
| }) |
| - t.Run("a store read failure is kept", func(t *testing.T) { |
| + t.Run("a store read failure is unreadable", func(t *testing.T) { |
| w := held(t, time.Hour, moves.OfflineKindMove) |
| lt := <Store{Store: w.fx.rel} |
| w.fx.srv.cfg.Releases = lt |
| lt.setHook(failNth("get", acmeKey, 1, true)) |
| - kept(w) |
| + unreadable(w) |
| }) |
| - t.Run("a fence error is kept", func(t *testing.T) { |
| + t.Run("a store read failure at the fence after a good read is unreadable", func(t *testing.T) { |
| + w := held(t, time.Hour, moves.OfflineKindMove) |
| + lt := <Store{Store: w.fx.rel} |
| + w.fx.srv.cfg.Releases = lt |
| + before := w.recordRaw(offlineHost) |
| + lt.setHook(failNth("get", acmeKey, 2, true)) |
| + unreadable(w) |
| + if w.recordRaw(offlineHost) != before { |
| + t.Fatal("the record changed") |
| + } |
| + if ops := lt.opsOf(acmeKey); countOps(ops, "get") < 2 || countOps(ops, "put") != 0 { |
| + t.Fatalf("record calls %v, want the good read, the failed fence read and no write", ops) |
| + } |
| + }) |
| + t.Run("a fence error is unreadable", func(t *testing.T) { |
| w := held(t, time.Hour, moves.OfflineKindMove) |
| lt := <Store{Store: w.fx.rel} |
| w.fx.srv.cfg.Releases = lt |
| before := w.recordRaw(offlineHost) |
| lt.setHook(failNth("put", acmeKey, 1, true)) |
| - kept(w) |
| + unreadable(w) |
| if w.recordRaw(offlineHost) != before { |
| t.Fatal("the record changed") |
| } |
| }) |
| - t.Run("a corrupt record is kept", func(t *testing.T) { |
| + t.Run("a corrupt record is unreadable", func(t *testing.T) { |
| w := held(t, time.Hour, moves.OfflineKindMove) |
| offlinePut(t, w.fx.rel, acmeKey, []byte("{not a record")) |
| - kept(w) |
| + unreadable(w) |
| }) |
| - t.Run("a pending lifetime operation on the host is kept", func(t *testing.T) { |
| + t.Run("a pending lifetime operation on the host is busy", func(t *testing.T) { |
| w := held(t, time.Hour, moves.OfflineKindMove) |
| if err := w.fx.st.BeginRetirement(Retirement{OpID: "other", Kind: retireOrphan, Account: "bab", App: "shop", Host: offlineHost, Incarnation: 1}); err != nil { |
| t.Fatal(err) |
| @@ -575,24 +599,26 @@ func TestReclaimOnlyOnEvidence(t *testing.T) { |
| } |
| }) |
| t.Run("self-serve rows: (c) and (d)", func(t *testing.T) { |
| + locked := errors.New("database is locked") |
| for _, c := range []struct { |
| name string |
| rows fakeRows |
| - ok bool |
| + want *offlineRefusal // nil: reclaimed |
| }{ |
| - {"no row", fakeRows{}, true}, |
| - {"terminal and released", fakeRows{terminal: map[string]bool{mvA: true}}, true}, |
| - {"not terminal", fakeRows{terminal: map[string]bool{mvA: false}}, false}, |
| - {"a request issued for the host", fakeRows{issued: map[string]bool{offlineHost: true}}, false}, |
| - {"a SQL failure", fakeRows{err: errors.New("database is locked")}, false}, |
| + {"no row", fakeRows{}, nil}, |
| + {"terminal and released", fakeRows{terminal: map[string]bool{mvA: true}}, nil}, |
| + {"not terminal", fakeRows{terminal: map[string]bool{mvA: false}}, refBusyAcme}, |
| + {"a request issued for the host", fakeRows{issued: map[string]bool{offlineHost: true}}, refBusyAcme}, |
| + {"a SQL failure", fakeRows{err: locked}, refUnread}, |
| + {"a SQL failure of the issued-request read alone", fakeRows{issuedErr: locked}, refUnread}, |
| } { |
| t.Run(c.name, func(t *testing.T) { |
| w := held(t, time.Hour, moves.OfflineKindMove) |
| w.fx.srv.selfServeRows = c.rows |
| - if c.ok { |
| + if c.want == nil { |
| reclaimed(w) |
| } else { |
| - kept(w) |
| + keptAs(w, c.want) |
| } |
| }) |
| } |
| @@ -601,20 +627,20 @@ func TestReclaimOnlyOnEvidence(t *testing.T) { |
| for _, c := range []struct { |
| name string |
| ev cleanupReportEvidence |
| - ok bool |
| + want *offlineRefusal // nil: reclaimed |
| }{ |
| - {"no report evidence yet", nil, false}, |
| - {"settled", fakeCleanup{settled: true}, true}, |
| - {"incomplete status", fakeCleanup{}, false}, |
| - {"a read failure", fakeCleanup{settled: true, err: errors.New("status read")}, false}, |
| + {"no report evidence yet", nil, refBusyAcme}, |
| + {"settled", fakeCleanup{settled: true}, nil}, |
| + {"incomplete status", fakeCleanup{}, refBusyAcme}, |
| + {"a read failure", fakeCleanup{settled: true, err: errors.New("status read")}, refUnread}, |
| } { |
| t.Run(c.name, func(t *testing.T) { |
| w := held(t, time.Hour, moves.OfflineKindCleanup) |
| w.fx.srv.cleanupReports = c.ev |
| - if c.ok { |
| + if c.want == nil { |
| reclaimed(w) |
| } else { |
| - kept(w) |
| + keptAs(w, c.want) |
| } |
| }) |
| } |
| @@ -995,6 +1021,100 @@ func TestOperatorDoorsUseReservations(t *testing.T) { |
| }) |
| } |
| +// Two lost races on active.json answer busy, as the brief says: naming the |
| +// holder a last read finds, or the reserver's own host when none is there. |
| +func TestReserveLosingTwoRacesIsBusy(t *testing.T) { |
| + t.Run("a holder appears", func(t *testing.T) { |
| + w := newReserveWorld(t) |
| + lt := <Store{Store: w.fx.rel} |
| + w.fx.srv.cfg.Releases = lt |
| + var n int |
| + lt.setHook(func(op, key string) error { |
| + if op != "put" || key != moves.OfflineActiveKey { |
| + return nil |
| + } |
| + if n++; n == 2 { |
| + w.putTable(map[string]moves.OfflineReservation{moves.OfflineSentinel: mres(mvB, betaHost, 0)}) |
| + } |
| + return release.ErrCASConflict |
| + }) |
| + w.wantReserve(sentinelSel(), mres(mvA, offlineHost, 0), refBusyBeta) |
| + if n != 2 { |
| + t.Fatalf("%d writes attempted, want two", n) |
| + } |
| + }) |
| + t.Run("no holder is ever seen", func(t *testing.T) { |
| + w := newReserveWorld(t) |
| + lt := <Store{Store: w.fx.rel} |
| + w.fx.srv.cfg.Releases = lt |
| + lt.setHook(func(op, key string) error { |
| + if op == "put" && key == moves.OfflineActiveKey { |
| + return release.ErrCASConflict |
| + } |
| + return nil |
| + }) |
| + w.wantReserve(sentinelSel(), mres(mvA, offlineHost, 0), refBusyAcme) |
| + if w.activeRaw() != "" { |
| + t.Fatal("a lost race wrote active.json") |
| + } |
| + }) |
| +} |
| + |
| +// landThenConflict lands the first write of key and then answers it as a |
| +// lost race: the commit whose reply was lost, surfacing as the 412 of the |
| +// store's own retry. |
| +type landThenConflict struct { |
| + release.Store |
| + key string |
| + landed bool |
| +} |
| + |
| +func (f *landThenConflict) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if key == f.key && !f.landed { |
| + f.landed = true |
| + if _, err := f.Store.Put(ctx, key, body, ifMatch); err != nil { |
| + return "", err |
| + } |
| + return "", release.ErrCASConflict |
| + } |
| + return f.Store.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| +// A staged move's start and restage confirm a write whose reply was lost: |
| +// the record carrying the transition's stamp is a success, not the 409. |
| +func TestStagedDoorsConfirmALostReply(t *testing.T) { |
| + for _, c := range []struct{ name, path, state string }{ |
| + {"restage", "/stage", provision.OfflineStaging}, |
| + {"staged start", "", provision.OfflineStopping}, |
| + } { |
| + t.Run(c.name, func(t *testing.T) { |
| + w := newReserveWorld(t) |
| + v := w.ok("/stage", `{"region":"uk"}`, provision.OfflineStaging) |
| + staged := w.record() |
| + w.fx.srv.cfg.Releases = &landThenConflict{Store: w.fx.rel, key: provision.RecordKey("bab", "shop", offlineHost)} |
| + got := w.ok(c.path, `{"region":"uk"}`, c.state) |
| + rec := w.record() |
| + if got.ID != v.ID || rec.Move.ID != v.ID || rec.Move.State != c.state || rec.RecordSeq != staged.RecordSeq+1 { |
| + t.Fatalf("after the lost reply: view %+v, record %+v seq %d (was %d)", got, rec.Move, rec.RecordSeq, staged.RecordSeq) |
| + } |
| + }) |
| + } |
| + // A lost race that did not land is still the 409. |
| + w := newReserveWorld(t) |
| + w.ok("/stage", `{"region":"uk"}`, provision.OfflineStaging) |
| + lt := <Store{Store: w.fx.rel} |
| + w.fx.srv.cfg.Releases = lt |
| + lt.setHook(func(op, key string) error { |
| + if op == "put" && key == provision.RecordKey("bab", "shop", offlineHost) { |
| + return release.ErrCASConflict |
| + } |
| + return nil |
| + }) |
| + if code, out := w.req(http.MethodPost, "", `{"region":"uk"}`); code != http.StatusConflict || errorText(out) != "the instance changed while the move was being started. try again" { |
| + t.Fatalf("staged start over a lost race = %d %s", code, out) |
| + } |
| +} |
| + |
| // sequenceRecord rewrites a record's exact bytes with the record sequence |
| // on, as a fence or any sequenced writer does: record_seq advances. |
| func sequenceRecord(t *testing.T, rel release.Store, key string) { |