Space reconciliation passes by ten times the last pass's time (closes #250)
check / check (push) Canceled after 0s
check / check (push) Canceled after 0s
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. 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. Model: opus-5-5
This commit is contained in:
@@ -199,7 +199,9 @@ 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
|
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
|
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
|
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 after a look that took longer than 30 seconds, pixa waits
|
||||||
|
ten times as long as it took before looking again.
|
||||||
|
|
||||||
pixa's target is 1-5k r/s, which has not been measured at that rate (see Load
|
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
|
Test). Toward it, pixa keeps in memory the content types of the 10,000
|
||||||
|
|||||||
@@ -30,6 +30,8 @@ P2: security: per-origin rate limiting
|
|||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
|
- 2026-10-18 the next reconciliation pass waits 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-16 P1 security bug: the check that keeps pixa from fetching internal
|
- 2026-10-16 P1 security bug: the check that keeps pixa from fetching internal
|
||||||
addresses refuses the local-use NAT64 range (`64:ff9b:1::/48`, RFC 8215) as it
|
addresses refuses the local-use NAT64 range (`64:ff9b:1::/48`, RFC 8215) as it
|
||||||
refuses `64:ff9b::/96`, and refuses an IPv6 address in the IPv4-compatible
|
refuses `64:ff9b::/96`, and refuses an IPv6 address in the IPv4-compatible
|
||||||
|
|||||||
@@ -18,6 +18,14 @@ import (
|
|||||||
// write-pressure wakeups triggered by stores.
|
// write-pressure wakeups triggered by stores.
|
||||||
const DefaultEvictionInterval = 5 * time.Minute
|
const DefaultEvictionInterval = 5 * time.Minute
|
||||||
|
|
||||||
|
// reconciliationWaitFactor is how many times as long as a
|
||||||
|
// reconciliation pass took the eviction loop waits, 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
|
// evictionBatchSize is how many LRU candidates of each class (variants
|
||||||
// and source blobs) one eviction pass fetches from the database.
|
// and source blobs) one eviction pass fetches from the database.
|
||||||
const evictionBatchSize = 100
|
const evictionBatchSize = 100
|
||||||
@@ -438,10 +446,12 @@ func (c *Cache) notifyWritePressure() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// StartEviction launches the background eviction goroutine, which
|
// 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
|
// removes the expired negative cache rows at startup and again on every
|
||||||
// periodic tick thereafter, and evicts to the configured limit on the
|
// periodic tick thereafter, reconciles the database accounting with the
|
||||||
// given periodic interval and on write-pressure notifications. On a
|
// cache directories at startup and again on the first periodic tick
|
||||||
|
// that comes 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
|
// 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
|
// 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
|
// 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) {
|
func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
|
||||||
defer close(c.evictionDone)
|
defer close(c.evictionDone)
|
||||||
|
|
||||||
c.runReconciliationPass(ctx)
|
nextReconciliation := c.runReconciliationPass(ctx)
|
||||||
c.runCleanExpired(ctx)
|
c.runCleanExpired(ctx)
|
||||||
c.runEvictionPass(ctx)
|
c.runEvictionPass(ctx)
|
||||||
|
|
||||||
@@ -494,16 +504,20 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
|
|||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return
|
return
|
||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
// Reconciliation walks the cache directories, so it only
|
// Reconciliation walks the cache directories and reads a
|
||||||
// runs on the periodic ticker rather than on every
|
// row for every file, so it runs only on the periodic
|
||||||
// write-pressure wakeup, keeping it off the per-store hot
|
// ticker rather than on every write-pressure wakeup, and
|
||||||
// path. Reusing the eviction interval itself (rather than a
|
// not before nextReconciliation, which is ten times as long
|
||||||
// separate, longer one) is a deliberate choice: it is the
|
// after the last pass ended as that pass took. Passes then
|
||||||
// simplest option that still bounds how long a store's
|
// never run back to back, however long one takes. A file
|
||||||
// best-effort accounting insert can stay silently
|
// whose best-effort accounting insert failed in a store
|
||||||
// unaccounted for to one interval, on a process that is
|
// stays unaccounted for until the next pass, for about the
|
||||||
// already running this loop regardless.
|
// longer of one eviction interval and ten times the last
|
||||||
c.runReconciliationPass(ctx)
|
// pass's time.
|
||||||
|
if !time.Now().Before(nextReconciliation) {
|
||||||
|
nextReconciliation = c.runReconciliationPass(ctx)
|
||||||
|
}
|
||||||
|
|
||||||
c.runCleanExpired(ctx)
|
c.runCleanExpired(ctx)
|
||||||
case <-c.evictionPressure:
|
case <-c.evictionPressure:
|
||||||
}
|
}
|
||||||
@@ -527,17 +541,25 @@ func (c *Cache) runEvictionPass(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// runReconciliationPass runs one reconciliation pass, logging failures
|
// runReconciliationPass runs one reconciliation pass, logging failures
|
||||||
// instead of propagating them (the loop must keep running). It does
|
// instead of propagating them (the loop must keep running), and returns
|
||||||
// nothing once ctx is cancelled.
|
// the earliest time the next pass may start: reconciliationWaitFactor
|
||||||
func (c *Cache) runReconciliationPass(ctx context.Context) {
|
// 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 {
|
if ctx.Err() != nil {
|
||||||
return
|
return time.Now()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
started := time.Now()
|
||||||
|
|
||||||
err := c.reconcileAccounting(ctx)
|
err := c.reconcileAccounting(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
c.log.Warn("cache accounting reconciliation failed", "error", err)
|
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
|
// 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
|
// reconcileAccounting synchronizes the database size accounting with
|
||||||
// the actual contents of the cache directories. It runs at startup and
|
// the actual contents of the cache directories. It runs at startup and
|
||||||
// again on every periodic eviction tick thereafter, off the request
|
// again on later periodic eviction ticks, each no sooner than 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
|
// hot path: it adopts variant files that predate the accounting table
|
||||||
// (or whose accounting insert failed, e.g. StoreVariant's best-effort
|
// (or whose accounting insert failed, e.g. StoreVariant's best-effort
|
||||||
// insert under transient DB contention), drops accounting rows whose
|
// 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
|
// 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.
|
// 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
|
// Running it periodically, not just once, bounds how long such drift can
|
||||||
// accumulate unaccounted for on a long-running process to one eviction
|
// accumulate unaccounted for on a long-running process to about the
|
||||||
// interval. Once ctx is cancelled, it stops at the next file or row and
|
// longer of one eviction interval and ten times the last pass's time.
|
||||||
// returns ctx's error.
|
// 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 {
|
func (c *Cache) reconcileAccounting(ctx context.Context) error {
|
||||||
if c.disabled {
|
if c.disabled {
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
Reference in New Issue
Block a user