carlos / platform Public

host: only the selected target answers a return, and the box never deletes the index (final review F6, host half)

The first former source to scan wrote its report and deleted the region's
index entry, so the box the console selected as the return's target never
found the request. The box now answers only the current request whose
TargetBoxID is its own BoxID: another former source, and every box for a
request with no TargetBoxID (written before the field), writes nothing.
answerReturn and its DeleteIfMatch are gone: the console owns the entry's
lifetime (it deletes it on the return's stage or start, and sweeps it past
the request's Deadline), and the box only reads it.

Witness: TestReturnIndexWaitsForSelectedTarget (former sources ie-a and
ie-b, the request naming ie-b: ie-a scans first and writes and changes
nothing, ie-b reports, the entry unchanged; a request with no target
answered by nobody). The F5 and F7 witnesses now require the entry to
stay. TestOperatorReturnHandshake (e2e) requires the entry after the
report and its removal by the stage door.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 1 day ago a2abea21a870b6f05990efcb24a31eb1cfc5d394 parent fcef25a

3 files changed, +148 −55

internal/e2e/selfserve/return_test.go +7 −2
internal/host/movesoffline_return.go +28 −34
internal/host/movesoffline_return_test.go +113 −19

diff --git a/internal/e2e/selfserve/return_test.go b/internal/e2e/selfserve/return_test.go

index 2d76ade94..939f4aacd 100644

--- a/internal/e2e/selfserve/return_test.go

+++ b/internal/e2e/selfserve/return_test.go

@@ -97,8 +97,10 @@ func TestOperatorReturnHandshake(t *testing.T) {

if !ok || r1.RequestKey != first || r1.Ready == nil || r1.Ready.Leftovers == nil || r1.Ready.Leftovers.ForwardID != fwd {

t.Fatalf("the first report: %v %+v", ok, r1)

}

- if _, _, err := w.raw.Get(w.ctx, moves.ReturnIndexKey("", host.E2EHost, fwd)); err == nil {

- t.Fatal("the index entry is still there after the report")

+ // The box only reads the index entry; the console owns it (final

+ // review F6) and removes it once the return is staged.

+ if _, _, err := w.raw.Get(w.ctx, moves.ReturnIndexKey("", host.E2EHost, fwd)); err != nil {

+ t.Fatalf("the index entry after the report: %v", err)

}

second := w.returnRequest(fwd)

src.ArmOffline() // an agent restart between the request and its report

@@ -133,6 +135,9 @@ func TestOperatorReturnHandshake(t *testing.T) {

if rec := w.record(); rec.Move == nil || rec.Move.State != provision.OfflineStaging || rec.Move.ReturnOf != fwd {

t.Fatalf("after the return's stage: %+v", rec.Move)

}

+ if _, _, err := w.raw.Get(w.ctx, moves.ReturnIndexKey("", host.E2EHost, fwd)); err == nil {

+ t.Fatal("the index entry is still there after the return's stage")

+ }

}

// clearTimers deletes the box's timer rows for the host, as an operator.

diff --git a/internal/host/movesoffline_return.go b/internal/host/movesoffline_return.go

index 601b493dc..37032f4bb 100644

--- a/internal/host/movesoffline_return.go

+++ b/internal/host/movesoffline_return.go

@@ -1,10 +1,11 @@

package host

// The operator's return, the box's half (self-serve moves plan 2 Task 8;

-// spec §7.5). A box that was a source of a completed forward move answers

-// each return request with an immutable return-ready report of what the

-// forward left on it: the console's stage or start door accepts exactly

-// that set (internal/console offlinereturn.go).

+// spec §7.5). The box the console selected as a return's target, a source

+// of the completed forward move, answers each return request naming it

+// (ReturnRequest.TargetBoxID) with an immutable return-ready report of

+// what the forward left on it: the console's stage or start door accepts

+// exactly that set (internal/console offlinereturn.go).

// The scan: once every offlineReturnScanEvery the box lists the return

// index of its own region (moves.ReturnIndexPrefix: one LIST, whatever the

@@ -13,12 +14,18 @@ package host

// (cached, and a record it knows it may not answer for cached as a

// negative; with its own instance id not known yet, neither), lists

// that forward's requests, takes the current one

-// (moves.CurrentReturnRequest) and, when no report of this box answers it

-// yet, writes one create-only; then it deletes the index entry if it is

-// still the one it read (a newer request moved it, and the next scan

-// answers that). A create-only conflict is the request answered already;

-// nothing is overwritten, so no ordering between writers matters. Nothing

-// runs between scans.

+// (moves.CurrentReturnRequest) and, when it names this box and no report

+// of this box answers it yet, writes one create-only. A request naming

+// another box, or none (written before the field), is not this box's to

+// answer. A create-only conflict is the request answered already; nothing

+// is overwritten, so no ordering between writers matters. Nothing runs

+// between scans.

+//

+// The box only reads the index: the console owns each entry's lifetime

+// (final review F6). It deletes the entry when the return is staged or

+// started, and sweeps an entry whose current request is past its

+// ReturnRequest.Deadline. Until then every scan reads the entry again and

+// finds its report already written.

import (

"context"

@@ -145,7 +152,7 @@ func (w *movesWorker) offlineReturns(ctx context.Context, s release.Store) {

if !known {

continue

}

- if err := w.answerReturn(ctx, s, o, key); err != nil {

+ if err := w.writeReturnReport(ctx, s, o); err != nil {

w.notice(k+"#offline-return", "moves: %s: the return of %s: %v", host, fwd, err)

}

@@ -168,29 +175,10 @@ func (w *movesWorker) readOpen(ctx context.Context, s release.Store, host, id st

return &o, nil

}

-// answerReturn writes this box's report for the current request of the

-// forward o, once, and then deletes the index entry indexKey if it is still

-// the one read before. An error keeps the entry: the next scan tries again.

-func (w *movesWorker) answerReturn(ctx context.Context, s release.Store, o *moves.OpenRecord, indexKey string) error {

- _, indexETag, err := s.Get(ctx, indexKey)

- if errors.Is(err, os.ErrNotExist) {

- return nil // answered and removed meanwhile

- }

- if err != nil {

- return err

- }

- if err := w.writeReturnReport(ctx, s, o); err != nil {

- return err

- }

- if err := release.DeleteIfMatch(ctx, s, indexKey, indexETag); err != nil && !errors.Is(err, release.ErrCASConflict) {

- return fmt.Errorf("the index entry: %w", err)

- }

- return nil

// writeReturnReport writes this box's report for the current request of

-// the forward o when there is none. A request it cannot read is an error:

-// which request is current is not known until every one reads.

+// the forward o when the request names this box (TargetBoxID) and there is

+// none. A request it cannot read is an error, and the next scan tries

+// again: which request is current is not known until every one reads.

func (w *movesWorker) writeReturnReport(ctx context.Context, s release.Store, o *moves.OpenRecord) error {

cfg := w.cfg

prefix := moves.ReturnRequestPrefix(o.Host, o.MoveID)

@@ -200,7 +188,7 @@ func (w *movesWorker) writeReturnReport(ctx context.Context, s release.Store, o

}

// Every request is read, or none is answered: with one unread, the

// newest is not known (an older one may be answered already, or

- // superseded), and the error keeps the index entry for the next scan.

+ // superseded).

var reqs []moves.ReturnRequest

for _, k := range keys {

raw, _, err := s.Get(ctx, k)

@@ -215,6 +203,12 @@ func (w *movesWorker) writeReturnReport(ctx context.Context, s release.Store, o

if !ok {

return nil

}

+ // Only the target the console selected answers: another former source

+ // of the forward writes nothing, and a request naming no box (written

+ // before TargetBoxID) is nobody's; the operator asks again.

+ if cur.TargetBoxID == "" || cur.TargetBoxID != cfg.BoxID {

+ return nil

+ }

key := moves.ReturnReportKey(o.Host, o.MoveID, cfg.BoxID, cur.Key)

if key == "" {

return fmt.Errorf("no report key for request %s", cur.Key)

diff --git a/internal/host/movesoffline_return_test.go b/internal/host/movesoffline_return_test.go

index 2b29426f6..8a445cc65 100644

--- a/internal/host/movesoffline_return_test.go

+++ b/internal/host/movesoffline_return_test.go

@@ -184,38 +184,66 @@ func (rw *returnWorld) put(key string, v any) {

}

-// request writes a return request as the return door does, and points the

-// index entry at it; it returns the request's key.

-func (rw *returnWorld) request(n int) string {

+// request writes a return request naming ie-a its target, as the return

+// door does, and points the index entry at it; it returns the request's

+// key.

+func (rw *returnWorld) request(n int) string { return rw.requestFor(n, "ie-a") }

+// requestFor is request with the target the console selected: target is

+// the request's TargetBoxID ("" for a request written before the field).

+func (rw *returnWorld) requestFor(n int, target string) string {

rw.t.Helper()

at := rw.t0.Add(time.Duration(n) * time.Minute)

key := fmt.Sprintf("%d-console-%08x", at.UnixNano(), n)

- rw.put(moves.ReturnRequestKey(offHost, offID, key), moves.ReturnRequest{RequestedAt: at, Region: "", Console: "console", Key: key})

+ rw.put(moves.ReturnRequestKey(offHost, offID, key), moves.ReturnRequest{RequestedAt: at, Region: "", Console: "console", Key: key, TargetBoxID: target})

rw.put(rw.indexKey, moves.ReturnIndexEntry{Host: offHost, ReturnOf: offID, RequestKey: key})

return key

}

-func (rw *returnWorld) scan() {

- rw.ie.w.offlineReturns(context.Background(), rw.fs)

+func (rw *returnWorld) scan() { rw.scanBy(rw.ie) }

+func (rw *returnWorld) scanBy(b *offBox) {

+ b.w.offlineReturns(context.Background(), rw.fs)

}

-func (rw *returnWorld) answered(key string) bool {

+func (rw *returnWorld) answered(key string) bool { return rw.answeredBy("ie-a", key) }

+func (rw *returnWorld) answeredBy(boxID, key string) bool {

rw.t.Helper()

- _, _, err := rw.fs.DirStore.Get(context.Background(), moves.ReturnReportKey(offHost, offID, "ie-a", key))

+ _, _, err := rw.fs.DirStore.Get(context.Background(), moves.ReturnReportKey(offHost, offID, boxID, key))

return err == nil

}

-func (rw *returnWorld) indexed() bool {

+func (rw *returnWorld) indexed() bool { return rw.indexETag() != "" }

+// indexETag is the index entry's ETag, "" when it is gone.

+func (rw *returnWorld) indexETag() string {

rw.t.Helper()

- _, _, err := rw.fs.DirStore.Get(context.Background(), rw.indexKey)

- return err == nil

+ _, etag, err := rw.fs.DirStore.Get(context.Background(), rw.indexKey)

+ if errors.Is(err, os.ErrNotExist) {

+ return ""

+ }

+ if err != nil {

+ rw.t.Fatal(err)

+ }

+ return etag

+// reports is every report under the forward's prefix.

+func (rw *returnWorld) reports() []string {

+ rw.t.Helper()

+ keys, err := rw.fs.DirStore.List(context.Background(), moves.OfflineReportPrefix(offHost, offID)+"reports/")

+ if err != nil {

+ rw.t.Fatal(err)

+ }

+ return keys

}

// TestReturnRequestReadFailurePreservesIndexAndRejectsOlderReport (final

// review F5): a scan that cannot read one of a forward's return requests

// cannot tell which request is newest, so it answers none of them and

// keeps the index entry; once the request reads again, the next scan

-// answers it and only then removes the entry.

+// answers it. The entry stays: the console owns it (F6).

func TestReturnRequestReadFailurePreservesIndexAndRejectsOlderReport(t *testing.T) {

t.Run("one request, unreadable", func(t *testing.T) {

rw := newReturnWorld(t)

@@ -233,8 +261,8 @@ func TestReturnRequestReadFailurePreservesIndexAndRejectsOlderReport(t *testing.

if !rw.answered(r1) {

t.Fatal("no report once the request reads again")

}

- if rw.indexed() {

- t.Fatal("the index entry outlived the answer")

+ if !rw.indexed() {

+ t.Fatal("the box deleted the index entry: the console owns it")

}

})

for _, older := range []string{"answered", "unanswered"} {

@@ -243,7 +271,7 @@ func TestReturnRequestReadFailurePreservesIndexAndRejectsOlderReport(t *testing.

r1 := rw.request(1)

if older == "answered" {

rw.scan()

- if !rw.answered(r1) || rw.indexed() {

+ if !rw.answered(r1) || !rw.indexed() {

t.Fatalf("the first request: answered %v, indexed %v", rw.answered(r1), rw.indexed())

}

@@ -267,8 +295,8 @@ func TestReturnRequestReadFailurePreservesIndexAndRejectsOlderReport(t *testing.

if older == "unanswered" && rw.answered(r1) {

t.Fatal("answered the superseded request")

}

- if rw.indexed() {

- t.Fatal("the index entry outlived the answer")

+ if !rw.indexed() {

+ t.Fatal("the box deleted the index entry: the console owns it")

}

})

}

@@ -298,8 +326,8 @@ func TestReturnScanRetriesUnknownInstanceID(t *testing.T) {

if !rw.answered(r1) {

t.Fatal("no report once the box's instance id is known")

}

- if rw.indexed() {

- t.Fatal("the index entry outlived the answer")

+ if !rw.indexed() {

+ t.Fatal("the box deleted the index entry: the console owns it")

}

// A box that is known not to be the pinned source (another instance)

@@ -313,3 +341,69 @@ func TestReturnScanRetriesUnknownInstanceID(t *testing.T) {

t.Fatal("a replaced box answered for the pinned source")

}

+// TestReturnIndexWaitsForSelectedTarget (final review F6, the host half):

+// a return request names the box the console selected as its target

+// (TargetBoxID), and only that box answers it. Former sources A (ie-a) and

+// B (ie-b), the request naming B: A scans first and writes nothing and

+// changes nothing; B then reports. No box deletes the index entry (the

+// console owns its lifetime). A request with no TargetBoxID (written

+// before the field) is answered by nobody.

+func TestReturnIndexWaitsForSelectedTarget(t *testing.T) {

+ setup := func(t *testing.T) (*returnWorld, *offBox) {

+ rw := newReturnWorld(t)

+ b := rw.sourceBox("ie-b")

+ ssIdentity(b, "i-ie-b")

+ b.w.cfg.ReturnScanEvery = -1

+ p := ssParticipants()

+ p.Source = append(p.Source, provision.Participant{BoxID: "ie-b", Label: "ie-b", InstanceID: "i-ie-b", Verified: true})

+ rec := rw.record()

+ rw.put(moves.OpenRecordKey(offHost, offID), moves.OpenRecord{MoveID: offID, Host: offHost, Account: rw.acct, App: "app", Incarnation: rec.Incarnation,

+ From: "", To: "uk", StageBucket: sealUK, StageRegion: sealUKRegion, Participants: &p, Layout: "flat", OpenedAt: rw.t0.Add(-time.Hour)})

+ return rw, b

+ }

+ t.Run("the request names B", func(t *testing.T) {

+ rw, b := setup(t)

+ r := rw.requestFor(1, "ie-b")

+ etag := rw.indexETag()

+ rw.scanBy(rw.ie)

+ if got := rw.reports(); len(got) != 0 {

+ t.Fatalf("a former source the console did not select wrote %v", got)

+ }

+ if rw.indexETag() != etag {

+ t.Fatal("a former source the console did not select changed the index entry")

+ }

+ rw.scanBy(b)

+ if !rw.answeredBy("ie-b", r) {

+ t.Fatal("the selected target did not report")

+ }

+ if rw.answeredBy("ie-a", r) {

+ t.Fatal("the other former source reported")

+ }

+ if rw.indexETag() != etag {

+ t.Fatal("the selected target changed the index entry: the console owns it")

+ }

+ rw.scanBy(rw.ie)

+ if got := rw.reports(); len(got) != 1 {

+ t.Fatalf("reports after a later scan by A: %v", got)

+ }

+ })

+ t.Run("a request with no target", func(t *testing.T) {

+ rw, b := setup(t)

+ rw.requestFor(1, "")

+ etag := rw.indexETag()

+ for i := 0; i < 2; i++ {

+ rw.scanBy(rw.ie)

+ rw.scanBy(b)

+ }

+ if got := rw.reports(); len(got) != 0 {

+ t.Fatalf("a request naming no box was answered: %v", got)

+ }

+ if rw.indexETag() != etag {

+ t.Fatal("a box changed the index entry of a request naming no box")

+ }

+ })