Check cached sources with upstream after the cache TTL (closes #69)
check / check (push) Canceled after 0s
check / check (push) Canceled after 0s
Cached images were served until eviction, so upstream changes were missed. After 24 hours a request checks the source with upstream, sending its stored ETag and Last-Modified. A 304 renews it with no download or conversion. New bytes replace it; older variants are converted again. A failed check serves the cached image and puts the URL in the negative cache. Images not checked for 30 days are evicted. Variant rows keep the source's ETag and Last-Modified for checking a variant of an evicted source. A variant file's modification time is its source's check time when it was made or renewed. Hits read it without a database call. The ETag includes it, so new bytes get a new ETag. Model: opus-5-5
This commit is contained in:
1 parent
b3b8408d18
commit
445ff628fd
29 files changed
+1709
-269
No files matched your search
+329
-38
@@ -3,8 +3,12 @@
|
||||
// - the variants made from them
|
||||
// - the negative cache: the source URLs whose fetch failed recently
|
||||
//
|
||||
// Each image is a file with a row in the database. Eviction removes the
|
||||
// least recently used images to keep the cache under its size limit.
|
||||
// Each image is a file with a row in the database. A source is used, and its
|
||||
// variants are served, until the cache TTL has passed since upstream last sent
|
||||
// the source or confirmed it unchanged. Then the source is due to be checked
|
||||
// with upstream again (see Stale). Eviction removes:
|
||||
// - the least recently used images, to keep the cache under its size limit
|
||||
// - the images not checked with upstream for EvictionAfterLastCheck
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
@@ -13,7 +17,9 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/fs"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
@@ -38,11 +44,31 @@ const cleanExpiredBatchSize = 1000
|
||||
// cache.
|
||||
const DefaultNegativeTTL = 5 * time.Minute
|
||||
|
||||
// DefaultCacheTTL is how long after upstream last sent a cached source, or
|
||||
// answered that it had not changed, pixad uses the source and serves its
|
||||
// variants without asking upstream again.
|
||||
const DefaultCacheTTL = 24 * time.Hour
|
||||
|
||||
// EvictionAfterLastCheck is how long after upstream last sent a cached
|
||||
// source, or answered that it had not changed, the source and its variants
|
||||
// are evicted (see EvictNotCheckedRecently). Until then a source whose checks
|
||||
// fail is used.
|
||||
const EvictionAfterLastCheck = 30 * 24 * time.Hour
|
||||
|
||||
// ErrStale is returned by Variant for a variant whose file's modification
|
||||
// time is CacheTTL or more ago (see Variant).
|
||||
var ErrStale = errors.New("cached variant is older than the cache TTL")
|
||||
|
||||
// Config holds cache configuration.
|
||||
type Config struct {
|
||||
StateDir string
|
||||
NegativeTTL time.Duration
|
||||
|
||||
// CacheTTL is how long a cached source is used, and its variants served,
|
||||
// before upstream is asked again whether the source changed. Zero means
|
||||
// DefaultCacheTTL.
|
||||
CacheTTL time.Duration
|
||||
|
||||
// MaxBytes is the disk cache size limit in bytes that eviction
|
||||
// enforces. Zero turns the disk cache off entirely: no cache
|
||||
// directories are created, lookups always miss, stores are no-ops,
|
||||
@@ -113,6 +139,10 @@ func NewCache(
|
||||
log = slog.Default()
|
||||
}
|
||||
|
||||
if config.CacheTTL == 0 {
|
||||
config.CacheTTL = DefaultCacheTTL
|
||||
}
|
||||
|
||||
c := &Cache{
|
||||
db: db,
|
||||
config: config,
|
||||
@@ -156,43 +186,167 @@ func NewCache(
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// Variant opens a cached variant and returns a reader, its size and its
|
||||
// content type, read from the variant file's first bytes. It opens that one
|
||||
// file and makes no database call. A variant it opens has its access time
|
||||
// handed to the background writer without waiting (see touchVariant). A
|
||||
// variant not in the cache, or any variant of a disabled cache, is
|
||||
// ErrNotFound; a variant file whose first bytes are not those of an output
|
||||
// format is another error.
|
||||
func (c *Cache) Variant(
|
||||
cacheKey imagerequest.VariantKey,
|
||||
) (io.ReadCloser, int64, string, error) {
|
||||
// VariantFile is the open file of a cached variant, as Variant,
|
||||
// VariantIgnoringTTL and RenewVariant return it.
|
||||
type VariantFile struct {
|
||||
// Content reads the file. The caller closes it.
|
||||
Content io.ReadCloser
|
||||
// Size is the file's size in bytes.
|
||||
Size int64
|
||||
// ContentType is the MIME type of the variant's format, read from the
|
||||
// file's first bytes.
|
||||
ContentType string
|
||||
// CheckedAt is the file's modification time: its source's check time
|
||||
// when the variant was made or last renewed.
|
||||
CheckedAt time.Time
|
||||
}
|
||||
|
||||
// Variant opens a cached variant and returns its file (see VariantFile). It
|
||||
// opens that one file and makes no database call: the file's modification
|
||||
// time is its source's check time when the variant was made or last renewed
|
||||
// (see PutVariant), and a variant whose time is CacheTTL or more ago is
|
||||
// ErrStale. A variant it opens has its access time handed to the background
|
||||
// writer without waiting (see touchVariant). A variant not in the cache, or
|
||||
// any variant of a disabled cache, is ErrNotFound; a variant file whose first
|
||||
// bytes are not those of an output format is another error.
|
||||
func (c *Cache) Variant(cacheKey imagerequest.VariantKey) (VariantFile, error) {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return nil, 0, "", ErrNotFound
|
||||
return VariantFile{}, ErrNotFound
|
||||
}
|
||||
|
||||
file, size, err := c.variants.Open(string(cacheKey))
|
||||
file, info, err := c.variants.Open(string(cacheKey))
|
||||
if err != nil {
|
||||
return nil, 0, "", err
|
||||
return VariantFile{}, err
|
||||
}
|
||||
|
||||
if c.Stale(info.ModTime()) {
|
||||
_ = file.Close()
|
||||
|
||||
return VariantFile{}, ErrStale
|
||||
}
|
||||
|
||||
return c.finishOpeningVariant(cacheKey, file, info)
|
||||
}
|
||||
|
||||
// VariantIgnoringTTL opens a cached variant whatever its age, and returns it
|
||||
// as Variant does. The image proxy reads it for a variant whose source is not
|
||||
// cached, before it checks the source with upstream (see LookupVariant).
|
||||
func (c *Cache) VariantIgnoringTTL(
|
||||
cacheKey imagerequest.VariantKey,
|
||||
) (VariantFile, error) {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return VariantFile{}, ErrNotFound
|
||||
}
|
||||
|
||||
file, info, err := c.variants.Open(string(cacheKey))
|
||||
if err != nil {
|
||||
return VariantFile{}, err
|
||||
}
|
||||
|
||||
return c.finishOpeningVariant(cacheKey, file, info)
|
||||
}
|
||||
|
||||
// RenewVariant opens a cached variant made from the source bytes that source
|
||||
// describes, whatever its age, and returns it as Variant does. It moves the
|
||||
// variant's check time forward to source.CheckedAt (see SetVariantCheckedAt).
|
||||
// A variant made before source.FetchedAt was made from older bytes, and is
|
||||
// ErrNotFound, as is a variant not in the cache.
|
||||
func (c *Cache) RenewVariant(
|
||||
ctx context.Context, cacheKey imagerequest.VariantKey, source SourceInfo,
|
||||
) (VariantFile, error) {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return VariantFile{}, ErrNotFound
|
||||
}
|
||||
|
||||
file, info, err := c.variants.Open(string(cacheKey))
|
||||
if err != nil {
|
||||
return VariantFile{}, err
|
||||
}
|
||||
|
||||
if info.ModTime().Before(source.FetchedAt) {
|
||||
_ = file.Close()
|
||||
|
||||
return VariantFile{}, ErrNotFound
|
||||
}
|
||||
|
||||
format, err := readFormat(file)
|
||||
if err != nil {
|
||||
_ = file.Close()
|
||||
|
||||
return nil, 0, "", err
|
||||
return VariantFile{}, err
|
||||
}
|
||||
|
||||
checkedAt := info.ModTime()
|
||||
|
||||
if checkedAt.Before(source.CheckedAt) {
|
||||
err = c.SetVariantCheckedAt(ctx, cacheKey, source.CheckedAt)
|
||||
if err != nil {
|
||||
_ = file.Close()
|
||||
|
||||
return VariantFile{}, err
|
||||
}
|
||||
|
||||
checkedAt = source.CheckedAt
|
||||
}
|
||||
|
||||
c.touchVariant(cacheKey)
|
||||
|
||||
return file, size, format.MIMEType(), nil
|
||||
return VariantFile{
|
||||
Content: file,
|
||||
Size: info.Size(),
|
||||
ContentType: format.MIMEType(),
|
||||
CheckedAt: checkedAt,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// PutVariant stores a processed variant by its cache key and records
|
||||
// it in the size accounting. On a disabled cache it is a no-op. The
|
||||
// accounting insert is best-effort (the startup reconciliation pass
|
||||
// adopts any variant file that misses its accounting row).
|
||||
// SetVariantCheckedAt records checkedAt as when upstream last sent the source
|
||||
// bytes the cached variant of cacheKey was made from, or answered that they
|
||||
// had not changed. It sets the variant file's modification time to checkedAt,
|
||||
// so Variant serves the file until CacheTTL after it. It sets the checked_at
|
||||
// of the variant's row too, so eviction keeps the variant until
|
||||
// EvictionAfterLastCheck after it. A failure to write the row is only logged.
|
||||
// On a disabled cache it is a no-op.
|
||||
func (c *Cache) SetVariantCheckedAt(
|
||||
ctx context.Context, cacheKey imagerequest.VariantKey, checkedAt time.Time,
|
||||
) error {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
err := c.variants.SetModTime(string(cacheKey), checkedAt)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
_, err = c.db.ExecContext(ctx,
|
||||
`UPDATE variant_content SET checked_at = ? WHERE cache_key = ?`,
|
||||
checkedAt.UTC(), string(cacheKey))
|
||||
if err != nil {
|
||||
c.log.Warn("failed to renew variant's checked_at",
|
||||
"cache_key", cacheKey, "error", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stale reports whether a source last checked with upstream at checkedAt,
|
||||
// and its variants, are due to be checked again.
|
||||
func (c *Cache) Stale(checkedAt time.Time) bool {
|
||||
return time.Since(checkedAt) >= c.config.CacheTTL
|
||||
}
|
||||
|
||||
// PutVariant stores a processed variant by its cache key and records it in
|
||||
// the size accounting. source describes the source bytes the variant was made
|
||||
// from. Its CheckedAt becomes the modification time of the variant's file,
|
||||
// which Variant and RenewVariant read. The variant's row keeps its CheckedAt,
|
||||
// ETag and LastModified (see LookupVariant). On a disabled cache it is a
|
||||
// no-op. The accounting insert is best-effort (the startup reconciliation
|
||||
// pass adopts any variant file that misses its accounting row).
|
||||
func (c *Cache) PutVariant(
|
||||
ctx context.Context, cacheKey imagerequest.VariantKey, content io.Reader,
|
||||
ctx context.Context,
|
||||
cacheKey imagerequest.VariantKey,
|
||||
source SourceInfo,
|
||||
content io.Reader,
|
||||
) error {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return nil
|
||||
@@ -208,15 +362,33 @@ func (c *Cache) PutVariant(
|
||||
return err
|
||||
}
|
||||
|
||||
// A file left with the time of its writing could be served as made from
|
||||
// bytes fetched after the ones it was made from
|
||||
err = c.variants.SetModTime(string(cacheKey), source.CheckedAt)
|
||||
if err != nil {
|
||||
deleteErr := c.variants.Delete(string(cacheKey))
|
||||
if deleteErr != nil {
|
||||
c.log.Warn("failed to remove variant file whose time was not set",
|
||||
"cache_key", cacheKey, "error", deleteErr)
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
size := int64(len(data))
|
||||
|
||||
_, err = c.db.ExecContext(ctx, `
|
||||
INSERT INTO variant_content (cache_key, size_bytes)
|
||||
VALUES (?, ?)
|
||||
INSERT INTO variant_content (cache_key, size_bytes, checked_at,
|
||||
upstream_etag, upstream_last_modified)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
ON CONFLICT(cache_key) DO UPDATE SET
|
||||
size_bytes = excluded.size_bytes,
|
||||
last_accessed_at = CURRENT_TIMESTAMP
|
||||
`, string(cacheKey), size)
|
||||
last_accessed_at = CURRENT_TIMESTAMP,
|
||||
checked_at = excluded.checked_at,
|
||||
upstream_etag = excluded.upstream_etag,
|
||||
upstream_last_modified = excluded.upstream_last_modified
|
||||
`, string(cacheKey), size, source.CheckedAt.UTC(), source.ETag,
|
||||
source.LastModified)
|
||||
if err != nil {
|
||||
c.log.Warn("failed to record variant in size accounting",
|
||||
"cache_key", cacheKey, "error", err)
|
||||
@@ -255,7 +427,7 @@ func (c *Cache) Source(
|
||||
return nil, 0, "", ErrNotFound
|
||||
}
|
||||
|
||||
file, size, err := c.srcContent.Open(string(key))
|
||||
file, info, err := c.srcContent.Open(string(key))
|
||||
if err != nil {
|
||||
return nil, 0, "", err
|
||||
}
|
||||
@@ -269,19 +441,107 @@ func (c *Cache) Source(
|
||||
|
||||
c.touchSourceContent(key)
|
||||
|
||||
return file, size, format, nil
|
||||
return file, info.Size(), format, nil
|
||||
}
|
||||
|
||||
// PutSource stores fetched source content: it writes the file named by
|
||||
// the source's key, then the source's row. When the row cannot be written
|
||||
// the file is removed and the store fails, as a file with no row is never
|
||||
// served or evicted; if that removal fails too, the next reconciliation
|
||||
// pass removes the file. On a disabled cache it is a no-op returning an
|
||||
// empty key.
|
||||
// SourceInfo is what the cache keeps in a cached source's row besides its
|
||||
// size and access time.
|
||||
type SourceInfo struct {
|
||||
// FetchedAt is when the cached bytes were fetched.
|
||||
FetchedAt time.Time
|
||||
// CheckedAt is when upstream last sent the bytes or answered that they
|
||||
// had not changed.
|
||||
CheckedAt time.Time
|
||||
// ETag is the ETag header upstream sent with the bytes, or "".
|
||||
ETag string
|
||||
// LastModified is the Last-Modified header upstream sent with the
|
||||
// bytes, or "".
|
||||
LastModified string
|
||||
}
|
||||
|
||||
// LookupSource returns what the cache keeps about the cached source of src.
|
||||
// A source with no row is ErrNotFound, as is any source of a disabled cache.
|
||||
func (c *Cache) LookupSource(
|
||||
ctx context.Context, src imagerequest.Source,
|
||||
) (SourceInfo, error) {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return SourceInfo{}, ErrNotFound
|
||||
}
|
||||
|
||||
var info SourceInfo
|
||||
|
||||
err := c.db.QueryRowContext(ctx, `
|
||||
SELECT fetched_at, checked_at, upstream_etag, upstream_last_modified
|
||||
FROM source_content WHERE source_key = ?
|
||||
`, string(src.Key())).Scan(
|
||||
&info.FetchedAt, &info.CheckedAt, &info.ETag, &info.LastModified)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return SourceInfo{}, ErrNotFound
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return SourceInfo{}, fmt.Errorf("failed to look up source: %w", err)
|
||||
}
|
||||
|
||||
return info, nil
|
||||
}
|
||||
|
||||
// LookupVariant returns what the row of the cached variant of cacheKey keeps
|
||||
// about the source bytes the variant was made from: the CheckedAt, ETag and
|
||||
// LastModified they had when the variant was made or last renewed. The row
|
||||
// keeps no FetchedAt, so it is zero. With the ETag and LastModified the image
|
||||
// proxy checks the variant with upstream after its source is evicted. A
|
||||
// variant with no row is ErrNotFound, as is any variant of a disabled cache.
|
||||
func (c *Cache) LookupVariant(
|
||||
ctx context.Context, cacheKey imagerequest.VariantKey,
|
||||
) (SourceInfo, error) {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return SourceInfo{}, ErrNotFound
|
||||
}
|
||||
|
||||
var info SourceInfo
|
||||
|
||||
err := c.db.QueryRowContext(ctx, `
|
||||
SELECT checked_at, upstream_etag, upstream_last_modified
|
||||
FROM variant_content WHERE cache_key = ?
|
||||
`, string(cacheKey)).Scan(&info.CheckedAt, &info.ETag, &info.LastModified)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return SourceInfo{}, ErrNotFound
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return SourceInfo{}, fmt.Errorf("failed to look up variant: %w", err)
|
||||
}
|
||||
|
||||
return info, nil
|
||||
}
|
||||
|
||||
// RenewSource records that upstream answered at checkedAt that the cached
|
||||
// source of src had not changed.
|
||||
func (c *Cache) RenewSource(
|
||||
ctx context.Context, src imagerequest.Source, checkedAt time.Time,
|
||||
) error {
|
||||
_, err := c.db.ExecContext(ctx,
|
||||
`UPDATE source_content SET checked_at = ? WHERE source_key = ?`,
|
||||
checkedAt.UTC(), string(src.Key()))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to renew source: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// PutSource stores fetched source content with what info says about it: it
|
||||
// writes the file named by the source's key, then the source's row. When the
|
||||
// row cannot be written the file is removed and the store fails, as a file
|
||||
// with no row is never served or evicted; if that removal fails too, the
|
||||
// next reconciliation pass removes the file. On a disabled cache it is a
|
||||
// no-op returning an empty key.
|
||||
func (c *Cache) PutSource(
|
||||
ctx context.Context,
|
||||
src imagerequest.Source,
|
||||
content io.Reader,
|
||||
info SourceInfo,
|
||||
) (imagerequest.SourceKey, error) {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return "", nil
|
||||
@@ -302,12 +562,18 @@ func (c *Cache) PutSource(
|
||||
size := int64(len(data))
|
||||
|
||||
_, err = c.db.ExecContext(ctx, `
|
||||
INSERT INTO source_content (source_key, size_bytes)
|
||||
VALUES (?, ?)
|
||||
INSERT INTO source_content (source_key, size_bytes, fetched_at,
|
||||
checked_at, upstream_etag, upstream_last_modified)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(source_key) DO UPDATE SET
|
||||
size_bytes = excluded.size_bytes,
|
||||
last_accessed_at = CURRENT_TIMESTAMP
|
||||
`, string(key), size)
|
||||
last_accessed_at = CURRENT_TIMESTAMP,
|
||||
fetched_at = excluded.fetched_at,
|
||||
checked_at = excluded.checked_at,
|
||||
upstream_etag = excluded.upstream_etag,
|
||||
upstream_last_modified = excluded.upstream_last_modified
|
||||
`, string(key), size, info.FetchedAt.UTC(), info.CheckedAt.UTC(), info.ETag,
|
||||
info.LastModified)
|
||||
if err != nil {
|
||||
deleteErr := c.srcContent.Delete(string(key))
|
||||
if deleteErr != nil {
|
||||
@@ -449,6 +715,31 @@ func (c *Cache) cleanExpiredBatch(ctx context.Context) (int64, error) {
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
// finishOpeningVariant returns a variant file that FileStore.Open opened, with
|
||||
// its information info, as Variant does. It reads the content type from the
|
||||
// file's first bytes and hands the variant's access time to the background
|
||||
// writer (see touchVariant). It closes the file when its first bytes are not
|
||||
// those of an output format.
|
||||
func (c *Cache) finishOpeningVariant(
|
||||
cacheKey imagerequest.VariantKey, file *os.File, info fs.FileInfo,
|
||||
) (VariantFile, error) {
|
||||
format, err := readFormat(file)
|
||||
if err != nil {
|
||||
_ = file.Close()
|
||||
|
||||
return VariantFile{}, err
|
||||
}
|
||||
|
||||
c.touchVariant(cacheKey)
|
||||
|
||||
return VariantFile{
|
||||
Content: file,
|
||||
Size: info.Size(),
|
||||
ContentType: format.MIMEType(),
|
||||
CheckedAt: info.ModTime(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// readFormat reads the first bytes of a source or variant file, without
|
||||
// moving its offset, and returns the format they show. Sources and variants
|
||||
// are in the same six formats.
|
||||
|
||||
@@ -49,16 +49,19 @@ func TestVariant_ContentTypeFromFirstBytes(t *testing.T) {
|
||||
|
||||
key := imagerequest.VariantKey("aabbccdd10" + string(format))
|
||||
|
||||
err = cache.PutVariant(t.Context(), key, bytes.NewReader(image))
|
||||
err = cache.PutVariant(t.Context(), key, SourceInfo{CheckedAt: time.Now()},
|
||||
bytes.NewReader(image))
|
||||
if err != nil {
|
||||
t.Fatalf("PutVariant(%s) error = %v", format, err)
|
||||
}
|
||||
|
||||
reader, size, contentType, err := cache.Variant(key)
|
||||
variant, err := cache.Variant(key)
|
||||
if err != nil {
|
||||
t.Fatalf("Variant(%s) error = %v", format, err)
|
||||
}
|
||||
|
||||
reader, size, contentType := variant.Content, variant.Size, variant.ContentType
|
||||
|
||||
served, err := io.ReadAll(reader)
|
||||
_ = reader.Close()
|
||||
|
||||
@@ -109,16 +112,19 @@ func TestVariant_RefusesFileOfNoOutputFormat(t *testing.T) {
|
||||
t.Fatalf("%s: storing the variant file: %v", tt.name, err)
|
||||
}
|
||||
|
||||
reader, _, contentType, err := cache.Variant(key)
|
||||
variant, err := cache.Variant(key)
|
||||
t.Logf("%s: Variant() error = %v", tt.name, err)
|
||||
|
||||
reader, contentType := variant.Content, variant.ContentType
|
||||
if reader != nil || contentType != "" || !errors.Is(err, tt.wantErr) {
|
||||
t.Errorf("%s: Variant() = reader %v, content type %q, error %v; "+
|
||||
"want no reader and %v", tt.name, reader, contentType, err, tt.wantErr)
|
||||
}
|
||||
}
|
||||
|
||||
reader, _, _, err := cache.Variant(testVariantKeyFour)
|
||||
variant, err := cache.Variant(testVariantKeyFour)
|
||||
|
||||
reader := variant.Content
|
||||
if reader != nil || !errors.Is(err, ErrNotFound) {
|
||||
t.Errorf("Variant() of a variant with no file = reader %v, error %v; "+
|
||||
"want no reader and %v", reader, err, ErrNotFound)
|
||||
@@ -126,7 +132,9 @@ func TestVariant_RefusesFileOfNoOutputFormat(t *testing.T) {
|
||||
|
||||
disabled, _ := newEvictionTestCache(t, 0)
|
||||
|
||||
reader, _, _, err = disabled.Variant(testVariantKeyOne)
|
||||
variant, err = disabled.Variant(testVariantKeyOne)
|
||||
|
||||
reader = variant.Content
|
||||
if reader != nil || !errors.Is(err, ErrNotFound) {
|
||||
t.Errorf("Variant() on a disabled cache = reader %v, error %v; "+
|
||||
"want no reader and %v", reader, err, ErrNotFound)
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"os"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
// TestEvictNotCheckedRecently stores a source and a variant last checked with
|
||||
// upstream 31 days ago and a source and a variant checked 29 days ago.
|
||||
// EvictNotCheckedRecently must remove the rows and files of the first two and
|
||||
// keep the other two.
|
||||
func TestEvictNotCheckedRecently(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
old := time.Now().Add(-31 * 24 * time.Hour)
|
||||
recent := time.Now().Add(-29 * 24 * time.Hour)
|
||||
|
||||
oldSource := imagerequest.Source{Host: testHostCDN, Path: "/old.jpg"}
|
||||
recentSource := imagerequest.Source{Host: testHostCDN, Path: "/recent.jpg"}
|
||||
|
||||
for src, checkedAt := range map[imagerequest.Source]time.Time{
|
||||
oldSource: old, recentSource: recent,
|
||||
} {
|
||||
_, err := cache.PutSource(t.Context(), src, bytes.NewReader([]byte("source")),
|
||||
SourceInfo{FetchedAt: checkedAt, CheckedAt: checkedAt})
|
||||
if err != nil {
|
||||
t.Fatalf("PutSource(%s) error = %v", src.Path, err)
|
||||
}
|
||||
}
|
||||
|
||||
oldVariant := imagerequest.VariantKey("aabbccdd0001")
|
||||
recentVariant := imagerequest.VariantKey("aabbccdd0002")
|
||||
|
||||
for key, checkedAt := range map[imagerequest.VariantKey]time.Time{
|
||||
oldVariant: old, recentVariant: recent,
|
||||
} {
|
||||
err := cache.PutVariant(t.Context(), key, SourceInfo{CheckedAt: checkedAt},
|
||||
bytes.NewReader([]byte("variant")))
|
||||
if err != nil {
|
||||
t.Fatalf("PutVariant(%s) error = %v", key, err)
|
||||
}
|
||||
}
|
||||
|
||||
err := cache.EvictNotCheckedRecently(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("EvictNotCheckedRecently() error = %v", err)
|
||||
}
|
||||
|
||||
sourceRows, err := queryStringColumn[imagerequest.SourceKey](t.Context(),
|
||||
cache.db, `SELECT source_key FROM source_content`, "sources", "source")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
variantRows, err := queryStringColumn[imagerequest.VariantKey](t.Context(),
|
||||
cache.db, `SELECT cache_key FROM variant_content`, "variants", "variant")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
t.Logf("source rows left = %v, variant rows left = %v", sourceRows, variantRows)
|
||||
|
||||
if !slices.Equal(sourceRows, []imagerequest.SourceKey{recentSource.Key()}) {
|
||||
t.Errorf("source rows left = %v, want only %s", sourceRows, recentSource.Key())
|
||||
}
|
||||
|
||||
if !slices.Equal(variantRows, []imagerequest.VariantKey{recentVariant}) {
|
||||
t.Errorf("variant rows left = %v, want only %s", variantRows, recentVariant)
|
||||
}
|
||||
|
||||
for path, wantKept := range map[string]bool{
|
||||
cache.srcContent.keyToPath(string(oldSource.Key())): false,
|
||||
cache.srcContent.keyToPath(string(recentSource.Key())): true,
|
||||
cache.variants.keyToPath(string(oldVariant)): false,
|
||||
cache.variants.keyToPath(string(recentVariant)): true,
|
||||
} {
|
||||
_, statErr := os.Stat(path)
|
||||
if kept := statErr == nil; kept != wantKept {
|
||||
t.Errorf("file %s kept = %t, want %t (%v)", path, kept, wantKept, statErr)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -131,6 +131,56 @@ func (c *Cache) EvictToLimit(ctx context.Context) error {
|
||||
}
|
||||
}
|
||||
|
||||
// EvictNotCheckedRecently evicts each source and variant whose checked_at is
|
||||
// EvictionAfterLastCheck or more ago, evictionBatchSize of each at a time,
|
||||
// until none is left. It is a no-op when the cache is disabled.
|
||||
func (c *Cache) EvictNotCheckedRecently(ctx context.Context) error {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
cutoff := time.Now().UTC().Add(-EvictionAfterLastCheck)
|
||||
|
||||
for {
|
||||
sources, err := queryStringColumn[imagerequest.SourceKey](ctx, c.db, `
|
||||
SELECT source_key FROM source_content WHERE checked_at < ? LIMIT ?
|
||||
`, "sources not checked recently", "source key", cutoff, evictionBatchSize)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, key := range sources {
|
||||
err = c.evictSourceBlob(ctx, key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
variants, err := queryStringColumn[imagerequest.VariantKey](ctx, c.db, `
|
||||
SELECT cache_key FROM variant_content WHERE checked_at < ? LIMIT ?
|
||||
`, "variants not checked recently", "variant key", cutoff, evictionBatchSize)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, key := range variants {
|
||||
err = c.evictVariant(ctx, key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if len(sources) > 0 || len(variants) > 0 {
|
||||
c.log.Info("evicted cache content not checked with upstream recently",
|
||||
"sources", len(sources), "variants", len(variants))
|
||||
}
|
||||
|
||||
if len(sources) < evictionBatchSize && len(variants) < evictionBatchSize {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// evictBatch fetches one batch of LRU candidates across variants and
|
||||
// source blobs and evicts them oldest-first until excessBytes are
|
||||
// freed or the batch is exhausted. It returns the bytes freed. A
|
||||
@@ -344,8 +394,9 @@ func (c *Cache) notifyWritePressure() {
|
||||
}
|
||||
|
||||
// StartEviction launches the background eviction goroutine, which
|
||||
// removes the expired negative cache rows at startup and again on every
|
||||
// periodic tick thereafter, reconciles the database accounting with the
|
||||
// removes the expired negative cache rows, and the sources and variants not
|
||||
// checked for EvictionAfterLastCheck, at startup and again on every periodic
|
||||
// tick thereafter, reconciles the database accounting with the
|
||||
// cache directories at startup and again on the first periodic tick
|
||||
// that comes at least ten times as long after a reconciliation pass
|
||||
// ended as that pass took, and evicts to the configured limit on the given
|
||||
@@ -392,6 +443,7 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
|
||||
|
||||
nextReconciliation := c.runReconciliationPass(ctx)
|
||||
c.runCleanExpired(ctx)
|
||||
c.runEvictNotCheckedRecently(ctx)
|
||||
c.runEvictionPass(ctx)
|
||||
|
||||
ticker := time.NewTicker(interval)
|
||||
@@ -417,6 +469,7 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
|
||||
}
|
||||
|
||||
c.runCleanExpired(ctx)
|
||||
c.runEvictNotCheckedRecently(ctx)
|
||||
case <-c.evictionPressure:
|
||||
}
|
||||
|
||||
@@ -474,6 +527,21 @@ func (c *Cache) runCleanExpired(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// runEvictNotCheckedRecently runs EvictNotCheckedRecently, logging a failure
|
||||
// instead of propagating it (the loop must keep running). It does nothing
|
||||
// once ctx is cancelled.
|
||||
func (c *Cache) runEvictNotCheckedRecently(ctx context.Context) {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err := c.EvictNotCheckedRecently(ctx)
|
||||
if err != nil {
|
||||
c.log.Warn("evicting cache content not checked recently failed",
|
||||
"error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// reconcileAccounting synchronizes the database size accounting with
|
||||
// the actual contents of the cache directories. It runs at startup and
|
||||
// then on the first periodic eviction tick that comes at least ten times
|
||||
@@ -609,16 +677,18 @@ func (c *Cache) reconcileVariantFile(
|
||||
}
|
||||
|
||||
// adoptVariantFile inserts an accounting row for a variant file that
|
||||
// has none, using the file's size and modification time.
|
||||
// has none, using the file's size and modification time, which is also
|
||||
// the variant's checked_at.
|
||||
func (c *Cache) adoptVariantFile(
|
||||
ctx context.Context, info fs.FileInfo, cacheKey imagerequest.VariantKey,
|
||||
) error {
|
||||
modTime := info.ModTime().UTC().Format(sqliteTimestampLayout)
|
||||
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
INSERT INTO variant_content (cache_key, size_bytes, last_accessed_at)
|
||||
VALUES (?, ?, ?)
|
||||
`, string(cacheKey), info.Size(), modTime)
|
||||
INSERT INTO variant_content (cache_key, size_bytes, last_accessed_at,
|
||||
checked_at)
|
||||
VALUES (?, ?, ?, ?)
|
||||
`, string(cacheKey), info.Size(), modTime, info.ModTime().UTC())
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to adopt variant file into accounting: %w", err)
|
||||
}
|
||||
|
||||
@@ -101,7 +101,7 @@ func storeEvictionTestSource(
|
||||
}
|
||||
|
||||
key, err := cache.PutSource(
|
||||
context.Background(), req.Source, bytes.NewReader(content),
|
||||
context.Background(), req.Source, bytes.NewReader(content), SourceInfo{},
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("PutSource(%s%s) failed: %v", host, path, err)
|
||||
@@ -117,7 +117,8 @@ func storeEvictionTestVariant(
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
err := cache.PutVariant(t.Context(), key, bytes.NewReader(content))
|
||||
err := cache.PutVariant(t.Context(), key, SourceInfo{CheckedAt: time.Now()},
|
||||
bytes.NewReader(content))
|
||||
if err != nil {
|
||||
t.Fatalf("PutVariant(%s) failed: %v", key, err)
|
||||
}
|
||||
@@ -432,13 +433,14 @@ func assertDisabledCacheWritesAreNoOps(
|
||||
ctx := t.Context()
|
||||
|
||||
err := cache.PutVariant(
|
||||
ctx, imagerequest.CacheKey(req), bytes.NewReader([]byte("data")),
|
||||
ctx, imagerequest.CacheKey(req), SourceInfo{}, bytes.NewReader([]byte("data")),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("PutVariant on disabled cache must be a no-op, got error: %v", err)
|
||||
}
|
||||
|
||||
key, err := cache.PutSource(ctx, req.Source, bytes.NewReader([]byte("data")))
|
||||
key, err := cache.PutSource(ctx, req.Source, bytes.NewReader([]byte("data")),
|
||||
SourceInfo{})
|
||||
if err != nil {
|
||||
t.Fatalf("PutSource on disabled cache must be a no-op, got error: %v", err)
|
||||
}
|
||||
|
||||
@@ -3,8 +3,10 @@ package imgcache
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Storage constants.
|
||||
@@ -54,26 +56,26 @@ func (s *FileStore) Write(key string, data []byte) error {
|
||||
return writeFileSynced(path, data)
|
||||
}
|
||||
|
||||
// Open opens the file for key and returns it with its size in bytes. A
|
||||
// missing file is ErrNotFound.
|
||||
func (s *FileStore) Open(key string) (*os.File, int64, error) {
|
||||
// Open opens the file for key and returns it with its information, which
|
||||
// gives its size and its modification time. A missing file is ErrNotFound.
|
||||
func (s *FileStore) Open(key string) (*os.File, fs.FileInfo, error) {
|
||||
f, err := os.Open(s.keyToPath(key))
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return nil, 0, ErrNotFound
|
||||
return nil, nil, ErrNotFound
|
||||
}
|
||||
|
||||
return nil, 0, fmt.Errorf("failed to open content: %w", err)
|
||||
return nil, nil, fmt.Errorf("failed to open content: %w", err)
|
||||
}
|
||||
|
||||
stat, err := f.Stat()
|
||||
info, err := f.Stat()
|
||||
if err != nil {
|
||||
_ = f.Close()
|
||||
|
||||
return nil, 0, fmt.Errorf("failed to stat content: %w", err)
|
||||
return nil, nil, fmt.Errorf("failed to stat content: %w", err)
|
||||
}
|
||||
|
||||
return f, stat.Size(), nil
|
||||
return f, info, nil
|
||||
}
|
||||
|
||||
// Size returns the size in bytes of the file for key. A missing file is
|
||||
@@ -91,6 +93,17 @@ func (s *FileStore) Size(key string) (int64, error) {
|
||||
return stat.Size(), nil
|
||||
}
|
||||
|
||||
// SetModTime sets the modification time of the file for key to modTime,
|
||||
// leaving its access time as it is.
|
||||
func (s *FileStore) SetModTime(key string, modTime time.Time) error {
|
||||
err := os.Chtimes(s.keyToPath(key), time.Time{}, modTime)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to set modification time: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Delete removes the file for key. A missing file is not an error.
|
||||
func (s *FileStore) Delete(key string) error {
|
||||
err := os.Remove(s.keyToPath(key))
|
||||
|
||||
@@ -38,15 +38,15 @@ func TestFileStore_WriteAndOpen(t *testing.T) {
|
||||
}
|
||||
|
||||
// Open and verify size and content
|
||||
r, size, err := storage.Open(key)
|
||||
r, info, err := storage.Open(key)
|
||||
if err != nil {
|
||||
t.Fatalf("Open() error = %v", err)
|
||||
}
|
||||
|
||||
defer func() { _ = r.Close() }()
|
||||
|
||||
if size != int64(len(content)) {
|
||||
t.Errorf("Open() size = %d, want %d", size, len(content))
|
||||
if info.Size() != int64(len(content)) {
|
||||
t.Errorf("Open() size = %d, want %d", info.Size(), len(content))
|
||||
}
|
||||
|
||||
loaded, err := io.ReadAll(r)
|
||||
|
||||
@@ -128,7 +128,7 @@ func TestPutSource_FailedRowWriteLeavesNoFile(t *testing.T) {
|
||||
Source: imagerequest.Source{Host: testHostCDN, Path: testPathCat},
|
||||
}
|
||||
key, err := cache.PutSource(t.Context(), req.Source,
|
||||
bytes.NewReader([]byte("source bytes")))
|
||||
bytes.NewReader([]byte("source bytes")), SourceInfo{})
|
||||
t.Logf("PutSource() = %q, error = %v", key, err)
|
||||
|
||||
if err == nil {
|
||||
@@ -170,7 +170,7 @@ func TestSource_FormatFromFirstBytes(t *testing.T) {
|
||||
t.Fatalf("reading the %s image: %v", format, err)
|
||||
}
|
||||
|
||||
_, err = cache.PutSource(t.Context(), src, bytes.NewReader(image))
|
||||
_, err = cache.PutSource(t.Context(), src, bytes.NewReader(image), SourceInfo{})
|
||||
if err != nil {
|
||||
t.Fatalf("PutSource(%s) error = %v", format, err)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user