amadan

carlos / platform Public

console: the run sweep reads past an unreadable record; only the orphan phase needs every one (final review M3)

ListAllRecords is all-or-nothing, so one unreadable account manifest or
record stopped expiry and lost-run closing for the whole fleet. The sweep
now reads with ListAllRecordsTolerant, which skips and logs what it cannot
read and says whether the read was complete, and runs the orphan phase only
on a complete read. ListAllRecords itself is unchanged.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 91a51f64b8c866955f29fd521bfa1722a738c6e2 parent 4228e7c
4 files changed, +114 −9
  • internal/console/runsweep.go +11 −2
  • internal/console/runsweep_test.go +32 −0
  • internal/provision/applisting_test.go +31 −0
  • internal/provision/reconcile.go +40 −7
diff --git a/internal/console/runsweep.go b/internal/console/runsweep.go
index 311d93b55..5ee0c011d 100644
--- a/internal/console/runsweep.go
+++ b/internal/console/runsweep.go
@@ -53,17 +53,26 @@ func (s *Server) runSweep(ctx context.Context, now time.Time) { s.runSweepPass(c
// a box picked in the meantime stays picked, a run a box finished stays
// finished. housekeeping adds the terminal-retention and orphan phases to the
// expiry and lost-run phases every pass runs. It is called from one goroutine.
+//
+// The records are read tolerantly: one account or record that will not read
+// must not stop expiry and lost-run closing for the rest of the fleet, and
+// each of those acts on a record it read. The orphan phase is the exception:
+// a run directory no record names is deleted, so it runs only on a read that
+// saw every record.
func (s *Server) runSweepPass(ctx context.Context, now time.Time, housekeeping bool) {
if s.cfg.Releases == nil {
return
}
ctx, cancel := context.WithTimeout(ctx, runSweepTimeout)
defer cancel()
- recs, err := provision.ListAllRecords(ctx, s.cfg.Releases)
+ recs, complete, err := provision.ListAllRecordsTolerant(ctx, s.cfg.Releases)
if err != nil {
log.Printf("console: run sweep: list records: %v", err)
return
}
+ if !complete {
+ log.Printf("console: run sweep: some records could not be read; their runs wait for a later pass, and no orphan run is deleted on this one")
+ }
named := map[string]bool{}
apps := map[[2]string]bool{}
for _, rec := range recs {
@@ -95,7 +104,7 @@ func (s *Server) runSweepPass(ctx context.Context, now time.Time, housekeeping b
s.sweepTerminalRun(ctx, c, r, now)
}
}
- if !housekeeping {
+ if !housekeeping || !complete {
return
}
for app := range apps {
diff --git a/internal/console/runsweep_test.go b/internal/console/runsweep_test.go
index 8a3364420..78c65a793 100644
--- a/internal/console/runsweep_test.go
+++ b/internal/console/runsweep_test.go
@@ -525,3 +525,35 @@ func TestRunSweepKeepsAnOrphanWhoseRowCouldNotBeMarked(t *testing.T) {
t.Fatalf("run-end rows %q", ends)
}
}
+
+// One record that will not read does not stop expiry for the rest of the
+// fleet, but it does stop the orphan phase: a run that record names could
+// look like an orphan.
+func TestRunSweepReadsPastAnUnreadableRecordButDeletesNoOrphan(t *testing.T) {
+ fx := runFixture(t)
+ ctx := context.Background()
+ putShopRecord(t, fx.consoleFixture, provision.Record{Host: "other.example.com", Channel: "stable", CreatedAt: time.Now().UTC()})
+ fx.startRun(t)
+ old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: "other.example.com", RequestedAt: time.Now().Add(-25 * time.Hour)}
+ fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old)
+ unreadable := provision.RecordKey(release.OpsAccount, "shop", "other.example.com")
+ broken := true
+ fx.srv.cfg.Releases = &faultStore{Store: fx.rel, get: func(key string) error {
+ if broken && key == unreadable {
+ return errors.New("slow down")
+ }
+ return nil
+ }}
+ fx.srv.runSweep(ctx, time.Now().Add(2*time.Minute))
+ if r := fx.record(t).ActiveRun; r.State != provision.RunExpired {
+ t.Fatalf("one unreadable record stopped the expiry: %+v", r)
+ }
+ if left := fx.left(old.ID); len(left) == 0 {
+ t.Fatal("an orphan was deleted on a pass that could not read every record")
+ }
+ broken = false
+ fx.srv.runSweep(ctx, time.Now().Add(2*time.Minute))
+ if left := fx.left(old.ID); len(left) != 0 {
+ t.Fatalf("the orphan survived a full read: %q", left)
+ }
+}
diff --git a/internal/provision/applisting_test.go b/internal/provision/applisting_test.go
index 71a61c3af..83e618a81 100644
--- a/internal/provision/applisting_test.go
+++ b/internal/provision/applisting_test.go
@@ -281,6 +281,37 @@ func TestListAllRecordsRefusesAPartialWalk(t *testing.T) {
}
}
+// ListAllRecordsTolerant reports what it could read and says it is
+// incomplete, where ListAllRecords refuses: an app it could not list, a
+// record it could not Get.
+func TestListAllRecordsTolerantSkipsWhatItCannotRead(t *testing.T) {
+ inner, sqid := setup(t)
+ first := acmeRecord(sqid)
+ first.Host, first.Slug = "one.example.com", "one"
+ second := acmeRecord(sqid)
+ second.Host, second.Slug = "two.example.com", "two"
+ for _, rec := range []Record{first, second} {
+ if err := CreateRecord(ctx, inner, rec); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if recs, complete, err := ListAllRecordsTolerant(ctx, inner); err != nil || !complete || len(recs) != 2 {
+ t.Fatalf("a clean read = %d records, complete %v, %v", len(recs), complete, err)
+ }
+ s := &countingGets{Store: inner, failKey: RecordKey(sqid, first.App, first.Host)}
+ recs, complete, err := ListAllRecordsTolerant(ctx, s)
+ if err != nil || complete || len(recs) != 1 || recs[0].Host != second.Host {
+ t.Fatalf("one record unreadable = %+v, complete %v, %v", recs, complete, err)
+ }
+ if _, err := ListAllRecords(ctx, s); err == nil {
+ t.Fatal("ListAllRecords read past an unreadable record")
+ }
+ walk := &bucketShaped{Store: inner, failList: RecordsPrefix(sqid, first.App)}
+ if recs, complete, err := ListAllRecordsTolerant(ctx, walk); err != nil || complete || len(recs) != 0 {
+ t.Fatalf("an app that will not list = %+v, complete %v, %v", recs, complete, err)
+ }
+}
+
// The sidecar reader over the walk mints the sidecar it mints over the full
// listing, and keeps it on the next pass. Returning no sidecar keys at all
// would otherwise pass every other test here.
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go
index 125691e5b..61d394e17 100644
--- a/internal/provision/reconcile.go
+++ b/internal/provision/reconcile.go
@@ -1740,21 +1740,50 @@ func trashGate(ctx context.Context, s release.Store, what string) func(acct, app
// skipped and logged (Reconcile's `protect` treatment, minus the registry
// bookkeeping). One malformed record must not blind a caller to every other.
func ListAllRecords(ctx context.Context, s release.Store) ([]Record, error) {
+ recs, _, err := readAllRecords(ctx, s, true)
+ return recs, err
+}
+
+// ListAllRecordsTolerant is ListAllRecords for a caller that acts on each
+// record it reads, one at a time, and draws nothing from a record's absence
+// (the run sweep's expiry and lost-run phases). An account manifest that
+// will not read cleanly, an app whose records will not list, or a record
+// that will not Get is logged and skipped, and complete comes back false;
+// every other record is returned. err is only the top-level listing failing,
+// when nothing could be read. A caller that also draws conclusions from
+// absence must act on that only when complete is true.
+func ListAllRecordsTolerant(ctx context.Context, s release.Store) (recs []Record, complete bool, err error) {
+ return readAllRecords(ctx, s, false)
+}
+
+// readAllRecords is both: strict fails on any read failure, as
+// ListAllRecords documents; tolerant logs it, skips what it could not read
+// and reports complete false.
+func readAllRecords(ctx context.Context, s release.Store, strict bool) (recs []Record, complete bool, err error) {
+ complete = true
known, readFailed, err := knownAccounts(ctx, s)
if err != nil {
- return nil, err
+ return nil, false, err
}
if readFailed {
- return nil, errors.New("provision: an account manifest could not be read cleanly — refusing to report a partial record set")
+ if strict {
+ return nil, false, errors.New("provision: an account manifest could not be read cleanly — refusing to report a partial record set")
+ }
+ log.Printf("provision: an account manifest could not be read cleanly: that account's records are skipped")
+ complete = false
}
// Sorted (listAppSubtree sorts) so the result — and anything folded
// from it — is stable call to call, whatever order the store lists in.
keys, partial, err := listAppSubtree(ctx, s, known, "instances/")
if err != nil {
- return nil, fmt.Errorf("provision: list apps: %w", err)
+ return nil, false, fmt.Errorf("provision: list apps: %w", err)
}
if partial {
- return nil, errors.New("provision: an app's records could not be listed — refusing to report a partial record set")
+ if strict {
+ return nil, false, errors.New("provision: an app's records could not be listed — refusing to report a partial record set")
+ }
+ log.Printf("provision: an app's records could not be listed: they are skipped")
+ complete = false
}
var recordKeys []string
for _, key := range keys {
@@ -1762,14 +1791,18 @@ func ListAllRecords(ctx context.Context, s release.Store) ([]Record, error) {
recordKeys = append(recordKeys, key)
}
}
- var recs []Record
for _, part := range chunks(recordKeys) {
bodies := getEach(ctx, s, part)
for _, key := range part {
acct, app, host, _ := ParseRecordKey(key, known)
raw, err := bodies[key].body, bodies[key].err
if err != nil {
- return nil, fmt.Errorf("provision: read record %s: %w", key, err)
+ if strict {
+ return nil, false, fmt.Errorf("provision: read record %s: %w", key, err)
+ }
+ log.Printf("provision: skipping record %s: unreadable: %v", key, err)
+ complete = false
+ continue
}
rec, err := DecodeRecord(raw)
if err != nil {
@@ -1784,7 +1817,7 @@ func ListAllRecords(ctx context.Context, s release.Store) ([]Record, error) {
recs = append(recs, rec)
}
}
- return recs, nil
+ return recs, complete, nil
}
// listAppSubtree returns the keys a record reader needs to see: every key