amadan

carlos / platform Public

offline return: the hourly store sweep also clears the return index (fix round 2, item 3)

trashSweep, the expiry pass's hourly store sweep on every claims-off
console, now runs returnSweepHourly whatever the self-serve mode, bounded
(256 entries, 30 s), so a switch-off console with no recovery evidence,
whose moves pass never runs, still clears expired entries. The pass's
sweep and the hourly one share the resume cursor under a mutex; both
delete if-match on the entry judged.

Witness: TestReturnIndexSweptOnADormantConsole.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 0d3763da7d612ebd613b62512464127ec861df4f parent 1fc116c
4 files changed, +61 −4
  • internal/console/lifetimes_purge.go +5 −1
  • internal/console/offlinereturn.go +28 −1
  • internal/console/offlinereturn_test.go +24 −0
  • internal/console/server.go +4 −2
diff --git a/internal/console/lifetimes_purge.go b/internal/console/lifetimes_purge.go
index bc1fa6e6b..615e1542f 100644
--- a/internal/console/lifetimes_purge.go
+++ b/internal/console/lifetimes_purge.go
@@ -154,7 +154,8 @@ const trashSweepEvery = time.Hour
// boxes stop). At most once per trashSweepEvery by now. Each app is purged
// under its lock with the marker re-read there, after purgeOne's own
// refusals (the control plane, email estate or an unreadable one); with
-// lifetimes off it is the legacy purge the boxes ran.
+// lifetimes off it is the legacy purge the boxes ran. It also sweeps the
+// operator return's index (returnSweepHourly).
func (s *Server) trashSweep(ctx context.Context, now time.Time) {
if s.cfg.HostClaim || s.cfg.Releases == nil {
return
@@ -168,6 +169,9 @@ func (s *Server) trashSweep(ctx context.Context, now time.Time) {
if !due {
return
}
+ // The return index's expired entries, on every claims-off console,
+ // whatever the self-serve mode (fix round 2): bounded, best effort.
+ s.returnSweepHourly(ctx, now)
apps, err := release.ExpiredTrashApps(ctx, s.cfg.Releases, now)
if err != nil {
log.Printf("console: trash sweep: list the trash: %v", err)
diff --git a/internal/console/offlinereturn.go b/internal/console/offlinereturn.go
index e13c2d40d..6bc030897 100644
--- a/internal/console/offlinereturn.go
+++ b/internal/console/offlinereturn.go
@@ -260,6 +260,33 @@ const selfServeReturnSweepMax = 32
// 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) {
+ s.returnSweep(ctx, now, selfServeReturnSweepMax)
+}
+
+// returnSweepHourlyMax bounds the entries the hourly store sweep examines,
+// and returnSweepHourlyTimeout its time.
+const (
+ returnSweepHourlyMax = 256
+ returnSweepHourlyTimeout = 30 * time.Second
+)
+
+// returnSweepHourly is the return index sweep the hourly store sweep runs
+// (trashSweep), whatever the self-serve mode (fix round 2): a switch-off
+// console with no recovery evidence never runs the moves pass, and its
+// expired entries are cleared here. Bounded in entries and time; the
+// deletes are the pass's (if-match on the entry judged), so a second
+// console sweeping at once deletes nothing twice.
+func (s *Server) returnSweepHourly(ctx context.Context, now time.Time) {
+ ctx, cancel := context.WithTimeout(ctx, returnSweepHourlyTimeout)
+ defer cancel()
+ s.returnSweep(ctx, now, returnSweepHourlyMax)
+}
+
+// returnSweep examines at most max index entries, resuming after the last
+// one any sweep examined (the cursor is shared under returnSweepMu).
+func (s *Server) returnSweep(ctx context.Context, now time.Time, max int) {
+ s.returnSweepMu.Lock()
+ defer s.returnSweepMu.Unlock()
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)
@@ -273,7 +300,7 @@ func (s *Server) selfServeReturnSweep(ctx context.Context, now time.Time) {
if start >= len(keys) {
start = 0
}
- for n := 0; n < selfServeReturnSweepMax && n < len(keys); n++ {
+ for n := 0; n < max && n < len(keys); n++ {
key := keys[(start+n)%len(keys)]
s.returnSweepAfter = key
raw, etag, err := s.cfg.Releases.Get(ctx, key)
diff --git a/internal/console/offlinereturn_test.go b/internal/console/offlinereturn_test.go
index 632e83c78..254f3f866 100644
--- a/internal/console/offlinereturn_test.go
+++ b/internal/console/offlinereturn_test.go
@@ -630,3 +630,27 @@ func TestReturnIndexCleanupIsBounded(t *testing.T) {
})
}
}
+
+// TestReturnIndexSweptOnADormantConsole (fix round 2, item 3): the return
+// index sweep also runs from the console's hourly store sweep (the expiry
+// pass's trash sweep), whatever the self-serve mode: a switch-off console
+// with no recovery evidence, whose moves pass never runs, still clears an
+// entry whose request is past its deadline.
+func TestReturnIndexSweptOnADormantConsole(t *testing.T) {
+ ctx := context.Background()
+ 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)
+ }
+ w.fx.srv.selfServeScanAtStart(ctx)
+ if w.fx.srv.selfServeMode() {
+ t.Fatal("the console is in self-serve mode; the witness needs a dormant one")
+ }
+ w.fx.srv.expiryPass(ctx, time.Now())
+ if _, _, err := w.fx.rel.Get(ctx, moves.ReturnIndexKey("uk", offlineHost, returnFwd)); err == nil {
+ t.Fatal("the hourly sweep left an entry whose request is past its deadline")
+ }
+}
diff --git a/internal/console/server.go b/internal/console/server.go
index 9c13674a5..103c9b20a 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -891,8 +891,10 @@ 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 is the last return index key a sweep examined
+ // (returnSweep resumes after it); the moves pass and the hourly store
+ // sweep share it under returnSweepMu.
+ returnSweepMu sync.Mutex
returnSweepAfter string
selfServeLeaseMu sync.Mutex
selfServeLease *leaseToken