amadan

carlos / platform Public

offline return: the door's index cleanup runs after the bump and the answer (fix round 3, item 2)

offlineBegin bumps the serial and answers 202 first; the return's index
cleanup then runs detached under its 10 s bound, counted in the server's
new detached WaitGroup, which Server.Shutdown waits for. A stalled index
store delays neither the bump nor the answer; the entry it leaves is the
sweep's. serve() has no graceful-shutdown path today, so Shutdown is
called by tests only.

Witness: TestReturnIndexCleanupIsBounded (the index store stalled; the 202
and the bump well under the bound, Shutdown returns once the cleanup gives
up, the entry left). RED: "the door answered after 2.04s: the stalled
cleanup (bound 2s) sat before the answer". TestReturnIndexIsTheConsoles
waits for the detached cleanup before checking the deletion.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 8086be0a3f1c484e738005351d4b898135048481 parent 59a7154
3 files changed, +58 −18
  • internal/console/apiinstancemove_offline.go +12 −7
  • internal/console/offlinereturn_test.go +25 −9
  • internal/console/server.go +21 −2
diff --git a/internal/console/apiinstancemove_offline.go b/internal/console/apiinstancemove_offline.go
index 070b12896..22c473b66 100644
--- a/internal/console/apiinstancemove_offline.go
+++ b/internal/console/apiinstancemove_offline.go
@@ -1004,17 +1004,22 @@ func (s *Server) offlineBegin(w http.ResponseWriter, r *http.Request, d offlineD
s.offlineWriteFailed(w, d, err)
return
}
+ s.bumpSerial(ctx)
+ writeJSON(w, http.StatusAccepted, s.offlineViewOf(ctx, got))
if sel.ReturnOf != "" {
// The return is staged or started: its index entry has done its
- // work (final review F6). Best effort, bounded: a stalled index
- // store never holds the serial bump or the answer past the bound,
- // and what it leaves the sweep removes (fix round 2).
+ // work (final review F6). After the bump and the answer, detached
+ // and tracked (Server.Shutdown), under its own bound: a stalled
+ // index store never delays either, and what the cleanup leaves the
+ // sweep removes (fix rounds 2 and 3).
cctx, ccancel := context.WithTimeout(context.WithoutCancel(ctx), offlineReturnCleanupTimeout)
- s.offlineReturnIndexDone(cctx, sel.To, host, sel.ReturnOf)
- ccancel()
+ s.detached.Add(1)
+ go func() {
+ defer s.detached.Done()
+ defer ccancel()
+ s.offlineReturnIndexDone(cctx, sel.To, host, sel.ReturnOf)
+ }()
}
- s.bumpSerial(ctx)
- writeJSON(w, http.StatusAccepted, s.offlineViewOf(ctx, got))
}
// offlineChangedMsg is the start and stage doors' refusal of a bound write
diff --git a/internal/console/offlinereturn_test.go b/internal/console/offlinereturn_test.go
index 254f3f866..34380b455 100644
--- a/internal/console/offlinereturn_test.go
+++ b/internal/console/offlinereturn_test.go
@@ -513,6 +513,10 @@ func TestReturnIndexIsTheConsoles(t *testing.T) {
if code, out := w.req(http.MethodPost, door, `{"region":"uk","return_of":"`+returnFwd+`"}`); code != http.StatusAccepted {
t.Fatalf("%s = %d %s", door, code, out)
}
+ // The cleanup runs detached after the answer: wait for it.
+ if err := w.fx.srv.Shutdown(ctx); err != nil {
+ t.Fatal(err)
+ }
if _, _, err := w.fx.rel.Get(ctx, indexKey); err == nil {
t.Fatal("the index entry outlived the return's stage or start")
}
@@ -589,18 +593,21 @@ func TestReturnSweepDeletesOnlyTheEntryItJudged(t *testing.T) {
}
}
-// TestReturnIndexCleanupIsBounded (fix round 2, item 2): the return stage
-// or start door's index cleanup is best effort under its own bound: an
-// index store that stalls cannot hold the response or the serial bump
-// past offlineReturnCleanupTimeout; the entry it leaves is the sweep's.
+// TestReturnIndexCleanupIsBounded (fix round 2, item 2; fix round 3): the
+// return stage or start door's index cleanup is off the answer's path: with
+// the index store stalled, the serial bump and the 202 come at once (well
+// under the cleanup's own bound), and the cleanup runs after them, detached
+// and tracked (Server.Shutdown waits for it), giving up at its bound and
+// leaving the entry for the sweep.
func TestReturnIndexCleanupIsBounded(t *testing.T) {
defer func(d time.Duration) { offlineReturnCleanupTimeout = d }(offlineReturnCleanupTimeout)
- offlineReturnCleanupTimeout = 300 * time.Millisecond
+ offlineReturnCleanupTimeout = 2 * time.Second
for _, door := range []string{"/stage", ""} {
t.Run(map[string]string{"/stage": "stage", "": "start"}[door], func(t *testing.T) {
w := newReturnWorld(t, nil)
w.report("uk-box", "uk-box", "", w.request(), time.Now(), w.ready())
- w.fx.srv.cfg.Releases = &stallStore{Store: w.fx.rel, key: moves.ReturnIndexKey("uk", offlineHost, returnFwd)}
+ indexKey := moves.ReturnIndexKey("uk", offlineHost, returnFwd)
+ w.fx.srv.cfg.Releases = &stallStore{Store: w.fx.rel, key: indexKey}
before := serialBytes(t, w.fx.rel)
type answer struct {
code int
@@ -615,18 +622,27 @@ func TestReturnIndexCleanupIsBounded(t *testing.T) {
var a answer
select {
case a = <-done:
- case <-time.After(5 * time.Second):
+ case <-time.After(10 * time.Second):
t.Fatal("the door did not answer while the index store stalled")
}
+ took := time.Since(start)
if a.code != http.StatusAccepted {
t.Fatalf("%s = %d %s", door, a.code, a.out)
}
- if took := time.Since(start); took > 3*time.Second {
- t.Fatalf("the door answered after %v, past the cleanup's bound", took)
+ if took > offlineReturnCleanupTimeout/4 {
+ t.Fatalf("the door answered after %v: the stalled cleanup (bound %v) sat before the answer", took, offlineReturnCleanupTimeout)
}
if serialBytes(t, w.fx.rel) == before {
t.Fatal("the door did not bump the serial")
}
+ ctx, cancel := context.WithTimeout(context.Background(), 3*offlineReturnCleanupTimeout)
+ defer cancel()
+ if err := w.fx.srv.Shutdown(ctx); err != nil {
+ t.Fatalf("the detached cleanup outlived its bound: %v", err)
+ }
+ if _, _, err := w.fx.rel.Get(context.Background(), indexKey); err != nil {
+ t.Fatalf("the stalled cleanup removed the entry the sweep owns now: %v", err)
+ }
})
}
}
diff --git a/internal/console/server.go b/internal/console/server.go
index 70a150cdc..aaa787ceb 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -929,8 +929,12 @@ type Server struct {
// from their request (apiinstanceseed.go, apiinstanceimport.go), so a
// test can wait for one to finish.
seedCopies sync.WaitGroup
- cfg Config
- store *Store
+ // detached counts best-effort work started after a request answered
+ // (the return door's index cleanup, offlineBegin); Shutdown waits for
+ // it.
+ detached sync.WaitGroup
+ cfg Config
+ store *Store
// appRegionOf and beforeClaimInsert are TEST SEAMS (nil in
// production): a store failure on the app-region read (C20), and a
// competing insert landing after the region was resolved.
@@ -5295,3 +5299,18 @@ func summarize(v AppView, acct string) appRow {
}
return row
}
+
+// Shutdown waits, until ctx is done, for the best-effort work the server
+// started detached from a request after answering it (detached: the return
+// door's index cleanup, each bounded by its own deadline). It stops
+// nothing else.
+func (s *Server) Shutdown(ctx context.Context) error {
+ done := make(chan struct{})
+ go func() { s.detached.Wait(); close(done) }()
+ select {
+ case <-done:
+ return nil
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+}