amadan

carlos / platform Public

host: a return request the scan cannot read answers nothing and keeps the index entry (final review F5, host half)

writeReturnReport skipped a request whose read failed, so with the newest
request unreadable it could answer an older one, or find none and return
nil, and answerReturn then deleted the index entry: the box never found
the newest request again. A failed read is now the scan's error: no
request is answered, the entry stays, and the next scan retries.

Witness: TestReturnRequestReadFailurePreservesIndexAndRejectsOlderReport
(one request; two requests with the newest unreadable, the older answered
or not).

Census: movesWorker.writeReturnReport's digest and the internal/host
envelope change; the re-pin is the console implementer's.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 036528e6eb50a8c33abc3c42a9e6f51eccbfa149 parent eda53b5
2 files changed, +167 −3
  • internal/host/movesoffline_return.go +7 −3
  • internal/host/movesoffline_return_test.go +160 −0
diff --git a/internal/host/movesoffline_return.go b/internal/host/movesoffline_return.go
index ce000b556..dc0fc914c 100644
--- a/internal/host/movesoffline_return.go
+++ b/internal/host/movesoffline_return.go
@@ -142,7 +142,7 @@ func (w *movesWorker) readOpen(ctx context.Context, s release.Store, host, id st
// 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.
+// 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) {
@@ -161,7 +161,8 @@ func (w *movesWorker) answerReturn(ctx context.Context, s release.Store, o *move
}
// writeReturnReport writes this box's report for the current request of
-// the forward o when there is none.
+// 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.
func (w *movesWorker) writeReturnReport(ctx context.Context, s release.Store, o *moves.OpenRecord) error {
cfg := w.cfg
prefix := moves.ReturnRequestPrefix(o.Host, o.MoveID)
@@ -169,11 +170,14 @@ func (w *movesWorker) writeReturnReport(ctx context.Context, s release.Store, o
if err != nil {
return err
}
+ // 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.
var reqs []moves.ReturnRequest
for _, k := range keys {
raw, _, err := s.Get(ctx, k)
if err != nil {
- continue
+ return fmt.Errorf("the return request %s: %w", k, err)
}
if rq, err := moves.ParseReturnRequest(raw, o.Host, o.MoveID, k); err == nil {
reqs = append(reqs, rq)
diff --git a/internal/host/movesoffline_return_test.go b/internal/host/movesoffline_return_test.go
index cec86306a..427b0137c 100644
--- a/internal/host/movesoffline_return_test.go
+++ b/internal/host/movesoffline_return_test.go
@@ -5,7 +5,9 @@ package host
import (
"context"
"encoding/json"
+ "errors"
"fmt"
+ "os"
"strings"
"sync"
"testing"
@@ -113,3 +115,161 @@ func TestReturnScanIsOneListPerMinute(t *testing.T) {
}
}
}
+
+// returnFaultStore fails the Gets of the keys in unreadable, as a store
+// that cannot serve one object for a while does.
+type returnFaultStore struct {
+ *release.DirStore
+ mu sync.Mutex
+ unreadable map[string]bool
+}
+
+func (f *returnFaultStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ f.mu.Lock()
+ bad := f.unreadable[key]
+ f.mu.Unlock()
+ if bad {
+ return nil, "", fmt.Errorf("get %s: the store is unavailable", key)
+ }
+ return f.DirStore.Get(ctx, key)
+}
+
+func (f *returnFaultStore) set(key string, bad bool) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if f.unreadable == nil {
+ f.unreadable = map[string]bool{}
+ }
+ f.unreadable[key] = bad
+}
+
+// returnWorld is a pending return the seal world's pinned source (ie-a)
+// answers: the forward offID's open record (ie to uk, the participants
+// pinned), an index entry in ie-a's region, and the requests the test
+// writes, over a store whose objects the test can make unreadable.
+type returnWorld struct {
+ *sealWorld
+ fs *returnFaultStore
+ indexKey string
+ t0 time.Time
+}
+
+func newReturnWorld(t *testing.T) *returnWorld {
+ sw := newSSWorld(t)
+ rw := &returnWorld{sealWorld: sw, fs: &returnFaultStore{DirStore: sw.s.(*release.DirStore)}, t0: sw.clock()}
+ p := ssParticipants()
+ rec := sw.record()
+ rw.put(moves.OpenRecordKey(offHost, offID), moves.OpenRecord{MoveID: offID, Host: offHost, Account: sw.acct, App: "app", Incarnation: rec.Incarnation,
+ From: "", To: "uk", StageBucket: sealUK, StageRegion: sealUKRegion, Participants: &p, Layout: "flat", OpenedAt: rw.t0.Add(-time.Hour)})
+ rw.indexKey = moves.ReturnIndexKey("", offHost, offID)
+ sw.ie.w.cfg.ReturnScanEvery = -1
+ return rw
+}
+
+// put writes key, over the object there if any (the return door moves an
+// index entry to the newest request if-match).
+func (rw *returnWorld) put(key string, v any) {
+ rw.t.Helper()
+ ctx := context.Background()
+ body, err := json.Marshal(v)
+ if err != nil {
+ rw.t.Fatal(err)
+ }
+ _, etag, err := rw.fs.DirStore.Get(ctx, key)
+ if err != nil && !errors.Is(err, os.ErrNotExist) {
+ rw.t.Fatal(err)
+ }
+ if _, err := rw.fs.DirStore.Put(ctx, key, body, etag); err != nil {
+ rw.t.Fatal(err)
+ }
+}
+
+// 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 {
+ 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(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) answered(key string) bool {
+ rw.t.Helper()
+ _, _, err := rw.fs.DirStore.Get(context.Background(), moves.ReturnReportKey(offHost, offID, "ie-a", key))
+ return err == nil
+}
+
+func (rw *returnWorld) indexed() bool {
+ rw.t.Helper()
+ _, _, err := rw.fs.DirStore.Get(context.Background(), rw.indexKey)
+ return err == nil
+}
+
+// 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.
+func TestReturnRequestReadFailurePreservesIndexAndRejectsOlderReport(t *testing.T) {
+ t.Run("one request, unreadable", func(t *testing.T) {
+ rw := newReturnWorld(t)
+ r1 := rw.request(1)
+ rw.fs.set(moves.ReturnRequestKey(offHost, offID, r1), true)
+ rw.scan()
+ if !rw.indexed() {
+ t.Fatal("the index entry was deleted while its only request could not be read")
+ }
+ if rw.answered(r1) {
+ t.Fatal("answered a request it could not read")
+ }
+ rw.fs.set(moves.ReturnRequestKey(offHost, offID, r1), false)
+ rw.scan()
+ if !rw.answered(r1) {
+ t.Fatal("no report once the request reads again")
+ }
+ if rw.indexed() {
+ t.Fatal("the index entry outlived the answer")
+ }
+ })
+ for _, older := range []string{"answered", "unanswered"} {
+ t.Run("two requests, the newest unreadable, the older "+older, func(t *testing.T) {
+ rw := newReturnWorld(t)
+ r1 := rw.request(1)
+ if older == "answered" {
+ rw.scan()
+ if !rw.answered(r1) || rw.indexed() {
+ t.Fatalf("the first request: answered %v, indexed %v", rw.answered(r1), rw.indexed())
+ }
+ }
+ r2 := rw.request(2)
+ rw.fs.set(moves.ReturnRequestKey(offHost, offID, r2), true)
+ rw.scan()
+ if !rw.indexed() {
+ t.Fatal("the index entry was deleted while the newest request could not be read")
+ }
+ if older == "unanswered" && rw.answered(r1) {
+ t.Fatal("answered the older request while the newest could not be read")
+ }
+ if rw.answered(r2) {
+ t.Fatal("answered a request it could not read")
+ }
+ rw.fs.set(moves.ReturnRequestKey(offHost, offID, r2), false)
+ rw.scan()
+ if !rw.answered(r2) {
+ t.Fatal("no report for the newest request once it reads again")
+ }
+ if older == "unanswered" && rw.answered(r1) {
+ t.Fatal("answered the superseded request")
+ }
+ if rw.indexed() {
+ t.Fatal("the index entry outlived the answer")
+ }
+ })
+ }
+}