Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b984b27cf8 | ||
|
|
af6ab4a668 |
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user