pixa/internal/imgcache/cache.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 cache's metrics, made and registered by registerMetrics (see
	// metrics.go)
	cacheHits         prometheus.Counter
	cacheMisses       prometheus.Counter
	sourceCacheHits   prometheus.Counter
	sourceCacheMisses prometheus.Counter
	usageBytes        prometheus.Gauge
	maxBytes          prometheus.Gauge
	evictions         prometheus.Counter
	evictedBytes      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 metrics 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.registerMetrics(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
		}
	}
}

// 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 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.
func readFormat(content io.ReaderAt) (imagerequest.Format, error) {
	firstBytes := make([]byte, magic.MinMagicBytes)

	_, err := content.ReadAt(firstBytes, 0)
	if err != nil {
		return "", fmt.Errorf("failed to read the first bytes: %w", err)
	}

	return magic.DetectFormat(firstBytes)
}