sneak
061a3545eb
The size-accounting INSERT in StoreVariant used db.Exec, and StoreVariant took no context at all, so the write was uncancellable and untraceable. Rather than paper over that with context.Background() at the call site, StoreVariant now takes a context.Context and uses ExecContext. The plumbing is small: the only production caller is Service.processAndStore, which already has the request context in scope, so the accounting insert now shares the lifetime of the request that produced the variant. Test callers pass t.Context(). The insert remains best-effort: a failure is logged and the startup and periodic reconciliation passes still adopt any variant file whose accounting row is missing.
973 lines
26 KiB
Go
973 lines
26 KiB
Go
package imgcache
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"database/sql"
|
|
"io/fs"
|
|
"os"
|
|
"path/filepath"
|
|
"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 is the variant cache key reused across the eviction
|
|
// tests as the first stored variant.
|
|
const testVariantKeyOne VariantKey = "aabbccdd0001"
|
|
|
|
// 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, "aabbccdd0002",
|
|
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, "aabbccdd0002", "aabbccdd0003", "aabbccdd0004",
|
|
}
|
|
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 cache.StopEviction()
|
|
|
|
keys := []VariantKey{testVariantKeyOne, "aabbccdd0002", "aabbccdd0003"}
|
|
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 cache.StopEviction()
|
|
|
|
keys := []VariantKey{testVariantKeyOne, "aabbccdd0002", "aabbccdd0003"}
|
|
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 cache.StopEviction()
|
|
|
|
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 cache.StopEviction()
|
|
|
|
// 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)
|
|
}
|
|
}
|
|
|
|
// 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) {
|
|
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
|
|
}
|