clawbot
1616e91a6a
check / check (push) Failing after 2s
StartEviction runs the eviction goroutine with its own context, which StopEviction cancels in place of the old stop channel, so a pass in progress stops at its next database call, file, row or eviction candidate instead of running to completion, and no new pass starts. StopEviction takes a context: when it ends before the goroutine exits, StopEviction stops waiting and returns an error wrapping it. The handlers' stop hook passes fx's stop context, so an eviction still running at fx's stop deadline fails the stop and the exit code is 1. A stop logs at most one warning. Model: opus-5-5
1225 行
33 KiB
Go
1225 行
33 KiB
Go
package imgcache
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"io/fs"
|
|
"log/slog"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
_ "modernc.org/sqlite"
|
|
"sneak.berlin/go/pixa/internal/database"
|
|
"sneak.berlin/go/pixa/internal/httpfetcher"
|
|
)
|
|
|
|
// 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 VariantKey = "aabbccdd0001"
|
|
testVariantKeyTwo VariantKey = "aabbccdd0002"
|
|
testVariantKeyThree VariantKey = "aabbccdd0003"
|
|
testVariantKeyFour 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,
|
|
CacheTTL: time.Hour,
|
|
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 content hash.
|
|
func storeEvictionTestSource(
|
|
t *testing.T, cache *Cache, host, path string, content []byte,
|
|
) ContentHash {
|
|
t.Helper()
|
|
|
|
req := &ImageRequest{
|
|
SourceHost: host,
|
|
SourcePath: path,
|
|
Format: FormatJPEG,
|
|
Quality: 85,
|
|
FitMode: FitCover,
|
|
}
|
|
|
|
result := &httpfetcher.FetchResult{
|
|
StatusCode: 200,
|
|
ContentType: testContentTypeJPEG,
|
|
ContentLength: int64(len(content)),
|
|
Headers: map[string][]string{
|
|
testHeaderContentType: {testContentTypeJPEG},
|
|
},
|
|
}
|
|
|
|
hash, err := cache.StoreSource(
|
|
context.Background(), req, bytes.NewReader(content), result,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("StoreSource(%s%s) failed: %v", host, path, err)
|
|
}
|
|
|
|
return hash
|
|
}
|
|
|
|
// storeEvictionTestVariant stores content as a processed variant under
|
|
// the given cache key.
|
|
func storeEvictionTestVariant(
|
|
t *testing.T, cache *Cache, key 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 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, hash ContentHash, when time.Time,
|
|
) {
|
|
t.Helper()
|
|
|
|
res, err := cache.db.ExecContext(t.Context(),
|
|
`UPDATE source_content SET last_accessed_at = ? WHERE content_hash = ?`,
|
|
when.UTC().Format(sqliteTimestampFormat), string(hash),
|
|
)
|
|
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)", hash, 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 content_hash FROM source_metadata
|
|
WHERE content_hash IS NOT NULL AND content_hash != ''`,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("failed to query source_metadata: %v", err)
|
|
}
|
|
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
for rows.Next() {
|
|
var hash string
|
|
|
|
err := rows.Scan(&hash)
|
|
if err != nil {
|
|
t.Fatalf("failed to scan content_hash: %v", err)
|
|
}
|
|
|
|
if !cache.srcContent.Exists(ContentHash(hash)) {
|
|
t.Errorf("source_metadata references content %s but the file is missing",
|
|
hash)
|
|
}
|
|
}
|
|
|
|
err = rows.Err()
|
|
if err != nil {
|
|
t.Fatalf("source_metadata 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(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 TestUsageBytesCountsMultiReferencedBlobOnce(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1<<30)
|
|
|
|
content := bytes.Repeat([]byte{0xCC}, 1200)
|
|
|
|
hashOne := storeEvictionTestSource(t, cache,
|
|
"src.example.com", "/one.jpg", content)
|
|
hashTwo := storeEvictionTestSource(t, cache,
|
|
"src.example.com", "/two.jpg", content)
|
|
|
|
if hashOne != hashTwo {
|
|
t.Fatalf("identical content produced different hashes: %s vs %s",
|
|
hashOne, hashTwo)
|
|
}
|
|
|
|
usage, err := cache.UsageBytes(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("UsageBytes failed: %v", err)
|
|
}
|
|
|
|
if usage != 1200 {
|
|
t.Errorf("UsageBytes = %d, want 1200 (deduplicated blob counted once)", usage)
|
|
}
|
|
}
|
|
|
|
func TestEvictToLimitEvictsLeastRecentlyUsedFirst(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
const limit = 3000
|
|
|
|
cache, _ := newEvictionTestCache(t, limit)
|
|
|
|
now := time.Now()
|
|
keys := []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 TestEvictionRemovesMultiReferencedBlobTogetherWithAllReferences(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
const limit = 1000
|
|
|
|
cache, _ := newEvictionTestCache(t, limit)
|
|
|
|
now := time.Now()
|
|
|
|
// One 800-byte blob referenced by two source paths.
|
|
sharedContent := bytes.Repeat([]byte{0xDD}, 800)
|
|
sharedHash := storeEvictionTestSource(t, cache,
|
|
"src.example.com", "/a.jpg", sharedContent)
|
|
|
|
h := storeEvictionTestSource(t, cache,
|
|
"src.example.com", "/b.jpg", sharedContent)
|
|
if h != sharedHash {
|
|
t.Fatalf("identical content produced different hashes: %s vs %s",
|
|
h, sharedHash)
|
|
}
|
|
|
|
// A newer 600-byte blob referenced by one source path.
|
|
recentHash := storeEvictionTestSource(t, cache, "src.example.com", "/c.jpg",
|
|
bytes.Repeat([]byte{0xEE}, 600))
|
|
|
|
setSourceLastAccessed(t, cache, sharedHash, now.Add(-2*time.Hour))
|
|
setSourceLastAccessed(t, cache, recentHash, now.Add(-time.Minute))
|
|
|
|
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)
|
|
}
|
|
|
|
// The multi-referenced blob must be gone from disk, from
|
|
// source_content, and from BOTH source_metadata rows: references
|
|
// are removed together with the blob, never left dangling.
|
|
if cache.srcContent.Exists(sharedHash) {
|
|
t.Errorf("evicted blob %s still exists on disk", sharedHash)
|
|
}
|
|
|
|
if n := countRows(t, cache,
|
|
`SELECT COUNT(*) FROM source_content WHERE content_hash = ?`, string(sharedHash),
|
|
); n != 0 {
|
|
t.Errorf("evicted blob %s still has %d source_content rows", sharedHash, n)
|
|
}
|
|
|
|
if n := countRows(t, cache,
|
|
`SELECT COUNT(*) FROM source_metadata WHERE content_hash = ?`, string(sharedHash),
|
|
); n != 0 {
|
|
t.Errorf("evicted blob %s still has %d source_metadata references", sharedHash, n)
|
|
}
|
|
|
|
// The JSON metadata sidecars for both referencing paths must be
|
|
// removed along with the rows.
|
|
for _, path := range []string{"/a.jpg", "/b.jpg"} {
|
|
pathHash := HashPath(path + "?")
|
|
if cache.srcMetadata.Exists("src.example.com", pathHash) {
|
|
t.Errorf("metadata sidecar for %s must be removed with its row", path)
|
|
}
|
|
}
|
|
|
|
// The more recently used blob survives fully intact.
|
|
if !cache.srcContent.Exists(recentHash) {
|
|
t.Errorf("recently used blob %s must survive eviction", recentHash)
|
|
}
|
|
|
|
if n := countRows(t, cache,
|
|
`SELECT COUNT(*) FROM source_metadata WHERE content_hash = ?`, string(recentHash),
|
|
); n != 1 {
|
|
t.Errorf("recently used blob %s has %d source_metadata rows, want 1",
|
|
recentHash, n)
|
|
}
|
|
|
|
assertNoDanglingReferences(t, cache)
|
|
}
|
|
|
|
func TestEvictionKeepsEverythingWhenUnderLimit(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1<<30)
|
|
|
|
content := bytes.Repeat([]byte{0xDF}, 800)
|
|
hash := storeEvictionTestSource(t, cache, "src.example.com", "/a.jpg", content)
|
|
|
|
h := storeEvictionTestSource(t, cache, "src.example.com", "/b.jpg", content)
|
|
if h != hash {
|
|
t.Fatalf("identical content produced different hashes: %s vs %s", h, hash)
|
|
}
|
|
|
|
storeEvictionTestVariant(t, cache, testVariantKeyOne,
|
|
bytes.Repeat([]byte{0xE0}, 500))
|
|
|
|
err := cache.EvictToLimit(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("EvictToLimit failed: %v", err)
|
|
}
|
|
|
|
if !cache.srcContent.Exists(hash) {
|
|
t.Errorf("blob %s must not be evicted while usage is under the limit", hash)
|
|
}
|
|
|
|
if n := countRows(t, cache,
|
|
`SELECT COUNT(*) FROM source_metadata WHERE content_hash = ?`, string(hash),
|
|
); n != 2 {
|
|
t.Errorf("blob %s has %d source_metadata rows, want 2", hash, 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{
|
|
SourceHost: "src.example.com",
|
|
SourcePath: "/a.jpg",
|
|
Format: FormatJPEG,
|
|
Quality: 85,
|
|
FitMode: 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,
|
|
) {
|
|
t.Helper()
|
|
|
|
ctx := t.Context()
|
|
|
|
err := cache.StoreVariant(
|
|
ctx, 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{},
|
|
}
|
|
|
|
hash, 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 hash != "" {
|
|
t.Errorf("StoreSource on disabled cache returned hash %q, want empty", hash)
|
|
}
|
|
}
|
|
|
|
// assertDisabledCacheReadsAlwaysMiss verifies that lookups against a
|
|
// disabled cache never report a hit.
|
|
func assertDisabledCacheReadsAlwaysMiss(
|
|
t *testing.T, cache *Cache, req *ImageRequest,
|
|
) {
|
|
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")
|
|
}
|
|
|
|
srcHash, srcType, err := cache.LookupSource(ctx, req)
|
|
if err != nil {
|
|
t.Fatalf("LookupSource on disabled cache failed: %v", err)
|
|
}
|
|
|
|
if srcHash != "" || srcType != "" {
|
|
t.Errorf("LookupSource on disabled cache = (%q, %q), want empty",
|
|
srcHash, 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)
|
|
}
|
|
|
|
if n := countRows(t, cache, `SELECT COUNT(*) FROM source_metadata`); n != 0 {
|
|
t.Errorf("disabled cache wrote %d source_metadata 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 := []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)
|
|
}
|
|
|
|
func TestEvictionRunsOnPeriodicSchedule(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
const limit = 1500
|
|
|
|
cache, _ := newEvictionTestCache(t, limit)
|
|
|
|
// Start the evictor while the cache is empty, then create tracked
|
|
// over-limit state WITHOUT going through the store methods, so no
|
|
// write-pressure notification fires and only the periodic ticker
|
|
// can trigger eviction.
|
|
cache.StartEviction(100 * time.Millisecond)
|
|
defer func() { _ = cache.StopEviction(t.Context()) }()
|
|
|
|
keys := []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)
|
|
}
|
|
|
|
_, err = cache.db.ExecContext(t.Context(),
|
|
`INSERT INTO variant_content (cache_key, size_bytes, content_type)
|
|
VALUES (?, ?, ?)`,
|
|
string(key), len(content), "image/webp",
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("failed to insert variant accounting row: %v", err)
|
|
}
|
|
}
|
|
|
|
usage := waitForUsageAtOrBelow(t, cache, limit, 5*time.Second)
|
|
if usage > limit {
|
|
t.Errorf("periodic schedule did not trigger eviction: usage = %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)
|
|
|
|
const interval = 100 * time.Millisecond
|
|
|
|
cache.StartEviction(interval)
|
|
defer func() { _ = cache.StopEviction(t.Context()) }()
|
|
|
|
// Let startup reconciliation run and settle on an empty cache
|
|
// before introducing the untracked file, so the adoption we assert
|
|
// below can only be the work of a later, periodic pass.
|
|
time.Sleep(3 * interval)
|
|
|
|
// 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)
|
|
}
|
|
|
|
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)
|
|
}
|
|
}
|
|
|
|
// 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 source blobs is being evicted, and checks that
|
|
// EvictToLimit then returns context.Canceled without evicting the other
|
|
// two or logging a warning for either of them.
|
|
func TestEvictToLimitStopsAtNextCandidateOnceCancelled(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1)
|
|
|
|
var logBuf bytes.Buffer
|
|
|
|
cache.log = slog.New(slog.NewJSONHandler(&logBuf, nil))
|
|
|
|
hashes := []ContentHash{
|
|
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, hash := range hashes {
|
|
setSourceLastAccessed(t, cache, hash, base.Add(time.Duration(i)*time.Minute))
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(t.Context())
|
|
defer cancel()
|
|
|
|
cache.evictSourceBlobTestHook = func(ContentHash) { cancel() }
|
|
|
|
err := cache.EvictToLimit(ctx)
|
|
t.Logf("EvictToLimit() error = %v", err)
|
|
t.Logf("log output: %s", logBuf.String())
|
|
|
|
if !errors.Is(err, context.Canceled) {
|
|
t.Errorf("EvictToLimit() error = %v, want context.Canceled", err)
|
|
}
|
|
|
|
if cache.srcContent.Exists(hashes[0]) {
|
|
t.Errorf("source blob %s, evicted when the context was cancelled, "+
|
|
"is still on disk", hashes[0])
|
|
}
|
|
|
|
for _, hash := range hashes[1:] {
|
|
if !cache.srcContent.Exists(hash) {
|
|
t.Errorf("source blob %s was evicted after the context was cancelled", hash)
|
|
}
|
|
}
|
|
|
|
if strings.Contains(logBuf.String(), `"level":"WARN"`) {
|
|
t.Errorf("EvictToLimit logged a warning after the context was cancelled")
|
|
}
|
|
|
|
assertNoDanglingReferences(t, cache)
|
|
}
|
|
|
|
// TestStopEvictionReturnsWhenItsContextEnds pauses an eviction pass where
|
|
// cancellation cannot reach it, after a source blob's rows are deleted and
|
|
// before its file is removed, and checks that StopEviction returns its
|
|
// context's error when that context ends instead of waiting for the pass.
|
|
// Once the pass goes on, the goroutine exits and no row points at a
|
|
// missing file.
|
|
func TestStopEvictionReturnsWhenItsContextEnds(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1)
|
|
|
|
paused := make(chan struct{})
|
|
resume := make(chan struct{})
|
|
|
|
cache.evictSourceBlobTestHook = func(ContentHash) {
|
|
close(paused)
|
|
<-resume
|
|
}
|
|
|
|
hash := storeEvictionTestSource(t, cache, "stop.example.com", "/a.jpg",
|
|
bytes.Repeat([]byte{0x61}, 1000))
|
|
|
|
cache.StartEviction(time.Hour)
|
|
|
|
select {
|
|
case <-paused:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("the eviction pass never reached the source blob")
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
assertNoDanglingReferences(t, cache)
|
|
|
|
if cache.srcContent.Exists(hash) {
|
|
t.Errorf("source blob %s is still on disk after its rows were deleted", hash)
|
|
}
|
|
}
|
|
|
|
// 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")
|
|
}
|
|
}
|
|
|
|
// TestEvictSourceBlobExcludesConcurrentStoreOfIdenticalContent exercises
|
|
// the exact TOCTOU window between evictSourceBlob's row-deletion
|
|
// transaction commit and its content file unlink: a concurrent
|
|
// StoreSource for a different source path whose content hashes to the
|
|
// same value (real SHA-256 dedup, not a contrived case) must not be
|
|
// able to insert a fresh row referencing the file while eviction is
|
|
// mid-unlink, and must not lose its own store once eviction has fully
|
|
// released the content hash.
|
|
func TestEvictSourceBlobExcludesConcurrentStoreOfIdenticalContent(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cache, _ := newEvictionTestCache(t, 1<<30)
|
|
ctx := context.Background()
|
|
|
|
content := bytes.Repeat([]byte{0x55}, 400)
|
|
|
|
hash := storeEvictionTestSource(t, cache,
|
|
"race.example.com", "/first.jpg", content)
|
|
|
|
proceed := make(chan struct{})
|
|
storeAttempted := make(chan struct{})
|
|
|
|
cache.evictSourceBlobTestHook = func(gotHash ContentHash) {
|
|
if gotHash != hash {
|
|
t.Errorf("test hook invoked for hash %s, want %s", gotHash, hash)
|
|
}
|
|
|
|
close(storeAttempted)
|
|
<-proceed
|
|
}
|
|
|
|
evictDone := make(chan error, 1)
|
|
|
|
go func() {
|
|
evictDone <- cache.evictSourceBlob(ctx, hash)
|
|
}()
|
|
|
|
// Wait until eviction has committed its delete transaction and is
|
|
// paused (inside the test hook) immediately before unlinking the
|
|
// content file: exactly the window the review flagged.
|
|
<-storeAttempted
|
|
|
|
storeDone := storeIdenticalContentConcurrently(ctx, cache, content)
|
|
|
|
// The concurrent store must not be able to complete while eviction
|
|
// still holds the content hash (i.e. before the file is unlinked):
|
|
// if it could, it would insert a row referencing a file about to be
|
|
// removed out from under it.
|
|
select {
|
|
case err := <-storeDone:
|
|
t.Fatalf("StoreSource for identical content completed (err=%v) while eviction "+
|
|
"still held the content hash open between commit and unlink; the store and "+
|
|
"the evict of identical content are not mutually exclusive", err)
|
|
case <-time.After(200 * time.Millisecond):
|
|
// Expected: the store is blocked behind eviction's exclusion.
|
|
}
|
|
|
|
close(proceed)
|
|
|
|
err := <-evictDone
|
|
if err != nil {
|
|
t.Fatalf("evictSourceBlob failed: %v", err)
|
|
}
|
|
|
|
err = <-storeDone
|
|
if err != nil {
|
|
t.Fatalf("StoreSource failed: %v", err)
|
|
}
|
|
|
|
assertNoDanglingReferences(t, cache)
|
|
|
|
dupHash, _, err := cache.LookupSource(ctx, &ImageRequest{
|
|
SourceHost: "race.example.com",
|
|
SourcePath: "/dup.jpg",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("LookupSource failed: %v", err)
|
|
}
|
|
|
|
if dupHash == "" {
|
|
t.Fatal("re-stored blob was lost: the store legitimately ran after eviction " +
|
|
"released the content hash and must have recreated the file and row")
|
|
}
|
|
|
|
if !cache.srcContent.Exists(dupHash) {
|
|
t.Errorf("source_content/source_metadata references %s but its file is missing",
|
|
dupHash)
|
|
}
|
|
}
|
|
|
|
// storeIdenticalContentConcurrently starts a StoreSource for a second
|
|
// source path whose body hashes to the same content hash, returning the
|
|
// channel its error is delivered on.
|
|
func storeIdenticalContentConcurrently(
|
|
ctx context.Context, cache *Cache, content []byte,
|
|
) chan error {
|
|
storeDone := make(chan error, 1)
|
|
|
|
go func() {
|
|
req := &ImageRequest{
|
|
SourceHost: "race.example.com",
|
|
SourcePath: "/dup.jpg",
|
|
Format: FormatJPEG,
|
|
Quality: 85,
|
|
FitMode: FitCover,
|
|
}
|
|
result := &httpfetcher.FetchResult{
|
|
StatusCode: 200,
|
|
ContentType: testContentTypeJPEG,
|
|
ContentLength: int64(len(content)),
|
|
Headers: map[string][]string{
|
|
testHeaderContentType: {testContentTypeJPEG},
|
|
},
|
|
}
|
|
|
|
_, err := cache.StoreSource(ctx, req, bytes.NewReader(content), result)
|
|
storeDone <- err
|
|
}()
|
|
|
|
return storeDone
|
|
}
|