Files
pixa/internal/imgcache/cache.go
T
clawbot efbf05a6fa
check / check (push) Waiting to run
Support SVG sources (closes #68)
The fetcher accepts image/svg+xml and internal/magic detects SVG. orig
of an SVG at the orig size is the SVG unchanged, cached as a variant; at
another size it is ErrSVGNotOriginalSize (400). Any other format renders
it with libvips' thumbnail at the least size covering the output, at
most 8192 on a side, so it stays sharp. An SVG's own size is not
limited; one with a side over 8192 times the other is ErrSVGTooThin
(415). The cache reads up to 1024 first bytes to find a file's format.
Tests that pinned the SVG refusal now expect SVG to be read. Render time
and memory are unbounded: P1 in TODO.md until the owner decides on
#68.

Model: opus-5-5
2026-11-11 13:21:12 +00:00

758 lines
24 KiB
Go

// Package imgcache is the image cache. It keeps:
// - the source images fetched from upstream
// - 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. 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 (
"context"
"database/sql"
"errors"
"fmt"
"io"
"io/fs"
"log/slog"
"os"
"path/filepath"
"time"
"github.com/prometheus/client_golang/prometheus"
"sneak.berlin/go/pixa/internal/imagerequest"
"sneak.berlin/go/pixa/internal/magic"
)
// maxPendingAccessTimes is how many access times of variants served from
// the cache, and how many of cached sources a miss used, may each wait to be
// written, at about 80 bytes each. An access time handed over while that
// many of its kind wait is dropped.
const maxPendingAccessTimes = 10000
// cleanExpiredBatchSize is the most expired negative cache rows one
// statement of CleanExpired deletes. Each batch is a statement of its
// own, so a request waits behind one batch at most, however many rows
// have expired.
const cleanExpiredBatchSize = 1000
// DefaultNegativeTTL is how long pixad keeps a failed fetch in the negative
// 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,
// and the eviction goroutine only removes the expired negative cache
// rows.
MaxBytes int64
// Logger receives accounting and eviction log output. A nil
// Logger means slog.Default().
Logger *slog.Logger
}
// Cache implements the caching layer for the image proxy.
type Cache struct {
db *sql.DB
srcContent *FileStore // source images by source key
variants *FileStore // processed variants by cache key
config Config
log *slog.Logger
// Eviction machinery. The channels are created in NewCache so
// stores can signal write pressure without racing StartEviction.
// evictionCancel, set by StartEviction, cancels the eviction
// goroutine's context.
evictionPressure chan struct{}
evictionDone chan struct{}
evictionCancel context.CancelFunc
// pendingVariantAccessTimes holds the variants served from the cache,
// and pendingSourceAccessTimes the cached sources a miss used, whose
// access time is still to be written by the goroutine that
// StartPendingAccessTimeWrites starts, so that no request waits for the
// database to record it. As for eviction, the channels are created in
// NewCache, so that requests can hand over access times before that
// goroutine starts; pendingAccessTimesDone is closed when it returns.
// pendingAccessTimesCancel, set by StartPendingAccessTimeWrites,
// cancels its context, which tells it to finish.
pendingVariantAccessTimes chan imagerequest.VariantKey
pendingSourceAccessTimes chan imagerequest.SourceKey
pendingAccessTimesDone chan struct{}
pendingAccessTimesCancel context.CancelFunc
// The counts of cache hits and misses, made and registered by
// registerCounters (see counters.go)
cacheHits prometheus.Counter
cacheMisses prometheus.Counter
// reconciliationPageSize is the most rows one read of a content table
// returns in the reconciliation pass. NewCache sets it to
// defaultReconciliationPageSize; tests set it smaller.
reconciliationPageSize int
// reconciliationReadTestHook, when set, is called after each of those
// reads with the number of rows the read covered, so tests can check
// that no read covers more than one page; production code leaves it
// nil.
reconciliationReadTestHook func(rows int)
}
// NewCache creates a new cache instance and registers its counters on
// metricsRegistry. pixad passes Prometheus' default registry, the one
// /metrics serves. It makes no database call.
func NewCache(
db *sql.DB, config Config, metricsRegistry prometheus.Registerer,
) (*Cache, error) {
log := config.Logger
if log == nil {
log = slog.Default()
}
if config.CacheTTL == 0 {
config.CacheTTL = DefaultCacheTTL
}
c := &Cache{
db: db,
config: config,
log: log,
evictionPressure: make(chan struct{}, 1),
evictionDone: make(chan struct{}),
pendingVariantAccessTimes: make(chan imagerequest.VariantKey, maxPendingAccessTimes),
pendingSourceAccessTimes: make(chan imagerequest.SourceKey, maxPendingAccessTimes),
pendingAccessTimesDone: make(chan struct{}),
reconciliationPageSize: defaultReconciliationPageSize,
}
err := c.registerCounters(metricsRegistry)
if err != nil {
return nil, err
}
if config.MaxBytes == 0 {
return c, nil
}
srcContent, err := NewFileStore(
filepath.Join(config.StateDir, "cache", "sources"),
)
if err != nil {
return nil, fmt.Errorf("failed to create source content storage: %w", err)
}
variants, err := NewFileStore(
filepath.Join(config.StateDir, "cache", "variants"),
)
if err != nil {
return nil, fmt.Errorf("failed to create variant storage: %w", err)
}
c.srcContent = srcContent
c.variants = variants
return c, nil
}
// 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 VariantFile{}, ErrNotFound
}
file, info, err := c.variants.Open(string(cacheKey))
if err != nil {
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 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 VariantFile{
Content: file,
Size: info.Size(),
ContentType: format.MIMEType(),
CheckedAt: checkedAt,
}, nil
}
// 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,
source SourceInfo,
content io.Reader,
) error {
if c.config.MaxBytes == 0 {
return nil
}
data, err := io.ReadAll(content)
if err != nil {
return fmt.Errorf("failed to read content: %w", err)
}
err = c.variants.Write(string(cacheKey), data)
if err != nil {
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, 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,
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)
}
c.notifyWritePressure()
return nil
}
// Source opens the cached source of src and returns a reader, its size in
// bytes and its format, read from the file's first bytes. A source whose row
// or file is missing is ErrNotFound, as is any source of a disabled cache; a
// file whose first bytes are not those of a source format is another error.
// A source it opens has its access time handed to the background writer
// without waiting (see touchSourceContent).
func (c *Cache) Source(
ctx context.Context, src imagerequest.Source,
) (io.ReadCloser, int64, imagerequest.Format, error) {
if c.config.MaxBytes == 0 {
return nil, 0, "", ErrNotFound
}
key := src.Key()
var rowExists bool
err := c.db.QueryRowContext(ctx, `
SELECT EXISTS(SELECT 1 FROM source_content WHERE source_key = ?)
`, string(key)).Scan(&rowExists)
if err != nil {
return nil, 0, "", fmt.Errorf("failed to lookup source: %w", err)
}
if !rowExists {
return nil, 0, "", ErrNotFound
}
file, info, err := c.srcContent.Open(string(key))
if err != nil {
return nil, 0, "", err
}
format, err := readFormat(file)
if err != nil {
_ = file.Close()
return nil, 0, "", err
}
c.touchSourceContent(key)
return file, info.Size(), format, nil
}
// 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
}
data, err := io.ReadAll(content)
if err != nil {
return "", fmt.Errorf("failed to read source content: %w", err)
}
key := src.Key()
err = c.srcContent.Write(string(key), data)
if err != nil {
return "", fmt.Errorf("failed to store source content: %w", err)
}
size := int64(len(data))
_, err = c.db.ExecContext(ctx, `
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,
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 {
c.log.Warn("failed to remove source file whose row was not written",
"source_key", key, "error", deleteErr)
}
return "", fmt.Errorf("failed to insert source content: %w", err)
}
c.notifyWritePressure()
return key, nil
}
// Failure reports whether src is in the negative cache. It only reads: a
// row past its expiry is not a hit, and the eviction goroutine deletes it
// with the other expired rows (see CleanExpired).
func (c *Cache) Failure(
ctx context.Context, src imagerequest.Source,
) (bool, error) {
var expiresAt time.Time
err := c.db.QueryRowContext(ctx, `
SELECT expires_at FROM negative_cache
WHERE source_host = ? AND source_path = ? AND source_query = ?
`, src.Host, src.Path, src.Query).Scan(&expiresAt)
if errors.Is(err, sql.ErrNoRows) {
return false, nil
}
if err != nil {
return false, fmt.Errorf("failed to check negative cache: %w", err)
}
expired := time.Now().After(expiresAt)
return !expired, nil
}
// PutFailure stores a negative cache entry for src, whose fetch failed.
func (c *Cache) PutFailure(ctx context.Context, src imagerequest.Source) error {
expiresAt := time.Now().UTC().Add(c.config.NegativeTTL)
_, err := c.db.ExecContext(ctx, `
INSERT INTO negative_cache
(source_host, source_path, source_query, expires_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(source_host, source_path, source_query) DO UPDATE SET
expires_at = excluded.expires_at
`, src.Host, src.Path, src.Query, expiresAt)
if err != nil {
return fmt.Errorf("failed to insert negative cache: %w", err)
}
return nil
}
// CleanExpired removes the negative cache rows past their expiry, one
// batch per statement, until none is left. Once ctx ends it starts no
// further batch and returns ctx's error. The eviction goroutine calls it
// at startup and on every periodic tick, so a row stays about
// NegativeTTL plus one eviction interval, whether or not its URL is
// requested again.
func (c *Cache) CleanExpired(ctx context.Context) error {
for {
if ctx.Err() != nil {
return ctx.Err()
}
deleted, err := c.cleanExpiredBatch(ctx)
if err != nil {
return err
}
if deleted < cleanExpiredBatchSize {
return nil
}
}
}
// IncrementStats counts a cache hit or miss. It only adds to the cache's
// Prometheus counters in memory (see counters.go), so it never waits for the
// database or fails.
func (c *Cache) IncrementStats(hit bool) {
if hit {
c.cacheHits.Inc()
} else {
c.cacheMisses.Inc()
}
}
// touchVariant hands the access time of a variant served from the cache to
// the goroutine StartPendingAccessTimeWrites starts, which writes it in its
// next batch. It never waits: while maxPendingAccessTimes access times of
// variants wait to be written, this one is dropped, and the variant only
// looks colder to eviction.
func (c *Cache) touchVariant(cacheKey imagerequest.VariantKey) {
select {
case c.pendingVariantAccessTimes <- cacheKey:
default:
}
}
// touchSourceContent hands the access time of a cached source a miss used to
// the goroutine StartPendingAccessTimeWrites starts, which writes it in its
// next batch. It never waits: while maxPendingAccessTimes access times of
// sources wait to be written, this one is dropped, and the source only looks
// colder to eviction.
func (c *Cache) touchSourceContent(key imagerequest.SourceKey) {
select {
case c.pendingSourceAccessTimes <- key:
default:
}
}
// cleanExpiredBatch deletes up to cleanExpiredBatchSize negative cache
// rows past their expiry in one statement, and returns how many it
// deleted. DELETE ... LIMIT needs a SQLite build option, so the subquery
// picks the rows.
func (c *Cache) cleanExpiredBatch(ctx context.Context) (int64, error) {
result, err := c.db.ExecContext(ctx, `
DELETE FROM negative_cache WHERE id IN (
SELECT id FROM negative_cache
WHERE expires_at < CURRENT_TIMESTAMP
LIMIT ?
)
`, cleanExpiredBatchSize)
if err != nil {
return 0, fmt.Errorf("failed to clean negative cache: %w", err)
}
deleted, err := result.RowsAffected()
if err != nil {
return 0, fmt.Errorf("failed to count cleaned negative cache rows: %w", err)
}
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 up to magic.MaxMagicBytes 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 seven formats; a variant is SVG only
// as orig of an SVG source. A file shorter than magic.MinMagicBytes is
// io.EOF.
func readFormat(content io.ReaderAt) (imagerequest.Format, error) {
firstBytes := make([]byte, magic.MaxMagicBytes)
n, err := content.ReadAt(firstBytes, 0)
if err != nil && (!errors.Is(err, io.EOF) || n < magic.MinMagicBytes) {
return "", fmt.Errorf("failed to read the first bytes: %w", err)
}
return magic.DetectFormat(firstBytes[:n])
}