amadan

carlos / platform Public

offline return: the hourly index sweep never waits and lists one bounded page (fix round 3, item 3)

returnSweepHourly takes the shared sweep mutex with TryLock: busy means
the moves pass's sweep holds it, so the hour is skipped (logged once per
busy streak) instead of waiting past the 30 s deadline. It lists the index
one page at a time through release.PageLister (at most 256 keys, resuming
after the last page next hour, starting over at the end), so neither the
listing nor the examination outgrows the deadline; a store without a
bounded listing (test fakes) degrades to List as the repo's convention
says. The per-entry judgment (sweepReturnEntry) is shared with the pass's
sweep, which is unchanged.

Witness: TestReturnSweepHourlyIsBounded (the mutex held by another
goroutine: returns at once, deletes nothing; 300 entries: one page of 256
the first hour, the rest the next, no whole-index listing). RED: "the
hourly sweep waited for the moves sweep's mutex" and "listed the whole
index 2 times".
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude d70e2ad25e470f664ec6a319685c5cb39d657a90 parent a7a3e13
3 files changed, +213 −35
  • internal/console/offlinereturn.go +87 −35
  • internal/console/offlinereturn_test.go +121 −0
  • internal/console/server.go +5 −0
diff --git a/internal/console/offlinereturn.go b/internal/console/offlinereturn.go
index 6bc030897..100fa544f 100644
--- a/internal/console/offlinereturn.go
+++ b/internal/console/offlinereturn.go
@@ -263,8 +263,8 @@ 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.
+// returnSweepHourlyMax bounds the page the hourly store sweep lists and
+// examines, and returnSweepHourlyTimeout its time.
const (
returnSweepHourlyMax = 256
returnSweepHourlyTimeout = 30 * time.Second
@@ -273,17 +273,62 @@ const (
// 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.
+// expired entries are cleared here. Bounded (fix round 3): it never waits
+// for the moves pass's sweep (the shared mutex busy, it skips this hour,
+// logged once), and it lists one page of at most returnSweepHourlyMax keys
+// (release.PageLister), resuming after the last page next hour, so its 30
+// s deadline holds whatever the index's size. A store without a bounded
+// listing (test fakes only) degrades to List, still examining one page's
+// worth. 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) {
+ if !s.returnSweepMu.TryLock() {
+ if !s.returnSweepBusyLogged {
+ s.returnSweepBusyLogged = true
+ log.Printf("console: self-serve moves: the return index sweep is busy in the moves pass; the hourly sweep tries next hour")
+ }
+ return
+ }
+ defer s.returnSweepMu.Unlock()
+ s.returnSweepBusyLogged = false
ctx, cancel := context.WithTimeout(ctx, returnSweepHourlyTimeout)
defer cancel()
- s.returnSweep(ctx, now, returnSweepHourlyMax)
+ root := moves.ReturnIndexRoot()
+ var keys []string
+ var err error
+ if pl, ok := s.cfg.Releases.(release.PageLister); ok {
+ keys, err = pl.ListPage(ctx, root, s.returnSweepPageAfter, returnSweepHourlyMax)
+ } else {
+ keys, err = s.cfg.Releases.List(ctx, root)
+ sort.Strings(keys)
+ i := sort.SearchStrings(keys, s.returnSweepPageAfter)
+ for i < len(keys) && keys[i] <= s.returnSweepPageAfter {
+ i++
+ }
+ keys = keys[i:]
+ if len(keys) > returnSweepHourlyMax {
+ keys = keys[:returnSweepHourlyMax]
+ }
+ }
+ if err != nil {
+ log.Printf("console: self-serve moves: the return index: %v (swept next hour)", err)
+ return
+ }
+ for _, key := range keys {
+ if ctx.Err() != nil {
+ return // the cursor stays at the last key examined
+ }
+ s.sweepReturnEntry(ctx, key, now)
+ s.returnSweepPageAfter = key
+ }
+ if len(keys) < returnSweepHourlyMax {
+ s.returnSweepPageAfter = "" // the end of the index: the next hour starts over
+ }
}
-// returnSweep examines at most max index entries, resuming after the last
-// one any sweep examined (the cursor is shared under returnSweepMu).
+// returnSweep is the moves pass's sweep: at most max index entries,
+// resuming after the last one it examined (its cursor, under
+// returnSweepMu).
func (s *Server) returnSweep(ctx context.Context, now time.Time, max int) {
s.returnSweepMu.Lock()
defer s.returnSweepMu.Unlock()
@@ -303,35 +348,42 @@ func (s *Server) returnSweep(ctx context.Context, now time.Time, max int) {
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)
- 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
- }
- }
- // 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)
+ s.sweepReturnEntry(ctx, key, now)
+ }
+}
+
+// sweepReturnEntry judges one index entry and deletes it, if-match on the
+// version judged, when its request is past its deadline or it names no
+// request. An entry or request that cannot be read is left.
+func (s *Server) sweepReturnEntry(ctx context.Context, key string, now time.Time) {
+ raw, etag, err := s.cfg.Releases.Get(ctx, key)
+ if err != nil {
+ return
+ }
+ var e moves.ReturnIndexEntry
+ if json.Unmarshal(raw, &e) != nil {
+ return
+ }
+ rkey := moves.ReturnRequestKey(e.Host, e.ReturnOf, e.RequestKey)
+ if rkey == "" {
+ return
+ }
+ rraw, _, err := s.cfg.Releases.Get(ctx, rkey)
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ case err != nil:
+ return
+ default:
+ rq, err := moves.ParseReturnRequest(rraw, e.Host, e.ReturnOf, rkey)
+ if err != nil || now.Before(rq.Deadline()) {
+ return
}
}
+ // 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)
+ }
}
// readReturnReport is the return-ready report of the current request of
diff --git a/internal/console/offlinereturn_test.go b/internal/console/offlinereturn_test.go
index 34380b455..dc8b77388 100644
--- a/internal/console/offlinereturn_test.go
+++ b/internal/console/offlinereturn_test.go
@@ -6,11 +6,14 @@ import (
"errors"
"fmt"
"net/http"
+ "sort"
+ "sync"
"testing"
"time"
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/status"
)
@@ -670,3 +673,121 @@ func TestReturnIndexSweptOnADormantConsole(t *testing.T) {
t.Fatal("the hourly sweep left an entry whose request is past its deadline")
}
}
+
+// pageCountStore records how the return index root is listed: whole
+// (List) or by page (ListPage, with its limit).
+type pageCountStore struct {
+ release.Store
+ mu sync.Mutex
+ wholeLists int
+ pageLimits []int
+}
+
+func (p *pageCountStore) List(ctx context.Context, prefix string) ([]string, error) {
+ if prefix == moves.ReturnIndexRoot() {
+ p.mu.Lock()
+ p.wholeLists++
+ p.mu.Unlock()
+ }
+ return p.Store.List(ctx, prefix)
+}
+
+func (p *pageCountStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
+ if prefix == moves.ReturnIndexRoot() {
+ p.mu.Lock()
+ p.pageLimits = append(p.pageLimits, limit)
+ p.mu.Unlock()
+ }
+ if pl, ok := p.Store.(release.PageLister); ok {
+ return pl.ListPage(ctx, prefix, startAfter, limit)
+ }
+ // The fixture's store has no bounded listing: one over its List.
+ keys, err := p.Store.List(ctx, prefix)
+ if err != nil {
+ return nil, err
+ }
+ sort.Strings(keys)
+ var out []string
+ for _, k := range keys {
+ if k > startAfter && (limit <= 0 || len(out) < limit) {
+ out = append(out, k)
+ }
+ }
+ return out, nil
+}
+
+func (p *pageCountStore) DeleteIfMatch(ctx context.Context, key, etag string) error {
+ return p.Store.(release.ConditionalDeleter).DeleteIfMatch(ctx, key, etag)
+}
+
+// TestReturnSweepHourlyIsBounded (fix round 3, item 3): the hourly sweep
+// never waits for the moves pass's sweep (the shared mutex busy: it
+// returns at once, deleting nothing, and tries next hour), and it lists the
+// index one bounded page at a time (at most 256 keys), never the whole
+// index: of 300 entries one hour clears one page and the next hour the
+// rest.
+func TestReturnSweepHourlyIsBounded(t *testing.T) {
+ ctx := context.Background()
+ t.Run("the moves sweep holds the mutex", 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)
+ }
+ w.fx.srv.returnSweepMu.Lock()
+ done := make(chan struct{})
+ go func() { w.fx.srv.returnSweepHourly(ctx, time.Now()); close(done) }()
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ w.fx.srv.returnSweepMu.Unlock()
+ <-done
+ t.Fatal("the hourly sweep waited for the moves sweep's mutex")
+ }
+ w.fx.srv.returnSweepMu.Unlock()
+ if _, _, err := w.fx.rel.Get(ctx, moves.ReturnIndexKey("uk", offlineHost, returnFwd)); err != nil {
+ t.Fatalf("a skipped sweep deleted the entry: %v", err)
+ }
+ })
+ t.Run("more than one page", func(t *testing.T) {
+ w := newReturnWorld(t, nil)
+ ps := &pageCountStore{Store: w.fx.rel}
+ w.fx.srv.cfg.Releases = ps
+ // 300 entries naming no request: each one the sweep reaches goes.
+ for i := 0; i < 300; i++ {
+ if err := w.fx.srv.offlineReturnIndex(ctx, "uk", fmt.Sprintf("h%03d.shop.example.com", i), returnFwd, fmt.Sprintf("%d-c-aaaaaaaa", i+1)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ left := func() int {
+ keys, err := w.fx.rel.List(ctx, moves.ReturnIndexRoot())
+ if err != nil {
+ t.Fatal(err)
+ }
+ return len(keys)
+ }
+ w.fx.srv.returnSweepHourly(ctx, time.Now())
+ if n := left(); n != 300-256 {
+ t.Fatalf("one hour's sweep left %d entries, want %d", n, 300-256)
+ }
+ w.fx.srv.returnSweepHourly(ctx, time.Now())
+ if n := left(); n != 0 {
+ t.Fatalf("the next hour's sweep left %d entries", n)
+ }
+ ps.mu.Lock()
+ defer ps.mu.Unlock()
+ if ps.wholeLists != 0 {
+ t.Fatalf("the hourly sweep listed the whole index %d times", ps.wholeLists)
+ }
+ for _, l := range ps.pageLimits {
+ if l <= 0 || l > 256 {
+ t.Fatalf("the hourly sweep listed a page of limit %d", l)
+ }
+ }
+ if len(ps.pageLimits) != 2 {
+ t.Fatalf("%d pages listed over two hours, want two", len(ps.pageLimits))
+ }
+ })
+}
diff --git a/internal/console/server.go b/internal/console/server.go
index aaa787ceb..d9bc460ae 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -896,6 +896,11 @@ type Server struct {
// sweep share it under returnSweepMu.
returnSweepMu sync.Mutex
returnSweepAfter string
+ // returnSweepPageAfter is the hourly sweep's page cursor, and
+ // returnSweepBusyLogged its once-per-busy-streak log (fix round 3);
+ // both under returnSweepMu, or written by the hourly sweep alone.
+ returnSweepPageAfter string
+ returnSweepBusyLogged bool
// selfServeStartFound and selfServeStartErr are the start scan's
// answer (SelfServeStartEvidence), written once in NewServer.
selfServeStartFound bool