amadan

carlos / platform Public

offline return: the request names its target and the console owns the index (final review F6, console half)

moves.ReturnRequest gains TargetBoxID (target_box_id), the BoxID of the
target the return door selects (offlineRegionChecks, as the stage door
selects it), and documents the contract for the boxes: a box answers only
a request naming its BoxID, a former source the console did not select
ignores it and never deletes the index, a request without the field names
no box. ReturnRequestLife (an hour) gives each request a Deadline.

The console owns the index entry's lifetime: a successful return stage or
start deletes it (if-match), and the pass sweeps, once per tick under the
lease and at most 32 entries resuming where it stopped, every entry whose
request is past its deadline (or names no request).

Witness: TestReturnIndexIsTheConsoles.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 748dec8b40e064bf6e573a8a9a47bbd3021abd46 parent f799f5d
6 files changed, +211 −3
  • internal/console/apiinstancemove_offline.go +5 −0
  • internal/console/offlinereturn.go +101 −1
  • internal/console/offlinereturn_test.go +65 −0
  • internal/console/selfserve_pass.go +5 −0
  • internal/console/server.go +3 −0
  • internal/moves/offline.go +32 −2
diff --git a/internal/console/apiinstancemove_offline.go b/internal/console/apiinstancemove_offline.go
index f789a8b93..7b530199d 100644
--- a/internal/console/apiinstancemove_offline.go
+++ b/internal/console/apiinstancemove_offline.go
@@ -997,6 +997,11 @@ func (s *Server) offlineBegin(w http.ResponseWriter, r *http.Request, d offlineD
s.offlineWriteFailed(w, d, err)
return
}
+ if sel.ReturnOf != "" {
+ // The return is staged or started: its index entry has done its
+ // work (final review F6).
+ s.offlineReturnIndexDone(context.WithoutCancel(ctx), sel.To, host, sel.ReturnOf)
+ }
s.bumpSerial(ctx)
writeJSON(w, http.StatusAccepted, s.offlineViewOf(ctx, got))
}
diff --git a/internal/console/offlinereturn.go b/internal/console/offlinereturn.go
index 0f822a996..f00e34d27 100644
--- a/internal/console/offlinereturn.go
+++ b/internal/console/offlinereturn.go
@@ -29,6 +29,7 @@ import (
"net/http"
"os"
"path"
+ "sort"
"strings"
"time"
@@ -137,8 +138,15 @@ func (s *Server) handleAPIInstanceOfflineReturn(w http.ResponseWriter, r *http.R
writeRefusal(w, ref)
return
}
+ // The return's target, selected as the stage door will select it: the
+ // request names it, and only that box answers (final review F6).
+ sel, ref := s.offlineRegionChecks(ctx, d.rec.Account, d.rec.App, d.host, d.rec, region, false, false, req.ReturnOf)
+ if ref != nil {
+ writeRefusal(w, ref)
+ return
+ }
now := time.Now().UTC()
- rq := moves.ReturnRequest{RequestedAt: now, Region: code, Console: s.selfServeConsoleID()}
+ rq := moves.ReturnRequest{RequestedAt: now, Region: code, Console: s.selfServeConsoleID(), TargetBoxID: sel.Target.BoxID}
rq.Key = moves.NewReturnRequestKey(now, rq.Console, rand.Reader)
key := moves.ReturnRequestKey(d.host, req.ReturnOf, rq.Key)
body, err := json.Marshal(rq)
@@ -191,6 +199,98 @@ func (s *Server) offlineReturnIndex(ctx context.Context, code, host, returnOf, r
return err
}
+// offlineReturnIndexDone deletes the return index entry of returnOf in
+// region code once the return is staged or started (the console owns the
+// entry's lifetime, final review F6): if-match on the entry it read, so
+// only that entry goes. A failure is logged; the pass's sweep deletes it
+// once its request is past its deadline.
+func (s *Server) offlineReturnIndexDone(ctx context.Context, code, host, returnOf string) {
+ key := moves.ReturnIndexKey(code, host, returnOf)
+ if key == "" {
+ return
+ }
+ if err := s.deleteReturnIndexEntry(ctx, key); err != nil {
+ log.Printf("console: offline move %s: deleting the return index of %s: %v (the pass sweeps it)", host, returnOf, err)
+ }
+}
+
+// deleteReturnIndexEntry deletes the index entry at key if it is still the
+// one read now; absent is done.
+func (s *Server) deleteReturnIndexEntry(ctx context.Context, key string) error {
+ _, etag, err := s.cfg.Releases.Get(ctx, key)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ cd, ok := s.cfg.Releases.(release.ConditionalDeleter)
+ if !ok {
+ return fmt.Errorf("the store cannot delete conditionally")
+ }
+ if err := cd.DeleteIfMatch(ctx, key, etag); err != nil && !errors.Is(err, os.ErrNotExist) && !errors.Is(err, release.ErrCASConflict) {
+ return err
+ }
+ return nil
+}
+
+// selfServeReturnSweepMax bounds the index entries one tick examines.
+const selfServeReturnSweepMax = 32
+
+// selfServeReturnSweep is the pass's sweep of the return index (once per
+// tick, under the lease): an entry whose request is past its deadline
+// (moves.ReturnRequest.Deadline) is deleted, if-match. At most
+// selfServeReturnSweepMax entries per tick, resuming after the last one
+// examined, so every entry is reached. An entry or request that cannot be
+// read is left for a later tick; an entry naming no request (the request is
+// written first, so it never existed or the entry is not the door's) is
+// deleted.
+func (s *Server) selfServeReturnSweep(ctx context.Context, now time.Time) {
+ keys, err := s.cfg.Releases.List(ctx, moves.ReturnIndexRoot())
+ if err != nil {
+ log.Printf("console: self-serve moves: the return index: %v (swept next tick)", err)
+ return
+ }
+ sort.Strings(keys)
+ start := sort.SearchStrings(keys, s.returnSweepAfter)
+ if start < len(keys) && keys[start] == s.returnSweepAfter {
+ start++
+ }
+ if start >= len(keys) {
+ start = 0
+ }
+ for n := 0; n < selfServeReturnSweepMax && n < len(keys); n++ {
+ key := keys[(start+n)%len(keys)]
+ s.returnSweepAfter = key
+ raw, _, err := s.cfg.Releases.Get(ctx, key)
+ if err != nil {
+ continue
+ }
+ var e moves.ReturnIndexEntry
+ if json.Unmarshal(raw, &e) != nil {
+ continue
+ }
+ rkey := moves.ReturnRequestKey(e.Host, e.ReturnOf, e.RequestKey)
+ if rkey == "" {
+ continue
+ }
+ rraw, _, err := s.cfg.Releases.Get(ctx, rkey)
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ case err != nil:
+ continue
+ default:
+ rq, err := moves.ParseReturnRequest(rraw, e.Host, e.ReturnOf, rkey)
+ if err != nil || now.Before(rq.Deadline()) {
+ continue
+ }
+ }
+ if err := s.deleteReturnIndexEntry(ctx, key); err != nil {
+ log.Printf("console: self-serve moves: sweeping the return index entry %s: %v (next tick)", key, err)
+ }
+ }
+}
+
// readReturnReport is the return-ready report of the current request of
// returnOf from the target box sel names: it must exist, be written within
// offlineReturnFresh, answer that request (its key and time), come from a
diff --git a/internal/console/offlinereturn_test.go b/internal/console/offlinereturn_test.go
index 7426658da..a738c29eb 100644
--- a/internal/console/offlinereturn_test.go
+++ b/internal/console/offlinereturn_test.go
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
+ "fmt"
"net/http"
"testing"
"time"
@@ -481,3 +482,67 @@ func TestReturnRequestReadFailureIsUnreadable(t *testing.T) {
}
}
}
+
+// TestReturnIndexIsTheConsoles (final review F6, console half): the return
+// request names the box the console selected as the return's target
+// (target_box_id), and the console owns the region index entry's lifetime:
+// a successful return stage or start deletes it, and the pass sweeps an
+// entry whose request is past its deadline, leaving a live one.
+func TestReturnIndexIsTheConsoles(t *testing.T) {
+ ctx := context.Background()
+ indexKey := moves.ReturnIndexKey("uk", offlineHost, returnFwd)
+ t.Run("the request names the selected target", func(t *testing.T) {
+ w := newReturnWorld(t, nil)
+ key := w.request()
+ raw, _, err := w.fx.rel.Get(ctx, moves.ReturnRequestKey(offlineHost, returnFwd, key))
+ if err != nil {
+ t.Fatal(err)
+ }
+ rq, err := moves.ParseReturnRequest(raw, offlineHost, returnFwd, moves.ReturnRequestKey(offlineHost, returnFwd, key))
+ if err != nil || rq.TargetBoxID != "uk-box" {
+ t.Fatalf("the request %s names target %q (%v), want uk-box", raw, rq.TargetBoxID, err)
+ }
+ })
+ for _, door := range []string{"/stage", ""} {
+ t.Run("a return "+map[string]string{"/stage": "staged", "": "started"}[door]+" deletes the index entry", func(t *testing.T) {
+ w := newReturnWorld(t, nil)
+ w.report("uk-box", "uk-box", "", w.request(), time.Now(), w.ready())
+ if _, _, err := w.fx.rel.Get(ctx, indexKey); err != nil {
+ t.Fatalf("no index entry before the return: %v", err)
+ }
+ if code, out := w.req(http.MethodPost, door, `{"region":"uk","return_of":"`+returnFwd+`"}`); code != http.StatusAccepted {
+ t.Fatalf("%s = %d %s", door, code, out)
+ }
+ if _, _, err := w.fx.rel.Get(ctx, indexKey); err == nil {
+ t.Fatal("the index entry outlived the return's stage or start")
+ }
+ })
+ }
+ t.Run("the pass sweeps an entry past its request's deadline", func(t *testing.T) {
+ w := newReturnWorld(t, nil)
+ old := time.Now().Add(-moves.ReturnRequestLife - time.Minute).UTC()
+ oldKey := fmt.Sprintf("%d-c-aaaaaaaa", old.UnixNano())
+ w.requestAt(old, oldKey)
+ if err := w.fx.srv.offlineReturnIndex(ctx, "uk", offlineHost, returnFwd, oldKey); err != nil {
+ t.Fatal(err)
+ }
+ // A live one of another forward of another host stays.
+ other := "mv_otherforwardotherforwardot"
+ liveKey := fmt.Sprintf("%d-c-bbbbbbbb", time.Now().UnixNano())
+ body, _ := json.Marshal(moves.ReturnRequest{RequestedAt: time.Now().UTC(), Region: "uk", Console: "c", Key: liveKey, TargetBoxID: "uk-box"})
+ offlinePut(t, w.fx.rel, moves.ReturnRequestKey("beta.shop.example.com", other, liveKey), body)
+ if err := w.fx.srv.offlineReturnIndex(ctx, "uk", "beta.shop.example.com", other, liveKey); err != nil {
+ t.Fatal(err)
+ }
+ w.fx.srv.selfServeIteration(ctx)
+ if _, _, err := w.fx.rel.Get(ctx, indexKey); err == nil {
+ t.Fatal("the pass left an entry whose request is past its deadline")
+ }
+ if _, _, err := w.fx.rel.Get(ctx, moves.ReturnIndexKey("uk", "beta.shop.example.com", other)); err != nil {
+ t.Fatalf("the pass swept a live entry: %v", err)
+ }
+ if _, _, err := w.fx.rel.Get(ctx, moves.ReturnRequestKey(offlineHost, returnFwd, oldKey)); err != nil {
+ t.Fatalf("the sweep deleted the request itself: %v", err)
+ }
+ })
+}
diff --git a/internal/console/selfserve_pass.go b/internal/console/selfserve_pass.go
index 0bf260f08..c0ac708b9 100644
--- a/internal/console/selfserve_pass.go
+++ b/internal/console/selfserve_pass.go
@@ -139,6 +139,11 @@ func (s *Server) selfServeTick(ctx context.Context) {
defer stop()
s.drainSelfServeEvents(ctx, t)
s.selfServeOrphanScan(ctx, s.selfServeNow())
+ if t.valid(s.selfServeNow()) {
+ sctx, cancel := context.WithDeadline(ctx, t.deadline())
+ s.selfServeReturnSweep(sctx, s.selfServeNow())
+ cancel()
+ }
rows, err := s.store.OpenSelfServeMoves()
if err != nil {
log.Printf("console: self-serve moves: the open rows: %v (retrying next tick)", err)
diff --git a/internal/console/server.go b/internal/console/server.go
index 3be110587..9c13674a5 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -891,6 +891,9 @@ type Server struct {
// (nil is time.Now; a TEST SEAM).
selfServeIDOnce sync.Once
selfServeConsole string
+ // returnSweepAfter is the last return index key the pass's sweep
+ // examined (selfServeReturnSweep resumes after it). The pass's only.
+ returnSweepAfter string
selfServeLeaseMu sync.Mutex
selfServeLease *leaseToken
selfServeClock func() time.Time
diff --git a/internal/moves/offline.go b/internal/moves/offline.go
index 428594c67..d98664e2c 100644
--- a/internal/moves/offline.go
+++ b/internal/moves/offline.go
@@ -1073,13 +1073,37 @@ func StagedMatches(r OfflineReport, bucket, region string, request time.Time, ta
// prefix at ReturnRequestKey. Key is the request's identity: a new call is
// a new request, and the current request is the one whose (RequestedAt,
// Key) sorts last (CurrentReturnRequest); older ones are ignored.
+//
+// TargetBoxID is the BoxID of the box the console selected as the return's
+// target when it admitted the request (final review F6). The contract with
+// the boxes:
+//
+// - a box answers only a request naming its own BoxID; any other former
+// source of the forward (one the console did not select) ignores the
+// request, writes no report for it and never deletes the region's
+// index entry (ReturnIndexEntry);
+// - the console owns the index entry's lifetime: it deletes the entry
+// when the return is staged or started, and its moves pass deletes an
+// entry whose current request is past its Deadline. A box only reads
+// the index;
+// - a request with no TargetBoxID (written before the field existed)
+// names no box, so no box answers it; the operator asks again.
type ReturnRequest struct {
RequestedAt time.Time `json:"requested_at"`
Region string `json:"region"`
Console string `json:"console"`
Key string `json:"key"`
+ TargetBoxID string `json:"target_box_id"`
}
+// ReturnRequestLife is how long a return request stays pending: past
+// RequestedAt + ReturnRequestLife (Deadline) the console sweeps its index
+// entry, and the operator asks again if the return is still wanted.
+const ReturnRequestLife = time.Hour
+
+// Deadline is when the request stops being pending (ReturnRequestLife).
+func (r ReturnRequest) Deadline() time.Time { return r.RequestedAt.Add(ReturnRequestLife) }
+
// NewReturnRequestKey is a request key: the request's time in Unix
// nanoseconds, the console's id and 8 random hex digits, "" when console is
// not a plain name or random fails.
@@ -1172,8 +1196,10 @@ func ReturnParticipant(o OpenRecord, boxID, instanceID, placement string) bool {
// ReturnIndexEntry is the region-scoped pointer the return door writes
// beside each request (ReturnIndexKey), so the boxes of the return's
// target region find pending returns with one listing: the forward and the
-// newest request's key. The box deletes it (if-match) once its report is
-// written; the request and the report stay the record.
+// newest request's key. The console owns its lifetime (ReturnRequest): it
+// deletes the entry when the return is staged or started, or when the
+// request it names is past its Deadline; the request and the report stay
+// the record.
type ReturnIndexEntry struct {
Host string `json:"host"`
ReturnOf string `json:"return_of"`
@@ -1199,6 +1225,10 @@ func ReturnIndexPrefix(code string) string {
return offlineReturnsRoot + seg + "/"
}
+// ReturnIndexRoot is where every region's return index lives (the
+// console's sweep lists it).
+func ReturnIndexRoot() string { return offlineReturnsRoot }
+
// offlineReturnsRoot is where the return index lives; "@returns" is never a
// host (provision.ValidHostShape), so it never meets a move's prefix.
const offlineReturnsRoot = "control/offline-moves/@returns/"