amadan

carlos / platform Public

console: bound each seeds pass by a deadline (Plan 8 Task 2 fix round 1)

seedsPassTimeout (15m, three intervals: two listings and a dozen
finishes at storeTeardownTimeout each) wraps every pass, so a hung S3
call ends the pass and the next tick resumes it; the loops stop at the
cut and log it once. Comments name the reservation bound's storeWalks
wait and the two-console clock caveat. Tests: a stall in the record walk,
the snapshot listing and a mid-pass delete each end the pass and the next
pass does the work; M6's rolled-back half deletes the snapshot while the
finish waits.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude cf39691b56801707eba3f6d0f4e73f92090079a5 parent 931b14c
2 files changed, +174 −3
  • internal/console/seedsweep.go +42 −3
  • internal/console/seedsweep_test.go +132 −0

CI passed — run details

diff --git a/internal/console/seedsweep.go b/internal/console/seedsweep.go
index cc78d4ffb..b93fa3d10 100644
--- a/internal/console/seedsweep.go
+++ b/internal/console/seedsweep.go
@@ -26,7 +26,12 @@ const (
// seedReserveTTL is how long a --from seed may stay reserving. The
// door's copy is bounded at about 52 minutes (store walk, a 30-minute
// restore, 60s, a 20-minute upload, the publish), so an hour is past
- // any live copy.
+ // any live copy. That bound leaves out the private-store path's wait
+ // for its store's walk lock (storeWalks), which has no deadline and
+ // queues behind holders of up to storeTeardownTimeout each, this
+ // sweep's own finishes among them. A copy that overruns the hour that
+ // way is aborted by a CAS with one winner: the user retries, and no
+ // data is lost.
seedReserveTTL = time.Hour
// seedImportTTL is how long an import may wait for its upload, or for
// the app to say its process is stopped.
@@ -36,6 +41,13 @@ const (
seedOrphanAge = time.Hour
)
+// seedsPassTimeout bounds one pass, so a hung list, read or delete cannot
+// stall the sweep for good (the S3 store sets no client timeout). It is
+// three intervals: enough for the two listings and a dozen finishes that
+// each run to storeTeardownTimeout (60s), the slowest step. A pass cut
+// short is resumed by the next tick. A variable so a test can shorten it.
+var seedsPassTimeout = 15 * time.Minute
+
// The tests' seams, nil in production. seedsPassBetweenListings runs after
// the snapshot listing and before the record listing.
// seedsPassBeforeAbort runs inside each abandonment's mutation, after the
@@ -75,8 +87,11 @@ func (s *Server) StartSeedsPass(ctx context.Context) {
// seedStale reports a seed stuck where no box moves it: a --from seed
// reserving for over seedReserveTTL, or an import waiting for its upload or
-// its stop for over seedImportTTL. Both times are the console's: At is
-// stamped by the door that reserved or began the seed.
+// its stop for over seedImportTTL. Both times are a console's: At is
+// stamped by the console whose door reserved or began the seed, now is the
+// sweeping console's. With one console per deployment they are one clock;
+// with more (region HA) their skew comes off the reservation's margin of
+// about 8 minutes, and an overrun costs a copy, never data.
func seedStale(seed *provision.SeedSource, now time.Time) bool {
switch {
case seed == nil:
@@ -110,10 +125,25 @@ func seedStale(seed *provision.SeedSource, now time.Time) bool {
//
// Every delete names one exact seeds/<host>/<id>/snapshot.db, so it never
// reaches another seed's.
+//
+// The pass is bounded by seedsPassTimeout. Cutting it anywhere is safe and
+// the next pass resumes: a listing cut short is an error, so nothing is
+// decided; an abandonment's CAS either landed (its snapshot then goes next
+// pass, as an aborted seed's) or did not (the seed is still stale next
+// pass); a snapshot not yet deleted is decided again from fresh listings;
+// a finish cut short leaves its record rolled-back, and every finish step
+// is idempotent.
func (s *Server) seedsPass(ctx context.Context, now time.Time) {
if s.cfg.HostClaim || s.cfg.Releases == nil || s.cfg.SeedData == nil {
return
}
+ ctx, cancel := context.WithTimeout(ctx, seedsPassTimeout)
+ defer cancel()
+ defer func() {
+ if ctx.Err() != nil {
+ log.Printf("console: seeds pass: cut short: %v (the next pass resumes)", ctx.Err())
+ }
+ }()
var snaps []release.KeyMod
ml, walk := s.cfg.SeedData.(release.ModLister)
if walk {
@@ -176,6 +206,9 @@ func (s *Server) seedsPass(ctx context.Context, now time.Time) {
// Step 4: the abandonments.
for _, rec := range recs {
+ if ctx.Err() != nil {
+ return
+ }
if !seedStale(rec.Seed, now) || held(rec.Host) {
continue
}
@@ -189,6 +222,9 @@ func (s *Server) seedsPass(ctx context.Context, now time.Time) {
// Step 5: the snapshot deletes.
sort.Strings(condemned)
for _, key := range condemned {
+ if ctx.Err() != nil {
+ return
+ }
if err := s.cfg.SeedData.Delete(ctx, key); err != nil && !errors.Is(err, os.ErrNotExist) {
log.Printf("console: seeds pass: delete %s: %v", key, err)
continue
@@ -198,6 +234,9 @@ func (s *Server) seedsPass(ctx context.Context, now time.Time) {
// Step 6: the finishes.
for _, rec := range recs {
+ if ctx.Err() != nil {
+ return
+ }
if rec.Seed == nil || rec.Seed.State != seeds.StateRolledBack || held(rec.Host) {
continue
}
diff --git a/internal/console/seedsweep_test.go b/internal/console/seedsweep_test.go
index 8d3ce250e..beda0770f 100644
--- a/internal/console/seedsweep_test.go
+++ b/internal/console/seedsweep_test.go
@@ -6,6 +6,7 @@ import (
"os"
"path/filepath"
"strings"
+ "sync/atomic"
"testing"
"time"
@@ -403,3 +404,134 @@ func TestSweepOffOnAHostClaimConsole(t *testing.T) {
t.Fatalf("a host-claim console swept: %s, orphan kept %v", got, f.snapshotExists(t, orphan))
}
}
+
+// hang blocks its first caller until ctx is done, as a stalled connection
+// does; every later call goes through.
+type hang struct{ done atomic.Bool }
+
+func (h *hang) once(ctx context.Context) error {
+ if h.done.Swap(true) {
+ return nil
+ }
+ <-ctx.Done()
+ return ctx.Err()
+}
+
+// hangingRecordWalk stalls the record walk's first listing.
+type hangingRecordWalk struct {
+ release.Store
+ h *hang
+}
+
+// DeleteIfMatch keeps the finish's conditional record delete reachable
+// through the wrapper.
+func (s hangingRecordWalk) DeleteIfMatch(ctx context.Context, key, etag string) error {
+ return release.DeleteIfMatch(ctx, s.Store, key, etag)
+}
+
+func (s hangingRecordWalk) List(ctx context.Context, prefix string) ([]string, error) {
+ if prefix == "accounts/" {
+ if err := s.h.once(ctx); err != nil {
+ return nil, err
+ }
+ }
+ return s.Store.List(ctx, prefix)
+}
+
+// hangingData stalls the data bucket's first snapshot listing (list) or
+// its first delete (!list).
+type hangingData struct {
+ release.Store
+ h *hang
+ list bool
+}
+
+func (s hangingData) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) {
+ if s.list {
+ if err := s.h.once(ctx); err != nil {
+ return nil, err
+ }
+ }
+ return s.Store.(release.ModLister).ListMod(ctx, prefix)
+}
+
+func (s hangingData) Delete(ctx context.Context, key string) error {
+ if !s.list {
+ if err := s.h.once(ctx); err != nil {
+ return err
+ }
+ }
+ return s.Store.Delete(ctx, key)
+}
+
+// A store call that never answers ends the pass at seedsPassTimeout, with
+// nothing decided on what it could not finish; the next pass does the
+// work. The delete case cuts the pass after its decisions, midway through
+// its deletes, before its finishes.
+func TestSweepPassIsBoundedByATimeout(t *testing.T) {
+ old := seedsPassTimeout
+ seedsPassTimeout = 100 * time.Millisecond
+ t.Cleanup(func() { seedsPassTimeout = old })
+ for name, stall := range map[string]func(f sweepFixture){
+ "record walk": func(f sweepFixture) { f.fx.srv.cfg.Releases = hangingRecordWalk{f.fx.rel, &hang{}} },
+ "snapshot listing": func(f sweepFixture) { f.fx.srv.cfg.SeedData = hangingData{f.data, &hang{}, true} },
+ "a snapshot delete": func(f sweepFixture) { f.fx.srv.cfg.SeedData = hangingData{f.data, &hang{}, false} },
+ } {
+ t.Run(name, func(t *testing.T) {
+ f := newSweepFixture(t)
+ now := time.Now()
+ f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour))
+ orphans := []string{
+ f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour),
+ f.snapshot(t, "right.shop.example.com", newSeedID(t), 3*time.Hour),
+ }
+ f.seeded(t, "done.shop.example.com", seeds.KindImport, seeds.StateRolledBack, now)
+ stall(f)
+ done := make(chan struct{})
+ go func() { f.pass(now); close(done) }()
+ select {
+ case <-done:
+ case <-time.After(10 * time.Second):
+ t.Fatal("a hung store call stalled the pass")
+ }
+ if got := f.state(t, "done.shop.example.com"); got != seeds.StateRolledBack {
+ t.Fatalf("a pass cut short finished a seed after its cut: %s", got)
+ }
+ f.pass(now)
+ if got := f.state(t, "stale.shop.example.com"); got != seeds.StateAborted {
+ t.Errorf("the next pass left the stale reservation %s", got)
+ }
+ if got := f.state(t, "done.shop.example.com"); got != "gone" {
+ t.Errorf("the next pass left the rolled-back record %s", got)
+ }
+ for _, key := range orphans {
+ if f.snapshotExists(t, key) {
+ t.Errorf("the next pass left the orphan %s", key)
+ }
+ }
+ })
+ }
+}
+
+// M6's rolled-back half on its own: a finish that cannot complete (a store
+// this seed made, and no backend to tear it down with) leaves the record
+// rolled-back, and the sweep still deletes the seed's snapshot.
+func TestSweepDeletesARolledBackSeedsSnapshotWhenItsFinishWaits(t *testing.T) {
+ f := finishFixture(t)
+ id := newSeedID(t)
+ f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
+ f.markedStore(t, id)
+ f.fx.srv.cfg.StoreProvisioner = nil
+ key := seeds.SnapshotKey(fromHost, id)
+ if _, err := f.data.Put(context.Background(), key, []byte("db"), ""); err != nil {
+ t.Fatal(err)
+ }
+ f.fx.srv.seedsPass(context.Background(), time.Now())
+ if _, _, err := f.data.Get(context.Background(), key); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the rolled-back seed's snapshot: %v, want deleted", err)
+ }
+ rec, err := f.record(t, fromHost)
+ if err != nil || rec.Seed.State != seeds.StateRolledBack {
+ t.Fatalf("the record after a finish that could not complete: %+v, %v; want it kept rolled-back", rec, err)
+ }
+}