Merge branch 'main' into golangci-v2.12.2
Absorbs the cache size management and LRU eviction work (#55). All four textual conflicts resolved in favor of main's implementation, with this branch's mechanical lint conformance re-applied on top: - internal/config/config.go: took main's cache_max_bytes wiring (CacheMaxBytes, cacheMaxBytesExplicit) verbatim and expressed the key through this branch's constant convention as keyCacheMaxBytes. - internal/imgcache/cache.go: took main's disabled-cache guards, LRU touch on lookup, and variant_content size accounting verbatim; re-applied this branch's signature wrapping for lll. - internal/imgcache/storage.go: took main's new writeIfAbsent helper and its temp-file cleanup defer verbatim. The auto-merge had silently dropped that defer in favor of this branch's inline cleanup form; restored so the cleanup semantics that arrived from main are intact. - TODO.md: kept both sides' Completed Steps entries. .golangci.yml resolves to this branch's canonical version (sha256 021cc83f4e6fc7c31b95b34b846723dfcf20b66b7baeea1dc40406e643346bcb). make build and make test are green; #55's code is not yet conformant with the canonical lint config, which the following commits address.
This commit is contained in:
@@ -2,12 +2,16 @@ package imgcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
@@ -27,6 +31,22 @@ type CacheConfig struct {
|
||||
StateDir string
|
||||
CacheTTL time.Duration
|
||||
NegativeTTL time.Duration
|
||||
|
||||
// MaxBytes is the disk cache size limit in bytes that eviction
|
||||
// enforces. Zero means no limit is enforced (no eviction). The
|
||||
// config layer supplies the computed default when the operator
|
||||
// omits cache_max_bytes.
|
||||
MaxBytes int64
|
||||
|
||||
// DisableDiskCache turns the disk cache off entirely: no cache
|
||||
// directories are created, lookups always miss, stores are
|
||||
// no-ops, and no eviction machinery runs. The config layer sets
|
||||
// this when the operator configures cache_max_bytes: 0.
|
||||
DisableDiskCache bool
|
||||
|
||||
// Logger receives accounting and eviction log output. A nil
|
||||
// Logger means slog.Default().
|
||||
Logger *slog.Logger
|
||||
}
|
||||
|
||||
// variantMeta stores content type for fast cache hits without reading .meta file.
|
||||
@@ -42,14 +62,61 @@ type Cache struct {
|
||||
variants *VariantStorage // processed variants by cache key
|
||||
srcMetadata *MetadataStorage // source metadata by host/path
|
||||
config CacheConfig
|
||||
log *slog.Logger
|
||||
|
||||
// disabled means the disk cache is turned off entirely: lookups
|
||||
// always miss, stores are no-ops, and no eviction runs.
|
||||
disabled bool
|
||||
|
||||
// Eviction machinery. The channels are created in NewCache so
|
||||
// stores can signal write pressure without racing StartEviction.
|
||||
evictionPressure chan struct{}
|
||||
evictionStop chan struct{}
|
||||
evictionDone chan struct{}
|
||||
evictionStarted bool
|
||||
evictionStopOnce sync.Once
|
||||
|
||||
// In-memory cache of variant metadata (content type, size) to avoid
|
||||
// reading .meta files
|
||||
metaCache map[VariantKey]variantMeta
|
||||
|
||||
// contentLocks serializes StoreSource and evictSourceBlob per
|
||||
// content hash, closing the race window between an eviction's row
|
||||
// deletion and its file unlink against a concurrent store of
|
||||
// identical content.
|
||||
contentLocks *contentLock
|
||||
|
||||
// evictSourceBlobTestHook, when set, is invoked by evictSourceBlob
|
||||
// after its row-deletion transaction commits and before the
|
||||
// content file is unlinked. It exists solely so tests can
|
||||
// deterministically pause inside that window to exercise
|
||||
// concurrent stores against it; production code leaves it nil.
|
||||
evictSourceBlobTestHook func(ContentHash)
|
||||
}
|
||||
|
||||
// NewCache creates a new cache instance.
|
||||
func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
|
||||
log := config.Logger
|
||||
if log == nil {
|
||||
log = slog.Default()
|
||||
}
|
||||
|
||||
c := &Cache{
|
||||
db: db,
|
||||
config: config,
|
||||
log: log,
|
||||
disabled: config.DisableDiskCache,
|
||||
evictionPressure: make(chan struct{}, 1),
|
||||
evictionStop: make(chan struct{}),
|
||||
evictionDone: make(chan struct{}),
|
||||
metaCache: make(map[VariantKey]variantMeta),
|
||||
contentLocks: newContentLock(),
|
||||
}
|
||||
|
||||
if c.disabled {
|
||||
return c, nil
|
||||
}
|
||||
|
||||
srcContent, err := NewContentStorage(
|
||||
filepath.Join(config.StateDir, "cache", "sources"),
|
||||
)
|
||||
@@ -71,14 +138,11 @@ func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
|
||||
return nil, fmt.Errorf("failed to create source metadata storage: %w", err)
|
||||
}
|
||||
|
||||
return &Cache{
|
||||
db: db,
|
||||
srcContent: srcContent,
|
||||
variants: variants,
|
||||
srcMetadata: srcMetadata,
|
||||
config: config,
|
||||
metaCache: make(map[VariantKey]variantMeta),
|
||||
}, nil
|
||||
c.srcContent = srcContent
|
||||
c.variants = variants
|
||||
c.srcMetadata = srcMetadata
|
||||
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// LookupResult contains the result of a cache lookup.
|
||||
@@ -90,12 +154,15 @@ type LookupResult struct {
|
||||
CacheStatus CacheStatus
|
||||
}
|
||||
|
||||
// Lookup checks if a processed variant exists on disk (no DB access for hits).
|
||||
func (c *Cache) Lookup(_ context.Context, req *ImageRequest) (*LookupResult, error) {
|
||||
// Lookup checks if a processed variant exists on disk. Hits touch the
|
||||
// variant's LRU timestamp; a disabled cache always misses.
|
||||
func (c *Cache) Lookup(ctx context.Context, req *ImageRequest) (*LookupResult, error) {
|
||||
cacheKey := CacheKey(req)
|
||||
|
||||
// Check variant storage directly - no DB needed for cache hits
|
||||
if c.variants.Exists(cacheKey) {
|
||||
if !c.disabled && c.variants.Exists(cacheKey) {
|
||||
c.touchVariant(ctx, cacheKey)
|
||||
|
||||
return &LookupResult{
|
||||
Hit: true,
|
||||
CacheKey: cacheKey,
|
||||
@@ -110,20 +177,80 @@ func (c *Cache) Lookup(_ context.Context, req *ImageRequest) (*LookupResult, err
|
||||
}, nil
|
||||
}
|
||||
|
||||
// touchVariant updates the LRU timestamp of a variant, best-effort:
|
||||
// a failed touch only makes the entry look colder to eviction.
|
||||
func (c *Cache) touchVariant(ctx context.Context, cacheKey VariantKey) {
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
UPDATE variant_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE cache_key = ?
|
||||
`, string(cacheKey))
|
||||
if err != nil {
|
||||
c.log.Debug("failed to touch variant LRU timestamp",
|
||||
"cache_key", cacheKey, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// touchSourceContent updates the LRU timestamp of a source content
|
||||
// blob, best-effort: a failed touch only makes the blob look colder.
|
||||
func (c *Cache) touchSourceContent(ctx context.Context, contentHash ContentHash) {
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE content_hash = ?
|
||||
`, string(contentHash))
|
||||
if err != nil {
|
||||
c.log.Debug("failed to touch source content LRU timestamp",
|
||||
"content_hash", contentHash, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// GetVariant returns a reader, size, and content type for a cached variant.
|
||||
func (c *Cache) GetVariant(cacheKey VariantKey) (io.ReadCloser, int64, string, error) {
|
||||
if c.disabled {
|
||||
return nil, 0, "", ErrNotFound
|
||||
}
|
||||
|
||||
return c.variants.LoadWithMeta(cacheKey)
|
||||
}
|
||||
|
||||
// StoreSource stores fetched source content and metadata.
|
||||
// StoreSource stores fetched source content and metadata. On a
|
||||
// disabled cache it is a no-op returning an empty hash.
|
||||
func (c *Cache) StoreSource(
|
||||
ctx context.Context,
|
||||
req *ImageRequest,
|
||||
content io.Reader,
|
||||
result *httpfetcher.FetchResult,
|
||||
) (ContentHash, error) {
|
||||
// Store content
|
||||
contentHash, size, err := c.srcContent.Store(content)
|
||||
if c.disabled {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
// Hash the content ourselves (rather than via srcContent.Store,
|
||||
// which would hash internally) so the content hash is known before
|
||||
// any file or database work happens: that lets the entire store be
|
||||
// serialized, per hash, against a concurrent eviction of the same
|
||||
// content below.
|
||||
data, err := io.ReadAll(content)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to read source content: %w", err)
|
||||
}
|
||||
|
||||
sum := sha256.Sum256(data)
|
||||
contentHash := ContentHash(hex.EncodeToString(sum[:]))
|
||||
|
||||
// Hold the content hash's lock for the whole store operation. A
|
||||
// concurrent eviction of this exact hash (the real SHA-256 dedup
|
||||
// case: a different source path whose bytes hash identically)
|
||||
// deletes the accounting rows and unlinks the file inside the same
|
||||
// lock, so the two can never interleave: either this store
|
||||
// completes first (and a subsequent eviction removes it together
|
||||
// with its rows and file, correctly), or eviction completes first
|
||||
// (and this store finds the file already gone and recreates it
|
||||
// fresh) — never a fresh row left pointing at a file eviction is
|
||||
// mid-unlink on.
|
||||
unlock := c.contentLocks.Lock(string(contentHash))
|
||||
defer unlock()
|
||||
|
||||
size, err := c.srcContent.StoreHashed(contentHash, data)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to store source content: %w", err)
|
||||
}
|
||||
@@ -180,23 +307,56 @@ func (c *Cache) StoreSource(
|
||||
// A failure here is non-fatal; the metadata is in the database.
|
||||
_ = c.srcMetadata.Store(req.SourceHost, pathHash, meta)
|
||||
|
||||
c.notifyWritePressure()
|
||||
|
||||
return contentHash, nil
|
||||
}
|
||||
|
||||
// StoreVariant stores a processed variant by its cache key.
|
||||
// StoreVariant 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).
|
||||
func (c *Cache) StoreVariant(
|
||||
cacheKey VariantKey, content io.Reader, contentType string,
|
||||
) error {
|
||||
_, err := c.variants.Store(cacheKey, content, contentType)
|
||||
if c.disabled {
|
||||
return nil
|
||||
}
|
||||
|
||||
return err
|
||||
size, err := c.variants.Store(cacheKey, content, contentType)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
_, err = c.db.Exec(`
|
||||
INSERT INTO variant_content (cache_key, size_bytes, content_type)
|
||||
VALUES (?, ?, ?)
|
||||
ON CONFLICT(cache_key) DO UPDATE SET
|
||||
size_bytes = excluded.size_bytes,
|
||||
content_type = excluded.content_type,
|
||||
last_accessed_at = CURRENT_TIMESTAMP
|
||||
`, string(cacheKey), size, contentType)
|
||||
if err != nil {
|
||||
c.log.Warn("failed to record variant in size accounting",
|
||||
"cache_key", cacheKey, "error", err)
|
||||
}
|
||||
|
||||
c.notifyWritePressure()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// LookupSource checks if we have cached source content for a request.
|
||||
// Returns the content hash and content type if found, or empty values if not.
|
||||
// Returns the content hash and content type if found, or empty values
|
||||
// if not. Hits touch the blob's LRU timestamp; a disabled cache always
|
||||
// reports no cached source.
|
||||
func (c *Cache) LookupSource(
|
||||
ctx context.Context, req *ImageRequest,
|
||||
) (ContentHash, string, error) {
|
||||
if c.disabled {
|
||||
return "", "", nil
|
||||
}
|
||||
|
||||
var hashStr, contentType string
|
||||
|
||||
err := c.db.QueryRowContext(ctx, `
|
||||
@@ -219,6 +379,8 @@ func (c *Cache) LookupSource(
|
||||
return "", "", nil
|
||||
}
|
||||
|
||||
c.touchSourceContent(ctx, contentHash)
|
||||
|
||||
return contentHash, contentType, nil
|
||||
}
|
||||
|
||||
@@ -265,6 +427,10 @@ func (c *Cache) GetSourceMetadataID(
|
||||
|
||||
// GetSourceContent returns a reader for cached source content by its hash.
|
||||
func (c *Cache) GetSourceContent(contentHash ContentHash) (io.ReadCloser, error) {
|
||||
if c.disabled {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
|
||||
return c.srcContent.Load(contentHash)
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user