sneak
5d78e62ae9
check / check (push) Waiting to run
A source is now one source_content row and one file named by its SourceKey, the SHA-256 of its host, path and query, made as the variant key is. A store writes the file, then upserts the row, and removes the file when the row cannot be written; an eviction deletes the row, then the file, with no lock. source_metadata, the JSON files under cache/metadata/, MetadataStorage, the content lock, the eviction transaction and its test hook are gone, as is the reconciliation pass that dropped a source row whose file is missing. Two source URLs serving the same bytes store them twice. Model: opus-5-5
1218 lines
33 KiB
Go
1218 lines
33 KiB
Go
package imgcache
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"io/fs"
|
|
"log/slog"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"modernc.org/sqlite"
|
|
"sneak.berlin/go/pixa/internal/database"
|
|
"sneak.berlin/go/pixa/internal/httpfetcher"
|
|
"sneak.berlin/go/pixa/internal/imagerequest"
|
|
)
|
|
|
|
// sqliteTimestampFormat matches the format SQLite's CURRENT_TIMESTAMP
|
|
// produces, so injected timestamps compare correctly against ones the
|
|
// implementation writes.
|
|
const sqliteTimestampFormat = "2006-01-02 15:04:05"
|
|
|
|
// testVariantKeyOne through testVariantKeyFour are the variant cache
|
|
// keys reused across the eviction tests, in the order the tests store
|
|
// them.
|
|
const (
|
|
testVariantKeyOne imagerequest.VariantKey = "aabbccdd0001"
|
|
testVariantKeyTwo imagerequest.VariantKey = "aabbccdd0002"
|
|
testVariantKeyThree imagerequest.VariantKey = "aabbccdd0003"
|
|
testVariantKeyFour imagerequest.VariantKey = "aabbccdd0004"
|
|
)
|
|
|
|
// evictionTestDB creates an in-memory SQLite database with the real
|
|
// production schema, limited to a single connection so the background
|
|
// eviction goroutine shares the same in-memory database as the test.
|
|
func evictionTestDB(t *testing.T) *sql.DB {
|
|
t.Helper()
|
|
|
|
db, err := sql.Open("sqlite", ":memory:")
|
|
if err != nil {
|
|
t.Fatalf("failed to open test db: %v", err)
|
|
}
|
|
|
|
db.SetMaxOpenConns(1)
|
|
|
|
err = database.ApplyMigrations(context.Background(), db, nil)
|
|
if err != nil {
|
|
t.Fatalf("failed to apply migrations: %v", err)
|
|
}
|
|
|
|
t.Cleanup(func() { _ = db.Close() })
|
|
|
|
return db
|
|
}
|
|
|
|
// newEvictionTestCache creates a Cache backed by a temp directory and
|
|
// an in-memory database, with the given size limit.
|
|
func newEvictionTestCache(t *testing.T, maxBytes int64) (*Cache, string) {
|
|
t.Helper()
|
|
|
|
tmpDir := t.TempDir()
|
|
db := evictionTestDB(t)
|
|
|
|
// maxBytes zero mirrors the production mapping of
|
|
// cache_max_bytes: 0 (handlers sets DisableDiskCache); at the
|
|
// CacheConfig layer itself a zero MaxBytes means "no limit" for
|
|
// backwards compatibility with existing fixtures.
|
|
cache, err := NewCache(db, CacheConfig{
|
|
StateDir: tmpDir,
|
|
NegativeTTL: 5 * time.Minute,
|
|
MaxBytes: maxBytes,
|
|
DisableDiskCache: maxBytes == 0,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("failed to create cache: %v", err)
|
|
}
|
|
|
|
return cache, tmpDir
|
|
}
|
|
|
|
// storeEvictionTestSource stores content as a fetched source for
|
|
// host/path and returns the resulting source key.
|
|
func storeEvictionTestSource(
|
|
t *testing.T, cache *Cache, host, path string, content []byte,
|
|
) imagerequest.SourceKey {
|
|
t.Helper()
|
|
|
|
req := &imagerequest.Request{
|
|
Source: imagerequest.Source{
|
|
Host: host,
|
|
Path: path,
|
|
},
|
|
Transform: imagerequest.Transform{
|
|
Format: imagerequest.FormatJPEG,
|
|
Quality: 85,
|
|
FitMode: imagerequest.FitCover,
|
|
},
|
|
}
|
|
|
|
result := &httpfetcher.FetchResult{
|
|
StatusCode: 200,
|
|
ContentType: testContentTypeJPEG,
|
|
ContentLength: int64(len(content)),
|
|
Headers: map[string][]string{
|
|
testHeaderContentType: {testContentTypeJPEG},
|
|
},
|
|
}
|
|
|
|
key, err := cache.StoreSource(
|
|
context.Background(), req, bytes.NewReader(content), result,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("StoreSource(%s%s) failed: %v", host, path, err)
|
|
}
|
|
|
|
return key
|
|
}
|
|
|
|
// storeEvictionTestVariant stores content as a processed variant under
|
|
// the given cache key.
|
|
func storeEvictionTestVariant(
|
|
t *testing.T, cache *Cache, key imagerequest.VariantKey, content []byte,
|
|
) {
|
|
t.Helper()
|
|
|
|
err := cache.StoreVariant(t.Context(), key, bytes.NewReader(content), "image/webp")
|
|
if err != nil {
|
|
t.Fatalf("StoreVariant(%s) failed: %v", key, err)
|
|
}
|
|
}
|
|
|
|
// setVariantLastAccessed backdates the last access time of a tracked
|
|
// variant, to make LRU ordering deterministic in tests.
|
|
func setVariantLastAccessed(
|
|
t *testing.T, cache *Cache, key imagerequest.VariantKey, when time.Time,
|
|
) {
|
|
t.Helper()
|
|
|
|
res, err := cache.db.ExecContext(t.Context(),
|
|
`UPDATE variant_content SET last_accessed_at = ? WHERE cache_key = ?`,
|
|
when.UTC().Format(sqliteTimestampFormat), string(key),
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("failed to set variant last_accessed_at: %v", err)
|
|
}
|
|
|
|
affected, err := res.RowsAffected()
|
|
if err != nil {
|
|
t.Fatalf("failed to read affected rows: %v", err)
|
|
}
|
|
|
|
if affected != 1 {
|
|
t.Fatalf("variant %s has no accounting row (affected=%d); "+
|
|
"stores must track variants in the database", key, affected)
|
|
}
|
|
}
|
|
|
|
// setSourceLastAccessed backdates the last access time of a tracked
|
|
// source content blob.
|
|
func setSourceLastAccessed(
|
|
t *testing.T, cache *Cache, key imagerequest.SourceKey, when time.Time,
|
|
) {
|
|
t.Helper()
|
|
|
|
res, err := cache.db.ExecContext(t.Context(),
|
|
`UPDATE source_content SET last_accessed_at = ? WHERE source_key = ?`,
|
|
when.UTC().Format(sqliteTimestampFormat), string(key),
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("failed to set source last_accessed_at: %v", err)
|
|
}
|
|
|
|
affected, err := res.RowsAffected()
|
|
if err != nil {
|
|
t.Fatalf("failed to read affected rows: %v", err)
|
|
}
|
|
|
|
if affected != 1 {
|
|
t.Fatalf("source %s has no accounting row (affected=%d)", key, affected)
|
|
}
|
|
}
|
|
|
|
// countRows returns the number of rows the given query yields.
|
|
func countRows(t *testing.T, cache *Cache, query string, args ...any) int {
|
|
t.Helper()
|
|
|
|
var n int
|
|
|
|
err := cache.db.QueryRowContext(t.Context(), query, args...).Scan(&n)
|
|
if err != nil {
|
|
t.Fatalf("count query %q failed: %v", query, err)
|
|
}
|
|
|
|
return n
|
|
}
|
|
|
|
// assertNoDanglingReferences verifies the core eviction invariant:
|
|
// every database row that references cache content on disk points at a
|
|
// file that actually exists.
|
|
func assertNoDanglingReferences(t *testing.T, cache *Cache) {
|
|
t.Helper()
|
|
|
|
rows, err := cache.db.QueryContext(t.Context(),
|
|
`SELECT source_key FROM source_content`,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("failed to query source_content: %v", err)
|
|
}
|
|
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
for rows.Next() {
|
|
var key string
|
|
|
|
err := rows.Scan(&key)
|
|
if err != nil {
|
|
t.Fatalf("failed to scan source_key: %v", err)
|
|
}
|
|
|
|
if !cache.srcContent.Exists(imagerequest.SourceKey(key)) {
|
|
t.Errorf("source_content references content %s but the file is missing",
|
|
key)
|
|
}
|
|
}
|
|
|
|
err = rows.Err()
|
|
if err != nil {
|
|
t.Fatalf("source_content iteration failed: %v", err)
|
|
}
|
|
|
|
variantRows, err := cache.db.QueryContext(t.Context(),
|
|
`SELECT cache_key FROM variant_content`)
|
|
if err != nil {
|
|
t.Fatalf("failed to query variant_content: %v", err)
|
|
}
|
|
|
|
defer func() { _ = variantRows.Close() }()
|
|
|
|
for variantRows.Next() {
|
|
var key string
|
|
|
|
err := variantRows.Scan(&key)
|
|
if err != nil {
|
|
t.Fatalf("failed to scan cache_key: %v", err)
|
|
}
|
|
|
|
if !cache.variants.Exists(imagerequest.VariantKey(key)) {
|
|
t.Errorf("variant_content references key %s but the file is missing", key)
|
|
}
|
|
}
|
|
|
|
err = variantRows.Err()
|
|
if err != nil {
|
|
t.Fatalf("variant_content iteration failed: %v", err)
|
|
}
|
|
}
|
|
|
|
// waitForUsageAtOrBelow polls UsageBytes until it reaches limit or the
|
|
// timeout expires, returning the last observed usage.
|
|
func waitForUsageAtOrBelow(
|
|
t *testing.T, cache *Cache, limit int64, timeout time.Duration,
|
|
) int64 {
|
|
t.Helper()
|
|
|
|
deadline := time.Now().Add(timeout)
|
|
|
|
var usage int64
|
|
|
|
for time.Now().Before(deadline) {
|
|
var err error
|
|
|
|
usage, err = cache.UsageBytes(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("UsageBytes failed: %v", err)
|
|
}
|
|
|
|
if usage <= limit {
|
|
return usage
|
|
}
|
|
|
|
time.Sleep(25 * time.Millisecond)
|
|
}
|
|
|
|
return usage
|
|
}
|
|
|
|
func TestUsageBytesAccountsSourceAndVariantBytes(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1<<30)
|
|
|
|
storeEvictionTestSource(t, cache, "src.example.com", "/a.jpg",
|
|
bytes.Repeat([]byte{0xAA}, 1000))
|
|
storeEvictionTestSource(t, cache, "src.example.com", "/b.jpg",
|
|
bytes.Repeat([]byte{0xAB}, 2000))
|
|
storeEvictionTestVariant(t, cache, testVariantKeyOne,
|
|
bytes.Repeat([]byte{0xAC}, 500))
|
|
storeEvictionTestVariant(t, cache, testVariantKeyTwo,
|
|
bytes.Repeat([]byte{0xAD}, 250))
|
|
|
|
usage, err := cache.UsageBytes(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("UsageBytes failed: %v", err)
|
|
}
|
|
|
|
if usage != 3750 {
|
|
t.Errorf("UsageBytes = %d, want 3750 (1000+2000+500+250)", usage)
|
|
}
|
|
}
|
|
|
|
func TestEvictToLimitEvictsLeastRecentlyUsedFirst(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
const limit = 3000
|
|
|
|
cache, _ := newEvictionTestCache(t, limit)
|
|
|
|
now := time.Now()
|
|
keys := []imagerequest.VariantKey{
|
|
testVariantKeyOne, testVariantKeyTwo,
|
|
testVariantKeyThree, testVariantKeyFour,
|
|
}
|
|
fills := []byte{0x01, 0x02, 0x03, 0x04}
|
|
ages := []time.Duration{4 * time.Hour, 3 * time.Hour, 2 * time.Hour, 1 * time.Hour}
|
|
|
|
for i, key := range keys {
|
|
storeEvictionTestVariant(t, cache, key, bytes.Repeat([]byte{fills[i]}, 1000))
|
|
setVariantLastAccessed(t, cache, key, now.Add(-ages[i]))
|
|
}
|
|
|
|
err := cache.EvictToLimit(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("EvictToLimit failed: %v", err)
|
|
}
|
|
|
|
usage, err := cache.UsageBytes(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("UsageBytes failed: %v", err)
|
|
}
|
|
|
|
if usage > limit {
|
|
t.Errorf("usage after eviction = %d, want <= %d", usage, limit)
|
|
}
|
|
|
|
if cache.variants.Exists(keys[0]) {
|
|
t.Errorf("least recently used variant %s must be evicted", keys[0])
|
|
}
|
|
|
|
if n := countRows(t, cache,
|
|
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, string(keys[0]),
|
|
); n != 0 {
|
|
t.Errorf("evicted variant %s still has %d accounting rows", keys[0], n)
|
|
}
|
|
|
|
for _, key := range keys[1:] {
|
|
if !cache.variants.Exists(key) {
|
|
t.Errorf("more recently used variant %s must survive eviction", key)
|
|
}
|
|
}
|
|
|
|
assertNoDanglingReferences(t, cache)
|
|
}
|
|
|
|
func TestEvictionKeepsEverythingWhenUnderLimit(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1<<30)
|
|
|
|
content := bytes.Repeat([]byte{0xDF}, 800)
|
|
keys := []imagerequest.SourceKey{
|
|
storeEvictionTestSource(t, cache, "src.example.com", "/a.jpg", content),
|
|
storeEvictionTestSource(t, cache, "src.example.com", "/b.jpg", content),
|
|
}
|
|
|
|
storeEvictionTestVariant(t, cache, testVariantKeyOne,
|
|
bytes.Repeat([]byte{0xE0}, 500))
|
|
|
|
err := cache.EvictToLimit(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("EvictToLimit failed: %v", err)
|
|
}
|
|
|
|
for _, key := range keys {
|
|
if !cache.srcContent.Exists(key) {
|
|
t.Errorf("source %s must not be evicted while usage is under the limit", key)
|
|
}
|
|
|
|
if n := countRows(t, cache,
|
|
`SELECT COUNT(*) FROM source_content WHERE source_key = ?`, string(key),
|
|
); n != 1 {
|
|
t.Errorf("source %s has %d source_content rows, want 1", key, n)
|
|
}
|
|
}
|
|
|
|
if !cache.variants.Exists(testVariantKeyOne) {
|
|
t.Error("variant must not be evicted while usage is under the limit")
|
|
}
|
|
|
|
assertNoDanglingReferences(t, cache)
|
|
}
|
|
|
|
func TestZeroMaxBytesDisablesDiskCache(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, tmpDir := newEvictionTestCache(t, 0)
|
|
|
|
req := &imagerequest.Request{
|
|
Source: imagerequest.Source{
|
|
Host: "src.example.com",
|
|
Path: "/a.jpg",
|
|
},
|
|
Transform: imagerequest.Transform{
|
|
Format: imagerequest.FormatJPEG,
|
|
Quality: 85,
|
|
FitMode: imagerequest.FitCover,
|
|
},
|
|
}
|
|
|
|
assertDisabledCacheWritesAreNoOps(t, cache, req)
|
|
assertDisabledCacheReadsAlwaysMiss(t, cache, req)
|
|
assertDisabledCacheTracksNothing(t, cache)
|
|
assertDisabledCacheWroteNothingToDisk(t, tmpDir)
|
|
}
|
|
|
|
// assertDisabledCacheWritesAreNoOps verifies that stores against a
|
|
// disabled cache report success without recording anything.
|
|
func assertDisabledCacheWritesAreNoOps(
|
|
t *testing.T, cache *Cache, req *imagerequest.Request,
|
|
) {
|
|
t.Helper()
|
|
|
|
ctx := t.Context()
|
|
|
|
err := cache.StoreVariant(
|
|
ctx, imagerequest.CacheKey(req), bytes.NewReader([]byte("data")), "image/webp",
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("StoreVariant on disabled cache must be a no-op, got error: %v", err)
|
|
}
|
|
|
|
result := &httpfetcher.FetchResult{
|
|
StatusCode: 200,
|
|
ContentType: testContentTypeJPEG,
|
|
ContentLength: 4,
|
|
Headers: map[string][]string{},
|
|
}
|
|
|
|
key, err := cache.StoreSource(ctx, req, bytes.NewReader([]byte("data")), result)
|
|
if err != nil {
|
|
t.Fatalf("StoreSource on disabled cache must be a no-op, got error: %v", err)
|
|
}
|
|
|
|
if key != "" {
|
|
t.Errorf("StoreSource on disabled cache returned key %q, want empty", key)
|
|
}
|
|
}
|
|
|
|
// assertDisabledCacheReadsAlwaysMiss verifies that lookups against a
|
|
// disabled cache never report a hit.
|
|
func assertDisabledCacheReadsAlwaysMiss(
|
|
t *testing.T, cache *Cache, req *imagerequest.Request,
|
|
) {
|
|
t.Helper()
|
|
|
|
ctx := t.Context()
|
|
|
|
lookup, err := cache.Lookup(ctx, req)
|
|
if err != nil {
|
|
t.Fatalf("Lookup on disabled cache failed: %v", err)
|
|
}
|
|
|
|
if lookup.Hit {
|
|
t.Error("Lookup on disabled cache must always miss")
|
|
}
|
|
|
|
key, srcType, err := cache.LookupSource(ctx, req)
|
|
if err != nil {
|
|
t.Fatalf("LookupSource on disabled cache failed: %v", err)
|
|
}
|
|
|
|
if key != "" || srcType != "" {
|
|
t.Errorf("LookupSource on disabled cache = (%q, %q), want empty",
|
|
key, srcType)
|
|
}
|
|
}
|
|
|
|
// assertDisabledCacheTracksNothing verifies that a disabled cache
|
|
// records no usage and writes no accounting rows.
|
|
func assertDisabledCacheTracksNothing(t *testing.T, cache *Cache) {
|
|
t.Helper()
|
|
|
|
usage, err := cache.UsageBytes(t.Context())
|
|
if err != nil {
|
|
t.Fatalf("UsageBytes failed: %v", err)
|
|
}
|
|
|
|
if usage != 0 {
|
|
t.Errorf("UsageBytes on disabled cache = %d, want 0", usage)
|
|
}
|
|
|
|
if n := countRows(t, cache, `SELECT COUNT(*) FROM source_content`); n != 0 {
|
|
t.Errorf("disabled cache wrote %d source_content rows, want 0", n)
|
|
}
|
|
}
|
|
|
|
// assertDisabledCacheWroteNothingToDisk verifies that a disabled cache
|
|
// creates neither the cache directory tree nor any file under stateDir.
|
|
func assertDisabledCacheWroteNothingToDisk(t *testing.T, stateDir string) {
|
|
t.Helper()
|
|
|
|
_, err := os.Stat(filepath.Join(stateDir, "cache"))
|
|
if !os.IsNotExist(err) {
|
|
t.Errorf("disabled cache must not create the cache directory tree "+
|
|
"(stat err=%v)", err)
|
|
}
|
|
|
|
var foundFiles []string
|
|
|
|
walkErr := filepath.WalkDir(stateDir,
|
|
func(path string, d fs.DirEntry, err error) error {
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if !d.IsDir() {
|
|
foundFiles = append(foundFiles, path)
|
|
}
|
|
|
|
return nil
|
|
})
|
|
if walkErr != nil {
|
|
t.Fatalf("failed to walk state dir: %v", walkErr)
|
|
}
|
|
|
|
if len(foundFiles) != 0 {
|
|
t.Errorf("disabled cache wrote files to disk: %v", foundFiles)
|
|
}
|
|
}
|
|
|
|
func TestEvictionRunsUnderWritePressure(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
const limit = 1500
|
|
|
|
cache, _ := newEvictionTestCache(t, limit)
|
|
|
|
// An interval far longer than the test ensures only write
|
|
// pressure can trigger eviction here.
|
|
cache.StartEviction(time.Hour)
|
|
defer func() { _ = cache.StopEviction(t.Context()) }()
|
|
|
|
keys := []imagerequest.VariantKey{
|
|
testVariantKeyOne, testVariantKeyTwo, testVariantKeyThree,
|
|
}
|
|
fills := []byte{0x11, 0x12, 0x13}
|
|
|
|
for i, key := range keys {
|
|
storeEvictionTestVariant(t, cache, key,
|
|
bytes.Repeat([]byte{fills[i]}, 1000))
|
|
}
|
|
|
|
usage := waitForUsageAtOrBelow(t, cache, limit, 5*time.Second)
|
|
if usage > limit {
|
|
t.Errorf("write pressure did not trigger eviction: usage = %d, want <= %d",
|
|
usage, limit)
|
|
}
|
|
|
|
assertNoDanglingReferences(t, cache)
|
|
}
|
|
|
|
// TestEvictionRunsOnPeriodicSchedule writes three variant files straight
|
|
// to disk, bypassing StoreVariant, so they have no accounting rows and no
|
|
// write-pressure notification fires. Only a periodic reconciliation pass
|
|
// can then adopt them, and only the eviction pass that follows it can
|
|
// evict them.
|
|
func TestEvictionRunsOnPeriodicSchedule(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
const limit = 1500
|
|
|
|
cache, _ := newEvictionTestCache(t, limit)
|
|
|
|
// Hold the test database's only connection, so the startup pass
|
|
// waits for it after walking the still empty variant directory: the
|
|
// files written while it waits are first seen by a periodic pass.
|
|
conn, err := cache.db.Conn(t.Context())
|
|
if err != nil {
|
|
t.Fatalf("failed to take the database connection: %v", err)
|
|
}
|
|
|
|
defer func() { _ = conn.Close() }()
|
|
|
|
cache.StartEviction(100 * time.Millisecond)
|
|
defer func() { _ = cache.StopEviction(t.Context()) }()
|
|
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
|
|
for cache.db.Stats().WaitCount == 0 {
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("the startup pass never waited for the database")
|
|
}
|
|
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
|
|
keys := []imagerequest.VariantKey{
|
|
testVariantKeyOne, testVariantKeyTwo, testVariantKeyThree,
|
|
}
|
|
fills := []byte{0x21, 0x22, 0x23}
|
|
|
|
for i, key := range keys {
|
|
content := bytes.Repeat([]byte{fills[i]}, 1000)
|
|
|
|
_, err = cache.variants.Store(key, bytes.NewReader(content), "image/webp")
|
|
if err != nil {
|
|
t.Fatalf("failed to store variant file: %v", err)
|
|
}
|
|
}
|
|
|
|
_ = conn.Close()
|
|
|
|
// Only one of the 1000-byte files fits under the limit: wait until
|
|
// the evictor has removed the other two.
|
|
stored := len(keys)
|
|
deadline = time.Now().Add(5 * time.Second)
|
|
|
|
for stored > 1 && time.Now().Before(deadline) {
|
|
time.Sleep(25 * time.Millisecond)
|
|
|
|
stored = 0
|
|
|
|
for _, key := range keys {
|
|
if cache.variants.Exists(key) {
|
|
stored++
|
|
}
|
|
}
|
|
}
|
|
|
|
if stored > 1 {
|
|
t.Fatalf("periodic schedule did not trigger eviction: %d of %d "+
|
|
"variant files still on disk, want at most 1", stored, len(keys))
|
|
}
|
|
|
|
usage, err := cache.UsageBytes(t.Context())
|
|
if err != nil {
|
|
t.Fatalf("UsageBytes failed: %v", err)
|
|
}
|
|
|
|
if usage > limit {
|
|
t.Errorf("usage after eviction = %d, want <= %d", usage, limit)
|
|
}
|
|
|
|
assertNoDanglingReferences(t, cache)
|
|
}
|
|
|
|
func TestStartEvictionReconcilesAccountingWithDisk(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1<<30)
|
|
|
|
// An untracked variant file on disk (e.g. written before this
|
|
// feature existed) must be adopted into the accounting.
|
|
untracked := bytes.Repeat([]byte{0x31}, 1000)
|
|
|
|
_, err := cache.variants.Store(
|
|
testVariantKeyOne, bytes.NewReader(untracked), "image/webp",
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("failed to store untracked variant file: %v", err)
|
|
}
|
|
|
|
// An accounting row whose file is missing must be dropped.
|
|
_, err = cache.db.ExecContext(t.Context(),
|
|
`INSERT INTO variant_content (cache_key, size_bytes, content_type)
|
|
VALUES (?, ?, ?)`,
|
|
"deadbeef0001", 700, "image/webp",
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("failed to insert stale variant accounting row: %v", err)
|
|
}
|
|
|
|
cache.StartEviction(time.Hour)
|
|
defer func() { _ = cache.StopEviction(t.Context()) }()
|
|
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
|
|
var usage int64
|
|
|
|
for time.Now().Before(deadline) {
|
|
var err error
|
|
|
|
usage, err = cache.UsageBytes(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("UsageBytes failed: %v", err)
|
|
}
|
|
|
|
if usage == 1000 {
|
|
break
|
|
}
|
|
|
|
time.Sleep(25 * time.Millisecond)
|
|
}
|
|
|
|
if usage != 1000 {
|
|
t.Errorf("usage after reconciliation = %d, want 1000 "+
|
|
"(untracked file adopted, stale row dropped)", usage)
|
|
}
|
|
|
|
if n := countRows(t, cache,
|
|
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`,
|
|
string(testVariantKeyOne),
|
|
); n != 1 {
|
|
t.Errorf("untracked variant file was not adopted into accounting (rows=%d)", n)
|
|
}
|
|
|
|
if n := countRows(t, cache,
|
|
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, "deadbeef0001",
|
|
); n != 0 {
|
|
t.Errorf("stale accounting row without a file was not dropped (rows=%d)", n)
|
|
}
|
|
}
|
|
|
|
// TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup proves
|
|
// reconciliation is not a one-shot startup-only pass: it must also run
|
|
// on the periodic ticker, so a variant file that lands on disk with no
|
|
// accounting row well after startup (e.g. because StoreVariant's
|
|
// best-effort accounting insert failed under transient contention, or
|
|
// any other cause of an untracked file appearing during steady-state
|
|
// operation) is still adopted into accounting eventually, rather than
|
|
// staying invisible to UsageBytes/EvictToLimit until the next process
|
|
// restart.
|
|
func TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1<<30)
|
|
|
|
// Hold the test database's only connection, so the startup pass
|
|
// waits for it after walking the still empty variant directory: the
|
|
// file written while it waits is first seen by a periodic pass.
|
|
conn, err := cache.db.Conn(t.Context())
|
|
if err != nil {
|
|
t.Fatalf("failed to take the database connection: %v", err)
|
|
}
|
|
|
|
defer func() { _ = conn.Close() }()
|
|
|
|
cache.StartEviction(100 * time.Millisecond)
|
|
defer func() { _ = cache.StopEviction(t.Context()) }()
|
|
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
|
|
for cache.db.Stats().WaitCount == 0 {
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("the startup pass never waited for the database")
|
|
}
|
|
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
|
|
// Simulate a variant whose accounting insert failed after the
|
|
// process was already running and serving requests: the content
|
|
// file is written directly, bypassing StoreVariant's (and thus its
|
|
// accounting insert) entirely, exactly as would happen if that
|
|
// insert had failed and only the file write had succeeded.
|
|
untracked := bytes.Repeat([]byte{0x41}, 900)
|
|
|
|
_, err = cache.variants.Store(
|
|
"aabbccdd0099", bytes.NewReader(untracked), "image/webp",
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("failed to store untracked variant file: %v", err)
|
|
}
|
|
|
|
_ = conn.Close()
|
|
|
|
deadline = time.Now().Add(5 * time.Second)
|
|
|
|
var usage int64
|
|
|
|
for time.Now().Before(deadline) {
|
|
var err error
|
|
|
|
usage, err = cache.UsageBytes(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("UsageBytes failed: %v", err)
|
|
}
|
|
|
|
if usage == 900 {
|
|
break
|
|
}
|
|
|
|
time.Sleep(25 * time.Millisecond)
|
|
}
|
|
|
|
if usage != 900 {
|
|
t.Errorf("usage after periodic reconciliation = %d, want 900 "+
|
|
"(a file that appeared after startup reconciliation already ran must still "+
|
|
"be adopted by a later periodic pass)", usage)
|
|
}
|
|
|
|
if n := countRows(t, cache,
|
|
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, "aabbccdd0099",
|
|
); n != 1 {
|
|
t.Errorf("file that appeared after startup was not adopted by periodic "+
|
|
"reconciliation (rows=%d)", n)
|
|
}
|
|
}
|
|
|
|
// 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
|
|
// connection to come free, and that the stop logs one warning: the
|
|
// interrupted reconciliation's, with no eviction pass started after it.
|
|
func TestStopEvictionInterruptsPassInProgress(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1<<30)
|
|
|
|
var logBuf bytes.Buffer
|
|
|
|
cache.log = slog.New(slog.NewJSONHandler(&logBuf, nil))
|
|
|
|
conn, err := cache.db.Conn(t.Context())
|
|
if err != nil {
|
|
t.Fatalf("failed to take the database connection: %v", err)
|
|
}
|
|
|
|
defer func() { _ = conn.Close() }()
|
|
|
|
cache.StartEviction(time.Hour)
|
|
|
|
// The pass is in progress once it waits for the connection.
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
|
|
for cache.db.Stats().WaitCount == 0 {
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("the reconciliation pass never waited for the database")
|
|
}
|
|
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
|
|
defer cancel()
|
|
|
|
err = cache.StopEviction(ctx)
|
|
t.Logf("StopEviction() error = %v", err)
|
|
|
|
if err != nil {
|
|
t.Fatalf("StopEviction() error = %v, want nil: the pass waiting for "+
|
|
"the database did not stop", err)
|
|
}
|
|
|
|
t.Logf("log output: %s", logBuf.String())
|
|
|
|
warnings := strings.Count(logBuf.String(), `"level":"WARN"`)
|
|
if warnings != 1 {
|
|
t.Errorf("the stop logged %d warnings, want 1", warnings)
|
|
}
|
|
}
|
|
|
|
// TestEvictToLimitStopsAtNextCandidateOnceCancelled cancels the context
|
|
// once the oldest of three sources is evicted, while the next one waits
|
|
// for the database, and checks that EvictToLimit then returns
|
|
// context.Canceled, with that source gone and the other two kept, and
|
|
// logs no warning. The test database has one connection. When the
|
|
// deletion of the oldest source's row commits, a commit hook holds the
|
|
// deletion until the test waits for the connection, so the test takes it
|
|
// as soon as the deletion is done. The eviction of the next source then
|
|
// waits for the connection, and the test cancels the context while it
|
|
// waits.
|
|
func TestEvictToLimitStopsAtNextCandidateOnceCancelled(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1)
|
|
|
|
var logBuf bytes.Buffer
|
|
|
|
cache.log = slog.New(slog.NewJSONHandler(&logBuf, nil))
|
|
|
|
keys := []imagerequest.SourceKey{
|
|
storeEvictionTestSource(t, cache, "cancel.example.com", "/a.jpg",
|
|
bytes.Repeat([]byte{0x61}, 1000)),
|
|
storeEvictionTestSource(t, cache, "cancel.example.com", "/b.jpg",
|
|
bytes.Repeat([]byte{0x62}, 1000)),
|
|
storeEvictionTestSource(t, cache, "cancel.example.com", "/c.jpg",
|
|
bytes.Repeat([]byte{0x63}, 1000)),
|
|
}
|
|
|
|
base := time.Now().Add(-time.Hour)
|
|
|
|
for i, key := range keys {
|
|
setSourceLastAccessed(t, cache, key, base.Add(time.Duration(i)*time.Minute))
|
|
}
|
|
|
|
waitsBefore := cache.db.Stats().WaitCount
|
|
oldestRowDeleted := make(chan struct{})
|
|
|
|
var oldestRowDeletedOnce sync.Once
|
|
|
|
setCommitHook(t, cache, func() int32 {
|
|
oldestRowDeletedOnce.Do(func() {
|
|
close(oldestRowDeleted)
|
|
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
|
|
for cache.db.Stats().WaitCount == waitsBefore && time.Now().Before(deadline) {
|
|
time.Sleep(time.Millisecond)
|
|
}
|
|
})
|
|
|
|
return 0
|
|
})
|
|
|
|
ctx, cancel := context.WithCancel(t.Context())
|
|
defer cancel()
|
|
|
|
evicted := make(chan error, 1)
|
|
|
|
go func() { evicted <- cache.EvictToLimit(ctx) }()
|
|
|
|
select {
|
|
case <-oldestRowDeleted:
|
|
case err := <-evicted:
|
|
t.Fatalf("EvictToLimit() returned %v without deleting a row", err)
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("EvictToLimit never deleted a row")
|
|
}
|
|
|
|
releaseDatabase := holdDatabase(t, cache)
|
|
defer releaseDatabase()
|
|
|
|
waitForConnectionWaits(t, cache.db, waitsBefore+2)
|
|
cancel()
|
|
|
|
err := <-evicted
|
|
t.Logf("EvictToLimit() error = %v", err)
|
|
t.Logf("log output: %s", logBuf.String())
|
|
|
|
// The checks below read the database, so the connection goes back.
|
|
releaseDatabase()
|
|
|
|
setCommitHook(t, cache, nil)
|
|
|
|
if !errors.Is(err, context.Canceled) {
|
|
t.Errorf("EvictToLimit() error = %v, want context.Canceled", err)
|
|
}
|
|
|
|
assertOnlyOldestSourceEvicted(t, cache, keys)
|
|
|
|
if strings.Contains(logBuf.String(), `"level":"WARN"`) {
|
|
t.Errorf("EvictToLimit logged a warning after the context was cancelled")
|
|
}
|
|
}
|
|
|
|
// setCommitHook makes the SQLite driver call hook each time a write on
|
|
// the test database's one connection commits, or stops it when hook is
|
|
// nil.
|
|
func setCommitHook(t *testing.T, cache *Cache, hook sqlite.CommitHookFn) {
|
|
t.Helper()
|
|
|
|
conn, err := cache.db.Conn(t.Context())
|
|
if err != nil {
|
|
t.Fatalf("failed to take the database connection: %v", err)
|
|
}
|
|
|
|
defer func() { _ = conn.Close() }()
|
|
|
|
err = conn.Raw(func(driverConn any) error {
|
|
hooks, ok := driverConn.(sqlite.HookRegisterer)
|
|
if !ok {
|
|
t.Fatalf("the database connection is a %T, which takes no hooks",
|
|
driverConn)
|
|
}
|
|
|
|
hooks.RegisterCommitHook(hook)
|
|
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("failed to set the commit hook: %v", err)
|
|
}
|
|
}
|
|
|
|
// assertOnlyOldestSourceEvicted checks that the first of keys has neither
|
|
// its file nor its row, and that every other one has both.
|
|
func assertOnlyOldestSourceEvicted(
|
|
t *testing.T, cache *Cache, keys []imagerequest.SourceKey,
|
|
) {
|
|
t.Helper()
|
|
|
|
if cache.srcContent.Exists(keys[0]) {
|
|
t.Errorf("source %s, evicted when the context was cancelled, "+
|
|
"is still on disk", keys[0])
|
|
}
|
|
|
|
for _, key := range keys[1:] {
|
|
if !cache.srcContent.Exists(key) {
|
|
t.Errorf("source %s was evicted after the context was cancelled", key)
|
|
}
|
|
|
|
if n := countRows(t, cache,
|
|
`SELECT COUNT(*) FROM source_content WHERE source_key = ?`, string(key),
|
|
); n != 1 {
|
|
t.Errorf("source %s, kept, has %d source_content rows, want 1", key, n)
|
|
}
|
|
}
|
|
|
|
assertNoDanglingReferences(t, cache)
|
|
}
|
|
|
|
// TestStopEvictionReturnsWhenItsContextEnds pauses the startup
|
|
// reconciliation pass in the hook called after each of its reads, where
|
|
// cancellation cannot reach it, and checks that StopEviction returns an
|
|
// error wrapping its context's error when that context ends, instead of
|
|
// waiting for the pass. Once the pass goes on, the goroutine exits, and a
|
|
// second StopEviction returns nil.
|
|
func TestStopEvictionReturnsWhenItsContextEnds(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1<<30)
|
|
|
|
paused := make(chan struct{})
|
|
resume := make(chan struct{})
|
|
|
|
var pauseOnce sync.Once
|
|
|
|
cache.reconciliationReadTestHook = func(int) {
|
|
pauseOnce.Do(func() {
|
|
close(paused)
|
|
<-resume
|
|
})
|
|
}
|
|
|
|
cache.StartEviction(time.Hour)
|
|
|
|
select {
|
|
case <-paused:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("the reconciliation pass never made a read")
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
|
|
defer cancel()
|
|
|
|
err := cache.StopEviction(ctx)
|
|
t.Logf("StopEviction() error = %v", err)
|
|
|
|
if !errors.Is(err, context.DeadlineExceeded) {
|
|
t.Errorf("StopEviction() error = %v, want context.DeadlineExceeded", err)
|
|
}
|
|
|
|
close(resume)
|
|
|
|
err = cache.StopEviction(t.Context())
|
|
if err != nil {
|
|
t.Fatalf("second StopEviction() error = %v, want nil", err)
|
|
}
|
|
}
|
|
|
|
// TestReconciliationWalksStopOnceCancelled checks that both directory
|
|
// walks of a reconciliation pass return the context's error once it is
|
|
// cancelled, leaving in place a stale temp file they would otherwise
|
|
// remove.
|
|
func TestReconciliationWalksStopOnceCancelled(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1<<30)
|
|
|
|
ctx, cancel := context.WithCancel(t.Context())
|
|
cancel()
|
|
|
|
staleTime := time.Now().Add(-2 * staleTempFileAge)
|
|
|
|
walks := map[string]func(context.Context) error{
|
|
cache.variants.baseDir: cache.reconcileVariantFiles,
|
|
cache.srcContent.baseDir: cache.reconcileSourceFiles,
|
|
}
|
|
|
|
for dir, walk := range walks {
|
|
tempFile := filepath.Join(dir, tempFilePrefix+"stale")
|
|
|
|
err := os.WriteFile(tempFile, []byte("partial"), 0o600)
|
|
if err != nil {
|
|
t.Fatalf("failed to write temp file: %v", err)
|
|
}
|
|
|
|
err = os.Chtimes(tempFile, staleTime, staleTime)
|
|
if err != nil {
|
|
t.Fatalf("failed to backdate temp file: %v", err)
|
|
}
|
|
|
|
err = walk(ctx)
|
|
t.Logf("walk of %s: error = %v", dir, err)
|
|
|
|
if !errors.Is(err, context.Canceled) {
|
|
t.Errorf("walk of %s: error = %v, want context.Canceled", dir, err)
|
|
}
|
|
|
|
_, err = os.Stat(tempFile)
|
|
if err != nil {
|
|
t.Errorf("walk of %s went on after cancellation: %v", dir, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestReconciliationPassLogsNoWarningOnceCancelled checks that a
|
|
// reconciliation pass run with an already cancelled context logs no
|
|
// warning, so a periodic tick the loop takes after a stop adds no
|
|
// warning to the one from the pass the stop interrupted.
|
|
func TestReconciliationPassLogsNoWarningOnceCancelled(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1<<30)
|
|
|
|
var logBuf bytes.Buffer
|
|
|
|
cache.log = slog.New(slog.NewJSONHandler(&logBuf, nil))
|
|
|
|
ctx, cancel := context.WithCancel(t.Context())
|
|
cancel()
|
|
|
|
cache.runReconciliationPass(ctx)
|
|
t.Logf("log output: %s", logBuf.String())
|
|
|
|
if strings.Contains(logBuf.String(), `"level":"WARN"`) {
|
|
t.Errorf("runReconciliationPass logged a warning with a cancelled context")
|
|
}
|
|
}
|