From 9eacfa64064e9c731fe26272abd53f8bf7f17102 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 19 Oct 2026 05:00:48 +0200 Subject: [PATCH] Space reconciliation passes by ten times the last pass's time (closes #250) The reconciliation pass reads a row and stats a file for every cached file, and it ran on every eviction tick, so on a large cache a pass longer than the tick was followed by the next one at once. It still runs at startup and is considered on each tick, but it now starts no sooner than ten times as long after the last pass ended as that pass took, so it takes at most about a tenth of the time however large the cache. Eviction and the removal of expired negative cache rows still run on every tick. The comments that bound the drift, and README.md, give the new bound. No existing test changed. Model: opus-5-5 --- README.md | 6 +- TODO.md | 3 + internal/imgcache/eviction.go | 68 +++++++++++------ internal/imgcache/eviction_internal_test.go | 83 +++++++++++++++++++++ 4 files changed, 137 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index c069b90..b473c1a 100644 --- a/README.md +++ b/README.md @@ -199,7 +199,11 @@ into place, so a crash leaves no file under its own name with part of its data. When pixa starts, and every five minutes after, it removes each cached source or transformed image whose file is empty, or not the size the database records for it, with a warning naming the file; the next request for it fetches the source -or converts the image again. +or converts the image again. Looking for those files reads the database once for +every cached file, so pixa skips each five-minute look that would start less +than ten times as long after the last look ended as that look took. After a look +that took one minute, the next one starts 10 to 15 minutes after that look +ended. pixa's target is 1-5k r/s, which has not been measured at that rate (see Load Test). Toward it, pixa keeps in memory the content types of the 10,000 diff --git a/TODO.md b/TODO.md index 06e7082..44ae4c1 100644 --- a/TODO.md +++ b/TODO.md @@ -30,6 +30,9 @@ P2: security: per-origin rate limiting # Completed Steps +- 2026-10-18 the next reconciliation pass waits at least ten times as long as + the last one took, so passes on a large cache no longer run back to back + (closes #250). - 2026-10-18 a `blocked_networks` entry in the IPv4-mapped form refused nothing; `::ffff:10.0.0.0/104` is now read as `10.0.0.0/8`, and such an entry shorter than `/96` aborts startup (closes #301). diff --git a/internal/imgcache/eviction.go b/internal/imgcache/eviction.go index 2fe34fb..83828c4 100644 --- a/internal/imgcache/eviction.go +++ b/internal/imgcache/eviction.go @@ -18,6 +18,14 @@ import ( // write-pressure wakeups triggered by stores. const DefaultEvictionInterval = 5 * time.Minute +// reconciliationWaitFactor is how many times as long as a +// reconciliation pass took the eviction loop waits at least, after that +// pass ends, before it starts the next one. A pass reads a row and stats a +// file for every cached file, so its time grows with the cache; waiting +// ten times as long keeps reconciliation to at most a tenth of the time, +// however large the cache is. +const reconciliationWaitFactor = 10 + // evictionBatchSize is how many LRU candidates of each class (variants // and source blobs) one eviction pass fetches from the database. const evictionBatchSize = 100 @@ -438,10 +446,12 @@ func (c *Cache) notifyWritePressure() { } // StartEviction launches the background eviction goroutine, which -// reconciles the database accounting with the cache directories and // removes the expired negative cache rows at startup and again on every -// periodic tick thereafter, and evicts to the configured limit on the -// given periodic interval and on write-pressure notifications. On a +// periodic tick thereafter, reconciles the database accounting with the +// cache directories at startup and again on the first periodic tick +// that comes at least ten times as long after a reconciliation pass +// ended as that pass took, and evicts to the configured limit on the given +// periodic interval and on write-pressure notifications. On a // disabled cache it only removes the expired negative cache rows, as // the negative cache is used either way. It is a no-op when already // started. The goroutine outlives the caller, so it runs with its own @@ -482,7 +492,7 @@ func (c *Cache) StopEviction(ctx context.Context) error { func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) { defer close(c.evictionDone) - c.runReconciliationPass(ctx) + nextReconciliation := c.runReconciliationPass(ctx) c.runCleanExpired(ctx) c.runEvictionPass(ctx) @@ -494,16 +504,20 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) { case <-ctx.Done(): return case <-ticker.C: - // Reconciliation walks the cache directories, so it only - // runs on the periodic ticker rather than on every - // write-pressure wakeup, keeping it off the per-store hot - // path. Reusing the eviction interval itself (rather than a - // separate, longer one) is a deliberate choice: it is the - // simplest option that still bounds how long a store's - // best-effort accounting insert can stay silently - // unaccounted for to one interval, on a process that is - // already running this loop regardless. - c.runReconciliationPass(ctx) + // Reconciliation walks the cache directories and reads a + // row for every file, so it runs only on the periodic + // ticker rather than on every write-pressure wakeup, and + // only on the first tick at or after nextReconciliation, + // which is ten times as long after the last pass ended as + // that pass took. Passes then never run back to back, + // however long one takes. A file whose best-effort + // accounting insert failed in a store stays unaccounted for + // until the next pass, up to about one eviction interval + // plus ten times the last pass's time. + if !time.Now().Before(nextReconciliation) { + nextReconciliation = c.runReconciliationPass(ctx) + } + c.runCleanExpired(ctx) case <-c.evictionPressure: } @@ -527,17 +541,25 @@ func (c *Cache) runEvictionPass(ctx context.Context) { } // runReconciliationPass runs one reconciliation pass, logging failures -// instead of propagating them (the loop must keep running). It does -// nothing once ctx is cancelled. -func (c *Cache) runReconciliationPass(ctx context.Context) { +// instead of propagating them (the loop must keep running), and returns +// the earliest time the next pass may start: reconciliationWaitFactor +// times as long after this pass ended as it took. It does nothing once +// ctx is cancelled. +func (c *Cache) runReconciliationPass(ctx context.Context) time.Time { if ctx.Err() != nil { - return + return time.Now() } + started := time.Now() + err := c.reconcileAccounting(ctx) if err != nil { c.log.Warn("cache accounting reconciliation failed", "error", err) } + + ended := time.Now() + + return ended.Add(reconciliationWaitFactor * ended.Sub(started)) } // runCleanExpired runs CleanExpired, logging a failure instead of @@ -556,7 +578,8 @@ func (c *Cache) runCleanExpired(ctx context.Context) { // reconcileAccounting synchronizes the database size accounting with // the actual contents of the cache directories. It runs at startup and -// again on every periodic eviction tick thereafter, off the request +// then on the first periodic eviction tick that comes at least ten times +// as long after the last pass ended as that pass took, off the request // hot path: it adopts variant files that predate the accounting table // (or whose accounting insert failed, e.g. StoreVariant's best-effort // insert under transient DB contention), drops accounting rows whose @@ -566,9 +589,10 @@ func (c *Cache) runCleanExpired(ctx context.Context) { // files left behind by crashed writes, and last checks the total cache // usage against the tables. It reads the tables a page at a time. // Running it periodically, not just once, bounds how long such drift can -// accumulate unaccounted for on a long-running process to one eviction -// interval. Once ctx is cancelled, it stops at the next file or row and -// returns ctx's error. +// accumulate unaccounted for on a long-running process to about one +// eviction interval plus ten times the last pass's time. +// Once ctx is cancelled, it stops at the next file or row and returns +// ctx's error. func (c *Cache) reconcileAccounting(ctx context.Context) error { if c.disabled { return nil diff --git a/internal/imgcache/eviction_internal_test.go b/internal/imgcache/eviction_internal_test.go index 7d54df1..d1ddf31 100644 --- a/internal/imgcache/eviction_internal_test.go +++ b/internal/imgcache/eviction_internal_test.go @@ -918,6 +918,89 @@ func TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup(t *testing.T) { } } +// TestReconciliationPassesAreSpacedByTenTimesTheirTime makes the hook +// called after every read of the reconciliation pass sleep 10 ms, runs +// the eviction loop on a 10 ms interval, and records when each of those +// calls starts and ends. A pass starts before its first call and ends +// after its last, so the time from the first call's start to the last +// call's end is at most the pass's time. The next pass's first call must +// start at least ten times that long after the last call ended. +func TestReconciliationPassesAreSpacedByTenTimesTheirTime(t *testing.T) { + t.Parallel() + + cache, _ := newEvictionTestCache(t, 1<<30) + + // A pass made here, before the loop starts, counts the reads one + // pass makes on this empty cache, so the reads recorded below can be + // split into passes. + readsPerPass := 0 + cache.reconciliationReadTestHook = func(int) { readsPerPass++ } + + err := cache.reconcileAccounting(t.Context()) + if err != nil { + t.Fatalf("reconcileAccounting failed: %v", err) + } + + t.Logf("one reconciliation pass makes %d reads", readsPerPass) + + if readsPerPass == 0 { + t.Fatal("a reconciliation pass made no read") + } + + // The pass at startup and two after it. + const passes = 3 + + type read struct{ start, end time.Time } + + // The channel holds every read the test looks at, so none of those + // is dropped; reads after them are, and never block the loop. + reads := make(chan read, passes*readsPerPass) + + cache.reconciliationReadTestHook = func(int) { + start := time.Now() + + time.Sleep(10 * time.Millisecond) + + select { + case reads <- read{start: start, end: time.Now()}: + default: + } + } + + cache.StartEviction(10 * time.Millisecond) + defer func() { _ = cache.StopEviction(t.Context()) }() + + recorded := make([]read, 0, passes*readsPerPass) + timeout := time.After(15 * time.Second) + + for len(recorded) < passes*readsPerPass { + select { + case r := <-reads: + recorded = append(recorded, r) + case <-timeout: + t.Fatalf("saw %d reconciliation reads in 15 s, want %d (%d passes)", + len(recorded), passes*readsPerPass, passes) + } + } + + for pass := range passes - 1 { + first := recorded[pass*readsPerPass] + last := recorded[(pass+1)*readsPerPass-1] + next := recorded[(pass+1)*readsPerPass] + + passTime := last.end.Sub(first.start) + wait := next.start.Sub(last.end) + + t.Logf("pass %d took at least %v, and the next pass started %v after it", + pass+1, passTime, wait) + + if wait < 10*passTime { + t.Errorf("pass %d took at least %v, and the next pass started %v "+ + "after it, want at least %v", pass+1, passTime, wait, 10*passTime) + } + } +} + // TestStopEvictionInterruptsPassInProgress holds the test database's // only connection, so the startup reconciliation pass waits for it, and // checks that StopEviction stops that pass instead of waiting for the