Author SHA1 Message Date
sneak b984b27cf8 Run the reconciliation pass at startup and once a day (closes #250)
check / check (push) Waiting to run
The pass makes a database query for every cached file and ran on every
eviction tick, so on a large cache one pass could outlast the tick and
the next started at once. It now runs on a timer of its own that starts
when a pass ends, a day long. Eviction and the removal of expired
negative cache rows keep the eviction interval.

Model: opus-5-5
2026-10-15 20:43:30 +00:00
sneak af6ab4a668 Test that eviction ticks read nothing for reconciliation
The eviction loop runs the reconciliation pass on every periodic tick,
so this test fails until the pass runs at startup and then once a day.

Model: opus-5-5
2026-10-15 20:42:35 +00:00
4 changed files with 139 additions and 40 deletions
+1 -1
View File
@@ -168,7 +168,7 @@ references rather than using filesystem refcounting.
Source and transformed image files, and the `.meta` file beside each transformed
image, are written under a temporary name, synced to disk, and only then renamed
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 once a day 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.
+17 -7
View File
@@ -30,6 +30,16 @@ P2: security: per-origin rate limiting
# Completed Steps
- 2026-10-15 the reconciliation pass runs at startup and then once a day (closes
#250): it ran on every eviction tick, every 5 minutes in pixad, and makes a
database query for every cached file, on the one connection requests also use.
On a large cache a pass could take longer than 5 minutes, and the next one
then started at once, so the passes never stopped. The next pass now starts a
day after the previous one ends, so passes never run back to back. Eviction
and the removal of expired negative cache rows keep their 5-minute tick, and
eviction also runs after each reconciliation pass. A variant file whose
database row failed to be written is now counted in the cache size up to about
a day later, rather than 5 minutes.
- 2026-10-15 the canonical files are re-vendored from `sneak/prompts` commit
`e3f79c4` (closes #221): `.dockerignore`, `.editorconfig`,
`.gitea/workflows/check.yml`, `.gitignore`, `.golangci.yml`,
@@ -81,13 +91,13 @@ P2: security: per-origin rate limiting
each transformed image's `.meta` file, are synced to disk before they are
renamed into place, and their directory after; the `.meta` file is written
before the image, and a failed directory sync removes the renamed file. The
reconciliation pass, at startup and every five minutes, removes a cached
source or transformed image whose file is empty, or not the size the database
records for it, with its rows and a warning naming the file, so the next
request fetches the source or converts the image again; requests compare no
sizes. Storing a source over a file of another size replaces that file. A
source that fails to decode is not removed for that alone: a damaged upstream
image would then be fetched on every request.
reconciliation pass, at startup and once a day, removes a cached source or
transformed image whose file is empty, or not the size the database records
for it, with its rows and a warning naming the file, so the next request
fetches the source or converts the image again; requests compare no sizes.
Storing a source over a file of another size replaces that file. A source that
fails to decode is not removed for that alone: a damaged upstream image would
then be fetched on every request.
- 2026-10-15 a transcode stops once its request's `downstream_timeout` has
ended, unless it has begun encoding (closes #243): the image processor looked
at its context only while it waited for a processing slot, so an image was
+41 -32
View File
@@ -18,6 +18,11 @@ import (
// write-pressure wakeups triggered by stores.
const DefaultEvictionInterval = 5 * time.Minute
// reconciliationInterval is how long after one reconciliation pass ends
// the next one starts. A pass reads a row for every cached file, so its
// cost grows with the cache, and it runs this rarely for that reason.
const reconciliationInterval = 24 * time.Hour
// evictionBatchSize is how many LRU candidates of each class (variants
// and source blobs) one eviction pass fetches from the database.
const evictionBatchSize = 100
@@ -437,15 +442,17 @@ 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
// 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
// context, which StopEviction cancels.
// StartEviction launches the background eviction goroutine. At startup
// it reconciles the database accounting with the cache directories,
// removes the expired negative cache rows and evicts to the configured
// limit. After that it reconciles again a day after each reconciliation
// pass ends, removes the expired negative cache rows on every tick of
// the given interval, and evicts to the limit on every tick, after each
// reconciliation pass 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 context, which
// StopEviction cancels.
func (c *Cache) StartEviction(interval time.Duration) {
if c.evictionCancel != nil {
return
@@ -486,6 +493,16 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
c.runCleanExpired(ctx)
c.runEvictionPass(ctx)
// Reconciliation walks the cache directories and reads a row for
// every file, so it runs on a timer of its own rather than on the
// periodic tick or on write-pressure wakeups. The timer starts only
// once a pass has ended, so passes never run back to back, however
// long one takes. Running it daily, not just at startup, bounds how
// long a store's best-effort accounting insert can stay silently
// unaccounted for to about one day, plus the time the passes take.
reconciliationTimer := time.NewTimer(reconciliationInterval)
defer reconciliationTimer.Stop()
ticker := time.NewTicker(interval)
defer ticker.Stop()
@@ -493,17 +510,10 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
select {
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.
case <-reconciliationTimer.C:
c.runReconciliationPass(ctx)
reconciliationTimer.Reset(reconciliationInterval)
case <-ticker.C:
c.runCleanExpired(ctx)
case <-c.evictionPressure:
}
@@ -556,19 +566,18 @@ 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
// 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
// files are missing, removes source blob files the database does not
// know (and rows whose files are gone), removes source and variant files
// that are empty or not the size their rows record, sweeps stale temp
// 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.
// then once a day, 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 files are missing, removes source blob
// files the database does not know (and rows whose files are gone),
// removes source and variant files that are empty or not the size their
// rows record, sweeps stale temp 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 daily, not just once, bounds how
// long such drift can accumulate unaccounted for on a long-running
// process to about one day. 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
@@ -9,7 +9,9 @@ import (
"log/slog"
"os"
"path/filepath"
"strconv"
"strings"
"sync/atomic"
"testing"
"time"
@@ -918,6 +920,84 @@ func TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup(t *testing.T) {
}
}
// TestEvictionTicksDoNotReconcile starts the eviction loop with an
// interval of 10 ms. The startup pass reconciles before it removes the
// expired negative cache rows, so once the expired row stored before the
// start is gone, the startup reconciliation has made all its reads. The
// test then waits for three periodic ticks, each seen when it removes an
// expired row stored after the previous one went. The reconciliation
// pass runs at startup and then once a day, so none of those ticks may
// read a content table for reconciliation.
func TestEvictionTicksDoNotReconcile(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
var reconciliationReads atomic.Int64
cache.reconciliationReadTestHook = func(int) { reconciliationReads.Add(1) }
insertExpiredNegativeCacheRow(t, cache, "/startup.jpg")
cache.StartEviction(10 * time.Millisecond)
defer func() { _ = cache.StopEviction(t.Context()) }()
waitForNegativeCacheRowRemoved(t, cache, "/startup.jpg")
startupReads := reconciliationReads.Load()
t.Logf("the startup pass made %d reconciliation reads", startupReads)
if startupReads == 0 {
t.Fatal("the startup pass made no reconciliation read")
}
for tick := range 3 {
path := "/tick-" + strconv.Itoa(tick) + ".jpg"
insertExpiredNegativeCacheRow(t, cache, path)
waitForNegativeCacheRowRemoved(t, cache, path)
}
tickReads := reconciliationReads.Load() - startupReads
t.Logf("three eviction ticks made %d reconciliation reads", tickReads)
if tickReads != 0 {
t.Errorf("three eviction ticks made %d reconciliation reads, want 0: "+
"the reconciliation pass ran on an eviction tick", tickReads)
}
}
// insertExpiredNegativeCacheRow adds a negative_cache row for sourcePath
// that expired an hour ago.
func insertExpiredNegativeCacheRow(t *testing.T, cache *Cache, sourcePath string) {
t.Helper()
_, err := cache.db.ExecContext(t.Context(), `
INSERT INTO negative_cache
(source_host, source_path, status_code, expires_at)
VALUES (?, ?, 404, datetime('now', '-1 hour'))
`, testHostExample, sourcePath)
if err != nil {
t.Fatalf("failed to insert an expired negative cache row: %v", err)
}
}
// waitForNegativeCacheRowRemoved waits up to five seconds for the
// eviction loop to remove the negative_cache row for sourcePath.
func waitForNegativeCacheRowRemoved(t *testing.T, cache *Cache, sourcePath string) {
t.Helper()
deadline := time.Now().Add(5 * time.Second)
for negativeCacheRowCount(t, cache, sourcePath) > 0 {
if time.Now().After(deadline) {
t.Fatalf("the expired negative cache row for %s was not removed", sourcePath)
}
time.Sleep(10 * time.Millisecond)
}
}
// 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