amadan
console: run sweep reads the record view; orphan housekeeping narrows to recent runs; seed sweep skips the walk with nothing to delete (#336)
CARLOS_CHANGE_INDEX=read arms a record view over the console's deployment store (not on a host-claim console, whose sweeps are off), rebuilt at start and every backstop (+-10%), each rebuild bounded at 5 minutes. The run sweep's tolerant record read and, when the snapshot listing is empty, the seed sweep's strict read take the view's snapshot only when Refresh returned nil and the snapshot is Complete; otherwise the bucket, as before. Every write, the orphan phase's runs/ listing, and the seed sweep's walk with any snapshot listed stay on the bucket. Armed, orphan housekeeping lists runs/ only for apps with records and a run-start row in the last 48 hours (Store.AppsWithActionSince), with a full sweep on the first housekeeping and then daily. Disarmed is unchanged.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
10 files changed,
+594
−8
cmd/console/changeindex_test.go+67 −0cmd/console/main.go+53 −0internal/console/recordview.go+70 −0internal/console/runsweep.go+45 −2internal/console/runsweep_test.go+194 −0internal/console/seedsweep.go+18 −6internal/console/seedsweep_test.go+64 −0internal/console/server.go+13 −0internal/console/store.go+27 −0internal/console/store_test.go+43 −0
diff --git a/cmd/console/changeindex_test.go b/cmd/console/changeindex_test.go| new file mode 100644 |
| index 000000000..7545d50f3 |
| --- /dev/null |
| +++ b/cmd/console/changeindex_test.go |
| @@ -0,0 +1,67 @@ |
| +package main |
| + |
| +import ( |
| + "context" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// Only CARLOS_CHANGE_INDEX=read arms the console's record view, and never on |
| +// a host-claim console, whose sweeps are off. |
| +func TestChangeIndexArming(t *testing.T) { |
| + t.Parallel() |
| + ds := release.NewDirStore(t.TempDir()) |
| + ds.EnableChangeIndex() |
| + env := func(v string) func(string) string { |
| + return func(k string) string { |
| + if k == "CARLOS_CHANGE_INDEX" { |
| + return v |
| + } |
| + return "" |
| + } |
| + } |
| + for _, c := range []struct { |
| + val string |
| + hostClaim bool |
| + armed bool |
| + }{ |
| + {"", false, false}, |
| + {"READ", false, false}, |
| + {"1", false, false}, |
| + {"read", true, false}, |
| + {"read", false, true}, |
| + } { |
| + if got := armChangeIndex(env(c.val), c.hostClaim, ds) != nil; got != c.armed { |
| + t.Errorf("CARLOS_CHANGE_INDEX=%q host claim %v: armed %v, want %v", c.val, c.hostClaim, got, c.armed) |
| + } |
| + } |
| +} |
| + |
| +// The backstop rebuilds the view as it starts, and stops with its context. |
| +func TestRecordViewBackstopRebuildsAtStart(t *testing.T) { |
| + t.Parallel() |
| + ds := release.NewDirStore(t.TempDir()) |
| + ds.EnableChangeIndex() |
| + v := armChangeIndex(func(string) string { return "read" }, false, ds) |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + done := make(chan struct{}) |
| + go func() { recordViewBackstopLoop(ctx, v); close(done) }() |
| + deadline := time.Now().Add(10 * time.Second) |
| + for v.Snapshot() == nil { |
| + if time.Now().After(deadline) { |
| + t.Fatal("the backstop published nothing at start") |
| + } |
| + time.Sleep(10 * time.Millisecond) |
| + } |
| + if !v.Snapshot().Complete { |
| + t.Fatal("the start rebuild of an empty bucket is not complete") |
| + } |
| + cancel() |
| + select { |
| + case <-done: |
| + case <-time.After(10 * time.Second): |
| + t.Fatal("the backstop did not stop with its context") |
| + } |
| +} |
diff --git a/cmd/console/main.go b/cmd/console/main.go| index 5dc62d8b2..a0151bbed 100644 |
| --- a/cmd/console/main.go |
| +++ b/cmd/console/main.go |
| @@ -90,6 +90,7 @@ import ( |
| "flag" |
| "fmt" |
| "log" |
| + "math/rand/v2" |
| "net" |
| "net/http" |
| "os" |
| @@ -484,6 +485,15 @@ func serve() { |
| if os.Getenv("CARLOS_MAIL_LEGACY_SWEEP") == "1" { |
| go sweepLegacyMail(cfg.HostClaim, rel, st) |
| } |
| + // The S3 change index's reader (CARLOS_CHANGE_INDEX=read, change-index |
| + // design §4.6): the run and seed sweeps read their records through a |
| + // view over the deployment store, rebuilt at start and every backstop. |
| + // Armed before any worker starts. Unset is nil: every sweep walks the |
| + // bucket as before. |
| + if v := armChangeIndex(os.Getenv, cfg.HostClaim, rel); v != nil { |
| + srv.SetRecordView(v) |
| + go recordViewBackstopLoop(context.Background(), v) |
| + } |
| // The move pass's pages (moves Plan 5 Task 7): the console's alert sinks, |
| // armed before the claim runtime binds the pass, so no pass runs unarmed. |
| armMovesPaging(context.Background(), cfg, os.Getenv) |
| @@ -1447,6 +1457,49 @@ func dbDoorWiring() (bucket, region, prefix, placement string) { |
| return cfg.Bucket, cfg.Region, cfg.Prefix, placement |
| } |
| +// recordViewWalkTimeout bounds one backstop rebuild, so a stuck index poll |
| +// or listing never holds the record view (and with it the sweeps' refresh) |
| +// for good. It is the run sweep's pass budget, which has always held one |
| +// full record walk. |
| +const recordViewWalkTimeout = 5 * time.Minute |
| + |
| +// armChangeIndex returns the console's record view over rel when |
| +// CARLOS_CHANGE_INDEX is exactly "read" (change-index design §7), or nil. A |
| +// host-claim console stays disarmed: the run and seed sweeps, the view's |
| +// only readers, are off there, so a view would only cost its backstop walks. |
| +func armChangeIndex(getenv func(string) string, hostClaim bool, rel release.Store) *provision.RecordView { |
| + if getenv("CARLOS_CHANGE_INDEX") != "read" { |
| + return nil |
| + } |
| + if hostClaim { |
| + log.Printf("console: change index: CARLOS_CHANGE_INDEX=read is ignored on a host-claim console, whose sweeps are off") |
| + return nil |
| + } |
| + v := provision.NewRecordView(rel, release.NewChangeWatch(rel)) |
| + log.Printf("console: change index armed: the run and seed sweeps read the record view (backstop %s)", release.ChangeIndexBackstop) |
| + return v |
| +} |
| + |
| +// recordViewBackstopLoop is the view's backstop (design §4.4): Rebuild at |
| +// start and every release.ChangeIndexBackstop, jittered by +-10%, each |
| +// bounded by recordViewWalkTimeout. A failed rebuild leaves the view unable |
| +// to vouch, so the sweeps walk the bucket until one succeeds. |
| +func recordViewBackstopLoop(ctx context.Context, v *provision.RecordView) { |
| + for { |
| + rctx, cancel := context.WithTimeout(ctx, recordViewWalkTimeout) |
| + if err := v.Rebuild(rctx); err != nil { |
| + log.Printf("console: record view backstop: %v", err) |
| + } |
| + cancel() |
| + b := release.ChangeIndexBackstop |
| + select { |
| + case <-ctx.Done(): |
| + return |
| + case <-time.After(b - b/10 + rand.N(b/5)): |
| + } |
| + } |
| +} |
| + |
| // applyMovesSwitch sets console.Config.MovesEnabled from the Phase 1 moves |
| // switch, moves.SwitchEnv. Off (the default) leaves the console answering |
| // exactly as main does today. |
diff --git a/internal/console/recordview.go b/internal/console/recordview.go| new file mode 100644 |
| index 000000000..75708aa26 |
| --- /dev/null |
| +++ b/internal/console/recordview.go |
| @@ -0,0 +1,70 @@ |
| +package console |
| + |
| +// The console's reads through the S3 change index (S3 change-index design |
| +// §4.6, discussion #336). Armed only by CARLOS_CHANGE_INDEX=read, which |
| +// cmd/console turns into a record view handed to SetRecordView; unarmed, |
| +// recordView stays nil and every sweep walks the bucket as before. |
| + |
| +import ( |
| + "context" |
| + "log" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// SetRecordView arms the run and seed sweeps' record reads with v. serve |
| +// calls it once, before it starts any worker; nil leaves them disarmed. |
| +func (s *Server) SetRecordView(v *provision.RecordView) { s.recordView = v } |
| + |
| +// The viewUse states; zero is none yet. |
| +const ( |
| + viewUseView int32 = 1 + iota |
| + viewUseWalk |
| +) |
| + |
| +// recordWalkStore is the store a sweep's record walk reads: the armed view's |
| +// snapshot over live, refreshed first, when the refresh returned nil and the |
| +// snapshot is Complete. Otherwise ok is false and the store is live, so the |
| +// sweep walks the bucket as it did before the index (spec §3.4). |
| +// |
| +// Only record walks take it (provision.ListAllRecords*). The snapshot's |
| +// ListDirs of apps/<a>/ names only apps with record-family state, so it must |
| +// never reach an app enumerator, and every write and every non-record read |
| +// (runs/, seeds/) stays on live. ctx bounds the refresh, so a stuck index |
| +// poll or walk cannot hold the view, or the sweep, for good. |
| +func (s *Server) recordWalkStore(ctx context.Context, live release.Store) (release.Store, bool) { |
| + v := s.recordView |
| + if v == nil || live == nil { |
| + return live, false |
| + } |
| + err := v.Refresh(ctx) |
| + sn := v.Snapshot() |
| + switch { |
| + case err != nil: |
| + s.noteViewUse(false, "the record view's refresh failed: "+err.Error()) |
| + return live, false |
| + case sn == nil || !sn.Complete: |
| + s.noteViewUse(false, "the record view's snapshot is incomplete") |
| + return live, false |
| + } |
| + s.noteViewUse(true, "") |
| + return sn.Store(live), true |
| +} |
| + |
| +// noteViewUse logs a switch between reading the view and walking the bucket, |
| +// once per change rather than on every pass. |
| +func (s *Server) noteViewUse(ok bool, why string) { |
| + state := viewUseWalk |
| + if ok { |
| + state = viewUseView |
| + } |
| + if s.viewUse.Swap(state) == state { |
| + return |
| + } |
| + if ok { |
| + log.Printf("console: record view: the sweeps' record walks read the view") |
| + return |
| + } |
| + log.Printf("console: record view: %s; the sweeps' record walks read the bucket until it can vouch again", why) |
| +} |
diff --git a/internal/console/runsweep.go b/internal/console/runsweep.go| index 5ee0c011d..2e1bd4c4c 100644 |
| --- a/internal/console/runsweep.go |
| +++ b/internal/console/runsweep.go |
| @@ -19,6 +19,14 @@ const ( |
| // pass, because a caller waits on them; retention and orphans have 24 |
| // hour thresholds, so they run on every 20th pass (about 10 minutes). |
| runSweepHousekeepingEvery = 20 |
| + // orphanCandidateWindow: with the record view armed, orphan |
| + // housekeeping lists runs/ only for apps with a run-start row this |
| + // recent, the window in which a run's objects can still be live or |
| + // awaiting cleanup (S3 change-index design §4.6, codex R2-1). |
| + orphanCandidateWindow = 48 * time.Hour |
| + // orphanFullSweepEvery: and once this often it lists every app with |
| + // records, as it always did, as the backstop. |
| + orphanFullSweepEvery = 24 * time.Hour |
| ) |
| // runSweepTimeout bounds one pass, so a hung list or read cannot stall the |
| @@ -59,13 +67,17 @@ func (s *Server) runSweep(ctx context.Context, now time.Time) { s.runSweepPass(c |
| // 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. |
| +// |
| +// With the record view armed the records are read from it (recordWalkStore); |
| +// every write, and the orphan phase's listing of runs/, stays on the bucket. |
| 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, complete, err := provision.ListAllRecordsTolerant(ctx, s.cfg.Releases) |
| + src, _ := s.recordWalkStore(ctx, s.cfg.Releases) |
| + recs, complete, err := provision.ListAllRecordsTolerant(ctx, src) |
| if err != nil { |
| log.Printf("console: run sweep: list records: %v", err) |
| return |
| @@ -107,13 +119,44 @@ func (s *Server) runSweepPass(ctx context.Context, now time.Time, housekeeping b |
| if !housekeeping || !complete { |
| return |
| } |
| - for app := range apps { |
| + candidates, full := s.orphanCandidates(apps, now) |
| + for app := range candidates { |
| if ctx.Err() != nil { |
| log.Printf("console: run sweep: pass cut short: %v", ctx.Err()) |
| return |
| } |
| s.sweepOrphanRuns(ctx, app[0], app[1], named, now) |
| } |
| + if full && s.recordView != nil { |
| + s.lastOrphanFull = now |
| + } |
| +} |
| + |
| +// orphanCandidates is the apps whose runs/ the orphan phase lists, out of |
| +// apps, every app with records, and whether that is all of them. Disarmed it |
| +// is all of them, as before the index. Armed (S3 change-index design §4.6, |
| +// codex R2-1) it is only those with a run-start row in the last |
| +// orphanCandidateWindow, except on the first housekeeping after start and |
| +// then once every orphanFullSweepEvery, which take every app, as does a |
| +// failed read of the rows. Orphans are still found by age, not by index |
| +// movement: runs/ is outside every scope. The narrowed set is always a |
| +// subset of apps, so it never lists an app the full sweep would not. |
| +func (s *Server) orphanCandidates(apps map[[2]string]bool, now time.Time) (map[[2]string]bool, bool) { |
| + if s.recordView == nil || s.lastOrphanFull.IsZero() || now.Sub(s.lastOrphanFull) >= orphanFullSweepEvery { |
| + return apps, true |
| + } |
| + recent, err := s.store.AppsWithActionSince("run-start", now.Add(-orphanCandidateWindow)) |
| + if err != nil { |
| + log.Printf("console: run sweep: recent runs: %v (every app's runs are swept this pass)", err) |
| + return apps, true |
| + } |
| + out := map[[2]string]bool{} |
| + for _, r := range recent { |
| + if app := [2]string{r.Sqid, r.App}; apps[app] { |
| + out[app] = true |
| + } |
| + } |
| + return out, false |
| } |
| // sweepExpireQueued expires a queued run past its pickup deadline. The |
diff --git a/internal/console/runsweep_test.go b/internal/console/runsweep_test.go| index 78c65a793..986c00293 100644 |
| --- a/internal/console/runsweep_test.go |
| +++ b/internal/console/runsweep_test.go |
| @@ -5,7 +5,10 @@ import ( |
| "context" |
| "encoding/json" |
| "errors" |
| + "reflect" |
| + "sort" |
| "strings" |
| + "sync" |
| "testing" |
| "time" |
| @@ -557,3 +560,194 @@ func TestRunSweepReadsPastAnUnreadableRecordButDeletesNoOrphan(t *testing.T) { |
| t.Fatalf("the orphan survived a full read: %q", left) |
| } |
| } |
| + |
| +// listCountingStore is a DirStore with the change index on that records every |
| +// listing made through it, so a test can say which prefixes a pass walked. |
| +// Its embedded DirStore keeps every optional extension (conditional reads, |
| +// ETag listings, ModLister) a record view and the sweeps use. |
| +type listCountingStore struct { |
| + *release.DirStore |
| + mu sync.Mutex |
| + lists []string // "op prefix", in call order |
| +} |
| + |
| +func (c *listCountingStore) note(op, prefix string) { |
| + c.mu.Lock() |
| + c.lists = append(c.lists, op+" "+prefix) |
| + c.mu.Unlock() |
| +} |
| + |
| +// taken returns the listings made since the last call and forgets them. |
| +func (c *listCountingStore) taken() []string { |
| + c.mu.Lock() |
| + defer c.mu.Unlock() |
| + out := c.lists |
| + c.lists = nil |
| + return out |
| +} |
| + |
| +func (c *listCountingStore) List(ctx context.Context, prefix string) ([]string, error) { |
| + c.note("List", prefix) |
| + return c.DirStore.List(ctx, prefix) |
| +} |
| + |
| +func (c *listCountingStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) { |
| + c.note("ListPage", prefix) |
| + return c.DirStore.ListPage(ctx, prefix, startAfter, limit) |
| +} |
| + |
| +func (c *listCountingStore) ListDirs(ctx context.Context, prefix string) ([]string, error) { |
| + c.note("ListDirs", prefix) |
| + return c.DirStore.ListDirs(ctx, prefix) |
| +} |
| + |
| +func (c *listCountingStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) { |
| + c.note("ListETags", prefix) |
| + return c.DirStore.ListETags(ctx, prefix) |
| +} |
| + |
| +// indexedDeployment swaps fx's deployment store for a listCountingStore over the |
| +// same bucket, with the change index on, as the console's opener builds it. |
| +func indexedDeployment(t *testing.T, fx *consoleFixture) *listCountingStore { |
| + t.Helper() |
| + d, ok := fx.rel.(*deletableStore) |
| + if !ok { |
| + t.Fatalf("fixture store is %T", fx.rel) |
| + } |
| + ds := release.NewDirStore(d.root) |
| + ds.EnableChangeIndex() |
| + cs := &listCountingStore{DirStore: ds} |
| + fx.rel, fx.srv.cfg.Releases = cs, cs |
| + return cs |
| +} |
| + |
| +// armRecordView arms srv's record view over live, as cmd/console does with |
| +// CARLOS_CHANGE_INDEX=read, and runs the backstop's first rebuild. |
| +func armRecordView(t *testing.T, srv *Server, live release.Store) *provision.RecordView { |
| + t.Helper() |
| + v := provision.NewRecordView(live, release.NewChangeWatch(live)) |
| + if err := v.Rebuild(context.Background()); err != nil { |
| + t.Fatal(err) |
| + } |
| + srv.recordView = v |
| + return v |
| +} |
| + |
| +// runsListings keeps the listings under a runs/ prefix. |
| +func runsListings(lists []string) []string { |
| + var out []string |
| + for _, l := range lists { |
| + if strings.Contains(l, "/runs/") { |
| + out = append(out, l) |
| + } |
| + } |
| + sort.Strings(out) |
| + return out |
| +} |
| + |
| +// Armed, the run sweep reads its records from the record view: once the |
| +// view has caught up, a pass with nothing written lists nothing at all. |
| +// Disarmed, the same pass walks the bucket as before the index. |
| +func TestRunSweepReadsViewWhenArmed(t *testing.T) { |
| + t.Parallel() |
| + fx := runFixture(t) |
| + cs := indexedDeployment(t, fx.consoleFixture) |
| + ctx := context.Background() |
| + id := fx.startRun(t) |
| + |
| + cs.taken() |
| + fx.srv.runSweepPass(ctx, time.Now(), false) |
| + if lists := cs.taken(); len(lists) == 0 { |
| + t.Fatal("a disarmed pass listed nothing: the counter does not see the walk") |
| + } |
| + |
| + armRecordView(t, fx.srv, cs) |
| + // The expiry acts on the run the view holds, and its write bumps the |
| + // app's scope, which the next pass's refresh re-reads. |
| + fx.srv.runSweepPass(ctx, time.Now().Add(2*time.Minute), false) |
| + if r := fx.record(t).ActiveRun; r == nil || r.ID != id || r.State != provision.RunExpired { |
| + t.Fatalf("armed pass did not expire the run it read from the view: %+v", r) |
| + } |
| + fx.srv.runSweepPass(ctx, time.Now().Add(3*time.Minute), false) |
| + cs.taken() |
| + for i := 0; i < 2; i++ { |
| + fx.srv.runSweepPass(ctx, time.Now().Add(4*time.Minute), false) |
| + } |
| + if lists := cs.taken(); len(lists) != 0 { |
| + t.Fatalf("armed passes with nothing written listed the bucket: %q", lists) |
| + } |
| +} |
| + |
| +// putAppRecord writes an instance record of app in the operator account. |
| +func putAppRecord(t *testing.T, s release.Store, app, host string) { |
| + t.Helper() |
| + rec := provision.Record{Schema: 1, Host: host, Slug: strings.Split(host, ".")[0], App: app, Account: release.OpsAccount, |
| + Channel: "stable", CreatedAt: time.Now().UTC()} |
| + if err := provision.CreateRecord(context.Background(), s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// Between daily full sweeps, orphan housekeeping lists runs/ only for apps |
| +// with a run-start row in the last 48 hours (codex R2-1): shop ran an hour |
| +// ago, cafe never did, so only shop's runs/ is listed. |
| +func TestOrphanHousekeepingListsOnlyRecentRunApps(t *testing.T) { |
| + t.Parallel() |
| + fx := runFixture(t) |
| + cs := indexedDeployment(t, fx.consoleFixture) |
| + putAppRecord(t, cs, "cafe", "cafe.example.com") |
| + now := time.Now() |
| + if err := fx.st.RecordActivity(OpsAccountID, "shop", "alice@keymail.dev", Activity{Action: "run-start"}, now.Add(-time.Hour)); err != nil { |
| + t.Fatal(err) |
| + } |
| + armRecordView(t, fx.srv, cs) |
| + fx.srv.lastOrphanFull = now.Add(-time.Hour) |
| + cs.taken() |
| + fx.srv.runSweepPass(context.Background(), now, true) |
| + want := []string{"ListDirs " + runs.RunsPrefix(release.OpsAccount, "shop")} |
| + if got := runsListings(cs.taken()); !reflect.DeepEqual(got, want) { |
| + t.Fatalf("runs/ listings %q, want %q", got, want) |
| + } |
| + if !fx.srv.lastOrphanFull.Equal(now.Add(-time.Hour)) { |
| + t.Fatalf("a narrowed sweep moved the full sweep's clock to %s", fx.srv.lastOrphanFull) |
| + } |
| +} |
| + |
| +// The first housekeeping after start, and then one a day, sweeps every app |
| +// with records, whatever the activity rows say. |
| +func TestOrphanHousekeepingFullSweepDaily(t *testing.T) { |
| + t.Parallel() |
| + fx := runFixture(t) |
| + cs := indexedDeployment(t, fx.consoleFixture) |
| + putAppRecord(t, cs, "cafe", "cafe.example.com") |
| + armRecordView(t, fx.srv, cs) |
| + ctx := context.Background() |
| + all := []string{ |
| + "ListDirs " + runs.RunsPrefix(release.OpsAccount, "cafe"), |
| + "ListDirs " + runs.RunsPrefix(release.OpsAccount, "shop"), |
| + } |
| + now := time.Now() |
| + |
| + cs.taken() |
| + fx.srv.runSweepPass(ctx, now, true) |
| + if got := runsListings(cs.taken()); !reflect.DeepEqual(got, all) { |
| + t.Fatalf("first housekeeping listed %q, want every app with records %q", got, all) |
| + } |
| + if !fx.srv.lastOrphanFull.Equal(now) { |
| + t.Fatalf("lastOrphanFull = %s after a full sweep at %s", fx.srv.lastOrphanFull, now) |
| + } |
| + |
| + fx.srv.runSweepPass(ctx, now.Add(10*time.Minute), true) |
| + if got := runsListings(cs.taken()); len(got) != 0 { |
| + t.Fatalf("housekeeping with no recent run listed %q", got) |
| + } |
| + |
| + later := now.Add(orphanFullSweepEvery + time.Minute) |
| + fx.srv.runSweepPass(ctx, later, true) |
| + if got := runsListings(cs.taken()); !reflect.DeepEqual(got, all) { |
| + t.Fatalf("a day later housekeeping listed %q, want %q", got, all) |
| + } |
| + if !fx.srv.lastOrphanFull.Equal(later) { |
| + t.Fatalf("lastOrphanFull = %s, want %s", fx.srv.lastOrphanFull, later) |
| + } |
| +} |
diff --git a/internal/console/seedsweep.go b/internal/console/seedsweep.go| index b93fa3d10..bae364b88 100644 |
| --- a/internal/console/seedsweep.go |
| +++ b/internal/console/seedsweep.go |
| @@ -111,11 +111,13 @@ func seedStale(seed *provision.SeedSource, now time.Time) bool { |
| // step 7, §5.1), so a snapshot in this listing whose seed is live has |
| // its record in the next one. A seed begun after this listing has no |
| // snapshot in it, so nothing below can reach it. |
| -// 2. Every record, strictly. Any listing error, here or in step 1, ends |
| -// the pass: nothing is decided on a partial set. A host with a record |
| -// that would not decode is unreadable; a host two records hold is |
| -// ambiguous. Nothing of either is touched: no snapshot deleted, no seed |
| -// abandoned or finished. |
| +// 2. Every record, strictly: from the bucket, or, with the record view |
| +// armed and step 1 empty (so nothing can be deleted), from the view. |
| +// Any listing error, here or in step 1, ends the pass: nothing is |
| +// decided on a partial set. A host with a record that would not |
| +// decode is unreadable; a host two records hold is ambiguous. Nothing |
| +// of either is touched: no snapshot deleted, no seed abandoned or |
| +// finished. |
| // 3. The decisions, one per snapshot, from the two listings alone. |
| // 4. Each stale seed abandoned by a CAS from the exact state read |
| // (provision.SeedAbortStale); the box rolls back its part. |
| @@ -156,7 +158,17 @@ func (s *Server) seedsPass(ctx context.Context, now time.Time) { |
| if seedsPassBetweenListings != nil { |
| seedsPassBetweenListings() |
| } |
| - recs, unreadable, err := provision.ListAllRecordsStrict(ctx, s.cfg.Releases) |
| + // Step 2's source. A snapshot listing with nothing in it gives the pass |
| + // nothing to delete, so with the record view armed and able to vouch the |
| + // records come from it (S3 change-index design §4.6): abandonments and |
| + // finishes each act by a CAS on, or a re-read of, the bucket's record. |
| + // With any snapshot listed the walk is the bucket's, strict, after the |
| + // snapshot listing, because a delete is proved by a record's absence. |
| + src := s.cfg.Releases |
| + if len(snaps) == 0 { |
| + src, _ = s.recordWalkStore(ctx, s.cfg.Releases) |
| + } |
| + recs, unreadable, err := provision.ListAllRecordsStrict(ctx, src) |
| if err != nil { |
| log.Printf("console: seeds pass: list records: %v (nothing swept this pass)", err) |
| return |
diff --git a/internal/console/seedsweep_test.go b/internal/console/seedsweep_test.go| index beda0770f..4a7eb450c 100644 |
| --- a/internal/console/seedsweep_test.go |
| +++ b/internal/console/seedsweep_test.go |
| @@ -535,3 +535,67 @@ func TestSweepDeletesARolledBackSeedsSnapshotWhenItsFinishWaits(t *testing.T) { |
| t.Fatalf("the record after a finish that could not complete: %+v, %v; want it kept rolled-back", rec, err) |
| } |
| } |
| + |
| +// recordWalks keeps the listings a record walk makes (accounts/ and apps/). |
| +func recordWalks(lists []string) []string { |
| + var out []string |
| + for _, l := range lists { |
| + _, prefix, _ := strings.Cut(l, " ") |
| + if strings.HasPrefix(prefix, "accounts/") || strings.HasPrefix(prefix, "apps/") { |
| + out = append(out, l) |
| + } |
| + } |
| + return out |
| +} |
| + |
| +// Armed, a pass whose snapshot listing is empty has nothing to delete, so |
| +// it makes no strict walk of the bucket: it reads the records from the |
| +// record view, and a stale seed is still abandoned from what the view holds. |
| +func TestSeedSweepSkipsWalkWithoutSnapshots(t *testing.T) { |
| + t.Parallel() |
| + f := newSweepFixture(t) |
| + cs := indexedDeployment(t, f.fx) |
| + now := time.Now() |
| + f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour)) |
| + armRecordView(t, f.fx.srv, cs) |
| + cs.taken() |
| + f.pass(now) |
| + if walks := recordWalks(cs.taken()); len(walks) != 0 { |
| + t.Fatalf("a pass with no snapshot walked the records: %q", walks) |
| + } |
| + if got := f.state(t, "stale.shop.example.com"); got != seeds.StateAborted { |
| + t.Fatalf("a stale reservation the view holds = %s, want aborted", got) |
| + } |
| +} |
| + |
| +// Armed, a pass that lists a snapshot keeps today's strict walk of the |
| +// bucket, after the snapshot listing: a record written between the two |
| +// listings that the view has not seen (here, a write that bumps no scope) |
| +// still names its snapshot, so the snapshot is kept. |
| +func TestSeedSweepWalksStrictWithSnapshots(t *testing.T) { |
| + f := newSweepFixture(t) |
| + bare := release.NewDirStore(f.fx.rel.(*deletableStore).root) // no change index: its writes bump nothing |
| + cs := indexedDeployment(t, f.fx) |
| + now := time.Now() |
| + host := "late.shop.example.com" |
| + id := newSeedID(t) |
| + key := f.snapshot(t, host, id, 3*time.Hour) |
| + armRecordView(t, f.fx.srv, cs) |
| + seedsPassBetweenListings = func() { |
| + cs.taken() |
| + rec := provision.Record{Schema: 1, Host: host, Slug: "late", App: sweepApp, Account: release.OpsAccount, |
| + Channel: "stable", CreatedAt: now.UTC(), |
| + Seed: &provision.SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateReady, By: "bob@example.com", At: now}} |
| + if err := provision.CreateRecord(context.Background(), bare, rec); err != nil { |
| + t.Error(err) |
| + } |
| + } |
| + t.Cleanup(func() { seedsPassBetweenListings = nil }) |
| + f.pass(now) |
| + if walks := recordWalks(cs.taken()); len(walks) == 0 { |
| + t.Fatal("a pass with a snapshot made no record walk after the snapshot listing") |
| + } |
| + if !f.snapshotExists(t, key) { |
| + t.Fatal("the snapshot a record names was deleted: the pass read the view, not the bucket") |
| + } |
| +} |
diff --git a/internal/console/server.go b/internal/console/server.go| index 05b4e71a3..808ec15f5 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -1140,6 +1140,19 @@ type Server struct { |
| // relayNow is the relay paths' clock seam: nil means time.Now. Tests |
| // drive lease expiry with it; nothing in production sets it. |
| relayNow func() time.Time |
| + |
| + // recordView is the record view the run and seed sweeps read through |
| + // (recordview.go), set by SetRecordView when CARLOS_CHANGE_INDEX=read |
| + // arms it; nil (disarmed) is every sweep walking the bucket as before |
| + // the index. |
| + recordView *provision.RecordView |
| + // viewUse is the last recordWalkStore outcome, so a switch between the |
| + // view and the bucket is logged once, not on every pass. |
| + viewUse atomic.Int32 |
| + // lastOrphanFull is when the run sweep's orphan housekeeping last swept |
| + // every app with records (runsweep.go); zero until the first, which is |
| + // therefore full. Only the run sweep's one goroutine touches it. |
| + lastOrphanFull time.Time |
| } |
| // edgeFoldTTL is how long one fold of the deployment's edges files stands |
diff --git a/internal/console/store.go b/internal/console/store.go| index 21523c90e..4b898e4c7 100644 |
| --- a/internal/console/store.go |
| +++ b/internal/console/store.go |
| @@ -2559,6 +2559,33 @@ FROM audit WHERE account_id = ? AND app = ? ORDER BY id DESC LIMIT ?`, accountID |
| return out, rows.Err() |
| } |
| +// AppRef names one app by its account's sqid, as the bucket keys it. |
| +type AppRef struct{ Sqid, App string } |
| + |
| +// AppsWithActionSince answers, once each, every (account, app) with an audit |
| +// row of action stamped at or after since. The run sweep's orphan |
| +// housekeeping lists runs/ only for apps with a recent run-start (S3 |
| +// change-index design §4.6). A plain read, like Events. |
| +func (s *Store) AppsWithActionSince(action string, since time.Time) ([]AppRef, error) { |
| + rows, err := s.db.Query(`SELECT DISTINCT accounts.sqid, audit.app |
| +FROM audit JOIN accounts ON accounts.id = audit.account_id |
| +WHERE audit.action = ? AND audit.created_at >= ? |
| +ORDER BY accounts.sqid, audit.app`, action, since.Unix()) |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer rows.Close() |
| + var out []AppRef |
| + for rows.Next() { |
| + var r AppRef |
| + if err := rows.Scan(&r.Sqid, &r.App); err != nil { |
| + return nil, err |
| + } |
| + out = append(out, r) |
| + } |
| + return out, rows.Err() |
| +} |
| + |
| // ---- keys ------------------------------------------------------------------- |
| // SigningKey returns the console's long-lived envelope-signing key (ECDSA |
diff --git a/internal/console/store_test.go b/internal/console/store_test.go| index 046880143..afac6ef3f 100644 |
| --- a/internal/console/store_test.go |
| +++ b/internal/console/store_test.go |
| @@ -3,6 +3,7 @@ package console |
| import ( |
| "path/filepath" |
| "reflect" |
| + "sort" |
| "sync/atomic" |
| "testing" |
| "time" |
| @@ -732,3 +733,45 @@ func TestCreateAppClaimOnce(t *testing.T) { |
| t.Fatal("HasOwners(kass) = false after grant") |
| } |
| } |
| + |
| +// AppsWithActionSince answers each (account, app) with a row of the action |
| +// at or after since, once, by the account's sqid. Older rows and other |
| +// actions are left out. |
| +func TestAppsWithActionSince(t *testing.T) { |
| + t.Parallel() |
| + s := openTestStore(t) |
| + now := time.Now() |
| + one, other := testAccountID(), testAccountID() |
| + for id, sqid := range map[int64]string{one: "acct-one", other: "acct-other"} { |
| + if err := s.CreateAccount(id, sqid, sqid, "bob@example.com", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + for _, r := range []struct { |
| + acct int64 |
| + app string |
| + action string |
| + at time.Time |
| + }{ |
| + {one, "shop", "run-start", now.Add(-time.Hour)}, |
| + {one, "shop", "run-start", now.Add(-2 * time.Hour)}, // the same pair again |
| + {one, "cafe", "run-start", now.Add(-49 * time.Hour)}, |
| + {one, "blog", "run-end", now.Add(-time.Hour)}, |
| + {one, "edge", "run-start", now.Add(-48 * time.Hour)}, // exactly since: in |
| + {other, "shop", "run-start", now.Add(-time.Minute)}, |
| + } { |
| + if err := s.RecordActivity(r.acct, r.app, "bob@example.com", Activity{Action: r.action}, r.at); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + got, err := s.AppsWithActionSince("run-start", now.Add(-48*time.Hour)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + want := []AppRef{{Sqid: "acct-one", App: "edge"}, {Sqid: "acct-one", App: "shop"}, {Sqid: "acct-other", App: "shop"}} |
| + sort.Slice(got, func(i, j int) bool { return got[i].Sqid+"/"+got[i].App < got[j].Sqid+"/"+got[j].App }) |
| + sort.Slice(want, func(i, j int) bool { return want[i].Sqid+"/"+want[i].App < want[j].Sqid+"/"+want[j].App }) |
| + if !reflect.DeepEqual(got, want) { |
| + t.Fatalf("AppsWithActionSince = %+v, want %+v", got, want) |
| + } |
| +} |