From 36a4ce22c3b0ab1ca7c7cb64ba1e6f797c31eb56 Mon Sep 17 00:00:00 2001 From: sneak Date: Thu, 15 Oct 2026 02:41:49 +0000 Subject: [PATCH] Sync cached files and refuse ones left empty or partial (closes #252) Source and variant files are synced to disk before the rename into place, and their directory after. Lookup and LookupSource compare the file's size with the size its row records: a file that is empty or of another size is removed with its rows and a warning naming it, and the request converts the image or fetches the source again. A variant whose row cannot be read is judged by its file alone. Storing a source over a file of another size replaces it. A source that fails to decode is not removed for that alone. https://git.eeqj.de/sneak/pixa/issues/252 Model: opus-5-5 --- README.md | 6 +++ TODO.md | 9 ++++ internal/imgcache/cache.go | 84 +++++++++++++++++++++++++++++----- internal/imgcache/service.go | 2 +- internal/imgcache/storage.go | 88 ++++++++++++++++++++++++++++++++---- 5 files changed, 168 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index 250463e..0a99751 100644 --- a/README.md +++ b/README.md @@ -160,6 +160,12 @@ name. Multiple source paths may reference the same content blob; the database tracks references rather than using filesystem refcounting. +Source and transformed image files are written under a temporary name, synced to +disk, and only then renamed into place, so a crash leaves no file under its own +name with part of its data. A request that finds a cached source or transformed +image whose file is empty, or not the size the database records for it, removes +it and fetches the source or converts the image again. + pixa's target is 1-5k r/s, which has not been measured at that rate (see Load Test). Toward it, pixa keeps in memory the content types of the 10,000 transformed images most recently cached or served, so a cache hit on one of them diff --git a/TODO.md b/TODO.md index b85bf37..8e28e7b 100644 --- a/TODO.md +++ b/TODO.md @@ -38,6 +38,15 @@ P2: security: per-origin rate limiting commented entry under `example.com`. Getting Started and "Running under upaas" in `README.md` say that every image on an allowlisted host is served without a signature, so an allowlisted host must be one the operator controls. +- 2026-10-15 a cached file left empty or partial is no longer served, or fetched + again on every request (closes #252). Source and transformed image files are + synced to disk before they are renamed into place, and their directory after. + A request that finds a cached source or transformed image whose file is empty, + or not the size the database records for it, removes it with a log line naming + it, and fetches the source or converts the image again. Storing a source over + a file of another size replaces that file. A source that fails to decode is + not removed for that alone: a damaged upstream image would then be fetched on + every request. - 2026-10-13 `fit=outside` no longer makes an output above 8192 on a side (closes #257): it scales the image to cover the requested size, and the side scaled past that size was held to no maximum, so `1x8192?fit=outside` of a diff --git a/internal/imgcache/cache.go b/internal/imgcache/cache.go index 1f73019..df21170 100644 --- a/internal/imgcache/cache.go +++ b/internal/imgcache/cache.go @@ -222,13 +222,13 @@ type LookupResult struct { CacheStatus CacheStatus } -// Lookup checks if a processed variant exists on disk. Hits touch the -// variant's LRU timestamp; a disabled cache always misses. +// Lookup checks if a processed variant is on disk whole (see +// variantFileIsWhole). 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.disabled && c.variants.Exists(cacheKey) { + if !c.disabled && c.variantFileIsWhole(ctx, cacheKey) { c.touchVariant(ctx, cacheKey) return &LookupResult{ @@ -393,8 +393,11 @@ func (c *Cache) StoreVariant( // 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. Hits touch the blob's LRU timestamp; a disabled cache always -// reports no cached source. +// if not. A content file that is empty, or not the size its +// source_content row records, is damaged: it is removed with its rows, +// and no cached source is reported, so the source is fetched again. 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) { @@ -402,12 +405,18 @@ func (c *Cache) LookupSource( return "", "", nil } - var hashStr, contentType string + var ( + hashStr, contentType string + recordedSize int64 + ) err := c.db.QueryRowContext(ctx, ` - SELECT content_hash, content_type FROM source_metadata + SELECT source_metadata.content_hash, source_metadata.content_type, + source_content.size_bytes + FROM source_metadata JOIN source_content USING (content_hash) WHERE source_host = ? AND source_path = ? AND source_query = ? - `, req.SourceHost, req.SourcePath, req.SourceQuery).Scan(&hashStr, &contentType) + `, req.SourceHost, req.SourcePath, req.SourceQuery). + Scan(&hashStr, &contentType, &recordedSize) if errors.Is(err, sql.ErrNoRows) { return "", "", nil @@ -419,8 +428,24 @@ func (c *Cache) LookupSource( contentHash := ContentHash(hashStr) - // Verify the content file exists - if !c.srcContent.Exists(contentHash) { + fileSize, err := c.srcContent.Size(contentHash) + if errors.Is(err, ErrNotFound) { + return "", "", nil + } + + if err != nil { + return "", "", err + } + + if fileSize == 0 || fileSize != recordedSize { + c.log.Warn("removing damaged cached source", "content_hash", contentHash, + "file_bytes", fileSize, "recorded_bytes", recordedSize) + + err = c.evictSourceBlob(ctx, contentHash) + if err != nil { + return "", "", fmt.Errorf("failed to remove damaged source: %w", err) + } + return "", "", nil } @@ -694,6 +719,43 @@ func (c *Cache) writeMetadataSidecar( } } +// variantFileIsWhole reports whether the variant file of cacheKey is on +// disk with the size its variant_content row records. A file that is +// empty or of another size is damaged: it is removed with its row, so the +// variant is converted again. A variant whose row cannot be read, as +// between StoreVariant's writing of its file and of its row, is judged +// by its file alone. +func (c *Cache) variantFileIsWhole(ctx context.Context, cacheKey VariantKey) bool { + fileSize, err := c.variants.Size(cacheKey) + if err != nil { + return false + } + + var recordedSize int64 + + err = c.db.QueryRowContext(ctx, + `SELECT size_bytes FROM variant_content WHERE cache_key = ?`, + string(cacheKey)).Scan(&recordedSize) + if err != nil { + recordedSize = fileSize + } + + if fileSize > 0 && fileSize == recordedSize { + return true + } + + c.log.Warn("removing damaged cached variant", "cache_key", cacheKey, + "file_bytes", fileSize, "recorded_bytes", recordedSize) + + err = c.evictVariant(ctx, cacheKey) + if err != nil { + c.log.Warn("failed to remove damaged cached variant", + "cache_key", cacheKey, "error", err) + } + + return false +} + // 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) { diff --git a/internal/imgcache/service.go b/internal/imgcache/service.go index 52684a9..9ac5d70 100644 --- a/internal/imgcache/service.go +++ b/internal/imgcache/service.go @@ -148,7 +148,7 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e return nil, fmt.Errorf("%w: %w", httpfetcher.ErrUpstreamError, ErrNegativeCached) } - // Check variant cache first (disk only, no DB) + // Check variant cache first result, err := s.cache.Lookup(ctx, req) if err != nil { s.log.Warn("cache lookup failed", "error", err) diff --git a/internal/imgcache/storage.go b/internal/imgcache/storage.go index 395178d..249e2d3 100644 --- a/internal/imgcache/storage.go +++ b/internal/imgcache/storage.go @@ -79,8 +79,8 @@ func (s *ContentStorage) Store(r io.Reader) (ContentHash, int64, error) { // from hash, without recomputing it. Callers that already know the // hash before writing (e.g. because they must hold a hash-keyed lock // across the whole store operation) use this instead of Store. Like -// Store, it is idempotent: content already on disk at that path is -// left untouched. +// Store, it is idempotent: a file of the content's size already at that +// path is left untouched, and a file of another size is replaced. func (s *ContentStorage) StoreHashed( hash ContentHash, data []byte, ) (int64, error) { @@ -151,16 +151,31 @@ func (s *ContentStorage) Exists(hash ContentHash) bool { return err == nil } -// writeIfAbsent writes data to the path derived from hash, unless -// content already exists there, via a temp-file-plus-rename so -// concurrent readers never observe a partial file. +// Size returns the size in bytes of the content with the given hash. +func (s *ContentStorage) Size(hash ContentHash) (int64, error) { + info, err := os.Stat(s.hashToPath(hash)) + if err != nil { + if os.IsNotExist(err) { + return 0, ErrNotFound + } + + return 0, fmt.Errorf("failed to stat content: %w", err) + } + + return info.Size(), nil +} + +// writeIfAbsent writes data to the path derived from hash, unless a +// file of the data's size is already there. It writes a temp file, +// syncs it to disk and renames it into place, so neither a concurrent +// reader nor a crash finds a partial file under that path. func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error { // Build path: /// path := s.hashToPath(hash) - // Check if already exists - _, statErr := os.Stat(path) - if statErr == nil { + // A file of another size there is damaged, and is replaced + info, statErr := os.Stat(path) + if statErr == nil && info.Size() == int64(len(data)) { return nil } @@ -194,6 +209,14 @@ func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error { return fmt.Errorf("failed to write content: %w", err) } + err = tmpFile.Sync() + if err != nil { + _ = tmpFile.Close() + _ = os.Remove(tmpPath) + + return fmt.Errorf("failed to sync temp file: %w", err) + } + err = tmpFile.Close() if err != nil { _ = os.Remove(tmpPath) @@ -209,6 +232,24 @@ func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error { return fmt.Errorf("failed to rename temp file: %w", err) } + return syncDirectory(dir) +} + +// syncDirectory syncs the directory dir to disk, so that a file just +// renamed into it is still there after a crash. +func syncDirectory(dir string) error { + d, err := os.Open(filepath.Clean(dir)) + if err != nil { + return fmt.Errorf("failed to open directory: %w", err) + } + + err = d.Sync() + _ = d.Close() + + if err != nil { + return fmt.Errorf("failed to sync directory: %w", err) + } + return nil } @@ -437,7 +478,9 @@ func (s *VariantStorage) Store( return 0, fmt.Errorf("failed to create directory: %w", err) } - // Write content to temp file first, then rename for atomicity + // Write content to temp file first, sync it to disk, then rename for + // atomicity, so neither a concurrent reader nor a crash finds a + // partial file under the key's path tmpFile, err := os.CreateTemp(dir, ".tmp-*") if err != nil { return 0, fmt.Errorf("failed to create temp file: %w", err) @@ -453,6 +496,14 @@ func (s *VariantStorage) Store( return 0, fmt.Errorf("failed to write content: %w", err) } + err = tmpFile.Sync() + if err != nil { + _ = tmpFile.Close() + _ = os.Remove(tmpPath) + + return 0, fmt.Errorf("failed to sync temp file: %w", err) + } + err = tmpFile.Close() if err != nil { _ = os.Remove(tmpPath) @@ -468,6 +519,11 @@ func (s *VariantStorage) Store( return 0, fmt.Errorf("failed to rename temp file: %w", err) } + err = syncDirectory(dir) + if err != nil { + return 0, err + } + // Write metadata meta := VariantMeta{ ContentType: contentType, @@ -564,6 +620,20 @@ func (s *VariantStorage) Exists(key VariantKey) bool { return err == nil } +// Size returns the size in bytes of the content at the given key. +func (s *VariantStorage) Size(key VariantKey) (int64, error) { + info, err := os.Stat(s.keyToPath(key)) + if err != nil { + if os.IsNotExist(err) { + return 0, ErrNotFound + } + + return 0, fmt.Errorf("failed to stat content: %w", err) + } + + return info.Size(), nil +} + // Delete removes the content at the given key together with its .meta // sidecar file. A missing file is not an error. func (s *VariantStorage) Delete(key VariantKey) error {