amadan

carlos / platform Public

offline return: the sweep deletes only the index entry it judged (fix round 2, item 1)

selfServeReturnSweep keeps the ETag of the entry it read and judged
expired and deletes under it (deleteReturnIndexEntryAt); a fresh /return
that replaced the entry meanwhile no longer matches and keeps its entry.
The door's cleanup after a return stage or start still deletes whatever
entry the forward has.

Witness: TestReturnSweepDeletesOnlyTheEntryItJudged.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b4b58a76931f3173b7ece3c54741797ac04e5b1a parent eb2904d
2 files changed, +58 −4
  • internal/console/offlinereturn.go +16 −4
  • internal/console/offlinereturn_test.go +42 −0
diff --git a/internal/console/offlinereturn.go b/internal/console/offlinereturn.go
index f00e34d27..10c2a8251 100644
--- a/internal/console/offlinereturn.go
+++ b/internal/console/offlinereturn.go
@@ -215,7 +215,8 @@ func (s *Server) offlineReturnIndexDone(ctx context.Context, code, host, returnO
}
// deleteReturnIndexEntry deletes the index entry at key if it is still the
-// one read now; absent is done.
+// one read now; absent is done. The return door's cleanup: once the return
+// is staged or started, whatever entry the forward has 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) {
@@ -224,6 +225,14 @@ func (s *Server) deleteReturnIndexEntry(ctx context.Context, key string) error {
if err != nil {
return err
}
+ return s.deleteReturnIndexEntryAt(ctx, key, etag)
+}
+
+// deleteReturnIndexEntryAt deletes the index entry at key only while it is
+// the version etag names (the one its caller judged). A mismatch is a
+// newer entry (a fresh /return replaced it): it is kept, and that is not an
+// error. Absent is done.
+func (s *Server) deleteReturnIndexEntryAt(ctx context.Context, key, etag string) error {
cd, ok := s.cfg.Releases.(release.ConditionalDeleter)
if !ok {
return fmt.Errorf("the store cannot delete conditionally")
@@ -239,7 +248,8 @@ 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
+// (moves.ReturnRequest.Deadline) is deleted, if-match on the ETag of the
+// entry it read and judged, never a newer one. 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
@@ -262,7 +272,7 @@ func (s *Server) selfServeReturnSweep(ctx context.Context, now time.Time) {
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)
+ raw, etag, err := s.cfg.Releases.Get(ctx, key)
if err != nil {
continue
}
@@ -285,7 +295,9 @@ func (s *Server) selfServeReturnSweep(ctx context.Context, now time.Time) {
continue
}
}
- if err := s.deleteReturnIndexEntry(ctx, key); err != nil {
+ // Only the entry judged: a fresh /return that replaced it since
+ // keeps its entry (fix round 2).
+ if err := s.deleteReturnIndexEntryAt(ctx, key, etag); err != nil {
log.Printf("console: self-serve moves: sweeping the return index entry %s: %v (next tick)", key, err)
}
}
diff --git a/internal/console/offlinereturn_test.go b/internal/console/offlinereturn_test.go
index a738c29eb..9fed74c22 100644
--- a/internal/console/offlinereturn_test.go
+++ b/internal/console/offlinereturn_test.go
@@ -546,3 +546,45 @@ func TestReturnIndexIsTheConsoles(t *testing.T) {
}
})
}
+
+// TestReturnSweepDeletesOnlyTheEntryItJudged (fix round 2, item 1): the
+// sweep deletes an index entry only under the ETag of the entry it judged
+// expired. A fresh /return that replaces the entry between the judgment
+// and the delete keeps its entry: the delete's ETag no longer matches, and
+// the sweep skips it.
+func TestReturnSweepDeletesOnlyTheEntryItJudged(t *testing.T) {
+ ctx := context.Background()
+ w := newReturnWorld(t, nil)
+ indexKey := moves.ReturnIndexKey("uk", offlineHost, returnFwd)
+ 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)
+ }
+ freshKey := fmt.Sprintf("%d-c-bbbbbbbb", time.Now().UnixNano())
+ lt := &ltStore{Store: w.fx.rel}
+ replaced := false
+ lt.setHook(func(op, key string) error {
+ if op == "get" && key == moves.ReturnRequestKey(offlineHost, returnFwd, oldKey) && !replaced {
+ // The sweep has read the entry and now reads its request: a
+ // fresh /return lands meanwhile.
+ replaced = true
+ w.requestAt(time.Now().UTC(), freshKey)
+ if err := w.fx.srv.offlineReturnIndex(ctx, "uk", offlineHost, returnFwd, freshKey); err != nil {
+ t.Fatal(err)
+ }
+ }
+ return nil
+ })
+ w.fx.srv.cfg.Releases = lt
+ w.fx.srv.selfServeReturnSweep(ctx, time.Now())
+ if !replaced {
+ t.Fatal("the sweep never judged the expired entry")
+ }
+ raw, _, err := w.fx.rel.Get(ctx, indexKey)
+ var e moves.ReturnIndexEntry
+ if err != nil || json.Unmarshal(raw, &e) != nil || e.RequestKey != freshKey {
+ t.Fatalf("the fresh request's index entry did not survive the sweep: %s %v", raw, err)
+ }
+}