sneak
e06192b7a5
check / check (push) Canceled after 0s
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, so TestReconciliationReadsAPageAtATime now counts 15 rows read, not 20. Two source URLs serving the same bytes store them twice. The tests of the removed sharing and lock go with them; the two eviction cancellation tests no longer use the hook. Model: opus-5-5
1215 lines
32 KiB
Go
1215 lines
32 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
|
|
// while the oldest of three sources is being evicted, 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) }()
|
|
|
|
<-oldestRowDeleted
|
|
|
|
conn, err := cache.db.Conn(t.Context())
|
|
if err != nil {
|
|
t.Fatalf("failed to take the database connection: %v", err)
|
|
}
|
|
|
|
defer func() { _ = conn.Close() }()
|
|
|
|
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.
|
|
_ = conn.Close()
|
|
|
|
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")
|
|
}
|
|
}
|