amadan

carlos / platform Public

console: the self-serve job resolves a request its phase left by the fence; the expired void names the wait

An unconfirmed request whose issuing phase the row has left is never
replayed by any phase function: it is fenced and confirmed or voided,
as an abandoned write is. The expired void records the phase's own
reason. TestHooks gain Advance (time passing inside a tick) and the
confirm-read point.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude ca05a90e59d0e97c57a69830631ca706f0afc026 parent b0df7b1
3 files changed, +45 −16
  • internal/console/selfserve_apply.go +1 −0
  • internal/console/selfserve_job.go +27 −15
  • internal/console/testhooks.go +17 −1
diff --git a/internal/console/selfserve_apply.go b/internal/console/selfserve_apply.go
index 0f0608fb4..433306d1b 100644
--- a/internal/console/selfserve_apply.go
+++ b/internal/console/selfserve_apply.go
@@ -274,6 +274,7 @@ func (s *Server) selfServeRun(ctx context.Context, t *leaseToken, row *SelfServe
if pr.Expired {
return nil, errRequestExpired
}
+ TestHooks.point(pr.Kind + ":confirm-read")
rec, err := selfServeRecordOf(s.observeRecord(ctx, row.Account, row.App, row.Host), row)
if errors.Is(err, errRecordGone) {
// Absent, retired or recreated: an If-Match write bound to the
diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go
index 8bf9344ff..744a55211 100644
--- a/internal/console/selfserve_job.go
+++ b/internal/console/selfserve_job.go
@@ -230,6 +230,12 @@ func (s *Server) selfServeStep(ctx context.Context, t *leaseToken, row *SelfServ
if err != nil && !gone {
return err
}
+ case errors.Is(rerr, errReplayNeeded) && row.Request.Phase != row.Phase:
+ // A request whose issuing phase the row has left: no phase
+ // function replays it, so it is resolved as an abandoned
+ // write is, by the fence (its receipt confirms it; absent, it
+ // is voided) and the row decides again next tick.
+ return s.selfServeResolveFenced(ctx, t, row)
case errors.Is(rerr, errReplayNeeded):
// The write has not landed and the record is unchanged: the
// phase function replays it (below), unless the reserving
@@ -455,6 +461,23 @@ func (s *Server) selfServeFence(ctx context.Context, t *leaseToken, row *SelfSer
return obs.Rec, nil
}
+// selfServeResolveFenced resolves the row's unconfirmed request without
+// replaying it: the record fenced (so the write can never land later),
+// the fenced observation's receipt confirming it, and otherwise a void.
+func (s *Server) selfServeResolveFenced(ctx context.Context, t *leaseToken, row *SelfServeMove) error {
+ TestHooks.point(row.Request.Kind + ":fence")
+ frec, err := s.selfServeFence(ctx, t, row)
+ if err != nil {
+ log.Printf("console: self-serve move %s: fencing the %s request: %v (pending)", row.ID, row.Request.Kind, err)
+ return nil
+ }
+ if frec != nil && frec.Incarnation == row.Incarnation && selfServeHasReceipt(frec, row.Request.Nonce) {
+ return s.selfServeConfirmFenced(ctx, row, frec)
+ }
+ TestHooks.point(row.Request.Kind + ":void")
+ return s.store.VoidSelfServeRequest(row.ID, row.Request.Nonce)
+}
+
// selfServeConfirmFenced confirms the row's request from a fenced
// observation that carries its receipt, and announces it.
func (s *Server) selfServeConfirmFenced(ctx context.Context, row *SelfServeMove, rec *provision.Record) error {
@@ -501,7 +524,8 @@ func (s *Server) selfServeExpired(ctx context.Context, t *leaseToken, row *SelfS
return s.selfServeConfirmFenced(ctx, row, frec)
}
TestHooks.point(pr.Kind + ":void-expired")
- if _, err := s.store.VoidExpiredSelfServeRequest(row.ID, pr.Nonce, s.selfServeNow(), msgSelfServeCouldNotStart); err != nil {
+ reason := s.selfServeExpiryReason(ctx, row, frec, s.selfServeNow())
+ if _, err := s.store.VoidExpiredSelfServeRequest(row.ID, pr.Nonce, s.selfServeNow(), reason); err != nil {
return err
}
return nil
@@ -587,20 +611,8 @@ func (s *Server) selfServeAbortStep(ctx context.Context, t *leaseToken, row *Sel
out, conf := selfServeOutstanding(row)
switch {
case out && !conf:
- TestHooks.point(row.Request.Kind + ":fence")
- frec, err := s.selfServeFence(ctx, t, row)
- if err != nil {
- log.Printf("console: self-serve move %s: fencing the %s request before the abort: %v (pending)", row.ID, row.Request.Kind, err)
- return true, nil
- }
- if frec != nil && frec.Incarnation == row.Incarnation && selfServeHasReceipt(frec, row.Request.Nonce) {
- return true, s.selfServeConfirmFenced(ctx, row, frec)
- }
- TestHooks.point(row.Request.Kind + ":void")
- if err := s.store.VoidSelfServeRequest(row.ID, row.Request.Nonce); err != nil {
- return true, err
- }
- return true, nil // the next tick decides from the record
+ // The next tick decides from the record.
+ return true, s.selfServeResolveFenced(ctx, t, row)
case out && conf && row.Request.Kind != "abort" && row.Request.Kind != "finish":
TestHooks.point(row.Request.Kind + ":abandon")
if _, err := s.store.AbandonSelfServeRequest(row.ID, row.Request.Nonce, row.StepDeadline); err != nil {
diff --git a/internal/console/testhooks.go b/internal/console/testhooks.go
index 6a7d1ef5e..b68bcbc19 100644
--- a/internal/console/testhooks.go
+++ b/internal/console/testhooks.go
@@ -16,6 +16,7 @@ var TestHooks = &testHooks{}
type testHooks struct {
mu sync.Mutex
fault func(point string)
+ now time.Time // the ticking pass's clock (Tick, Advance)
}
// testCrash is what Crash panics with: the console process dying at a
@@ -50,7 +51,14 @@ func (h *testHooks) point(name string) {
// pass's ticker would (the lease, the events, one step per open row), and
// reports whether the tick crashed at a point, and which.
func (h *testHooks) Tick(ctx context.Context, s *Server, now time.Time) (crashed string) {
- s.selfServeClock = func() time.Time { return now }
+ h.mu.Lock()
+ h.now = now
+ h.mu.Unlock()
+ s.selfServeClock = func() time.Time {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ return h.now
+ }
defer func() {
s.selfServeClock = nil
if r := recover(); r != nil {
@@ -65,6 +73,14 @@ func (h *testHooks) Tick(ctx context.Context, s *Server, now time.Time) (crashed
return ""
}
+// Advance moves the clock of the tick in progress by d, from a Fault
+// function: time passing while the job waits on a read or a write.
+func (h *testHooks) Advance(d time.Duration) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ h.now = h.now.Add(d)
+}
+
// NameConsole names s in the moves lease as id: a console restarted on the
// same host and database takes back the lease its predecessor held, as
// the next acquire of a lease that still names this console does.