diff --git a/README.md b/README.md index 0a99751..47bd93b 100644 --- a/README.md +++ b/README.md @@ -160,11 +160,13 @@ 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. +Source and transformed image files, and the `.meta` file beside each transformed +image, 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. +When pixa starts, and every five minutes after, it removes each cached source or +transformed image whose file is empty, or not the size the database records for +it, with a warning naming the file; the next request for it 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 diff --git a/TODO.md b/TODO.md index 8e28e7b..6a4b765 100644 --- a/TODO.md +++ b/TODO.md @@ -30,6 +30,18 @@ P2: security: per-origin rate limiting # Completed Steps +- 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, and + each transformed image's `.meta` file, are synced to disk before they are + renamed into place, and their directory after; the `.meta` file is written + before the image, and a failed directory sync removes the renamed file. The + reconciliation pass, at startup and every five minutes, removes a cached + source or transformed image whose file is empty, or not the size the database + records for it, with its rows and a warning naming the file, so the next + request fetches the source or converts the image again; requests compare no + sizes. 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-15 the example config allowlists no host (closes #249): it allowlisted `github.com` and `user-images.githubusercontent.com`, where anyone can put an image, so a deployment set up as Getting Started says served those images to @@ -38,15 +50,6 @@ 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 df21170..3d576a1 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 is on disk whole (see -// variantFileIsWhole). Hits touch the variant's LRU timestamp; a disabled -// cache always misses. +// 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) - if !c.disabled && c.variantFileIsWhole(ctx, cacheKey) { + // Check variant storage directly - no DB needed for cache hits + if !c.disabled && c.variants.Exists(cacheKey) { c.touchVariant(ctx, cacheKey) return &LookupResult{ @@ -393,11 +393,8 @@ 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. 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. +// 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) { @@ -405,18 +402,12 @@ func (c *Cache) LookupSource( return "", "", nil } - var ( - hashStr, contentType string - recordedSize int64 - ) + var hashStr, contentType string err := c.db.QueryRowContext(ctx, ` - SELECT source_metadata.content_hash, source_metadata.content_type, - source_content.size_bytes - FROM source_metadata JOIN source_content USING (content_hash) + SELECT content_hash, content_type FROM source_metadata WHERE source_host = ? AND source_path = ? AND source_query = ? - `, req.SourceHost, req.SourcePath, req.SourceQuery). - Scan(&hashStr, &contentType, &recordedSize) + `, req.SourceHost, req.SourcePath, req.SourceQuery).Scan(&hashStr, &contentType) if errors.Is(err, sql.ErrNoRows) { return "", "", nil @@ -428,24 +419,8 @@ func (c *Cache) LookupSource( contentHash := ContentHash(hashStr) - 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) - } - + // Verify the content file exists + if !c.srcContent.Exists(contentHash) { return "", "", nil } @@ -668,10 +643,10 @@ func (c *Cache) readHitAndMissCounts(ctx context.Context, stats *CacheStats) err // hold: it reads the content type from the variant's .meta file and // keeps it in metaCache, unless a StoreVariant has put one there // meanwhile, as the store's is newer. A read that finds no .meta file, -// as one can between a store's writing of the variant file and of its -// .meta file, serves application/octet-stream and keeps nothing, so -// metaCache only ever holds a type read from a .meta file or passed to -// StoreVariant. +// as one can when the store's .meta write failed or an eviction removed +// the file after the read opened the variant, serves +// application/octet-stream and keeps nothing, so metaCache only ever +// holds a type read from a .meta file or passed to StoreVariant. func (c *Cache) loadVariantWithMeta( cacheKey VariantKey, ) (io.ReadCloser, int64, string, error) { @@ -719,43 +694,6 @@ 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/eviction.go b/internal/imgcache/eviction.go index a5aff55..d5e4246 100644 --- a/internal/imgcache/eviction.go +++ b/internal/imgcache/eviction.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "encoding/json" + "errors" "fmt" "io/fs" "os" @@ -542,10 +543,11 @@ func (c *Cache) runReconciliationPass(ctx context.Context) { // (or whose accounting insert failed, e.g. StoreVariant's best-effort // insert under transient DB contention), drops accounting rows whose // files are missing, removes source blob files the database does not -// know (and rows whose files are gone), sweeps stale temp files left -// behind by crashed writes, and last checks the total cache usage -// against the tables. It reads the tables a page at a time. Running it -// periodically, not just once, bounds how long such drift can +// know (and rows whose files are gone), removes source and variant files +// that are empty or not the size their rows record, sweeps stale temp +// files left behind by crashed writes, and last checks the total cache +// usage against the tables. It reads the tables a page at a time. +// Running it periodically, not just once, bounds how long such drift can // accumulate unaccounted for on a long-running process to one eviction // interval. Once ctx is cancelled, it stops at the next file or row and // returns ctx's error. @@ -577,8 +579,9 @@ func (c *Cache) reconcileAccounting(ctx context.Context) error { return c.reconcileUsageTotal(ctx) } -// reconcileVariantFiles walks the variant storage directory, adopting -// files without accounting rows and sweeping stale temp files. +// reconcileVariantFiles walks the variant storage directory, checking +// each variant file against its accounting row (reconcileVariantFile) +// and sweeping stale temp files. func (c *Cache) reconcileVariantFiles(ctx context.Context) error { return filepath.WalkDir( c.variants.baseDir, @@ -603,38 +606,55 @@ func (c *Cache) reconcileVariantFiles(ctx context.Context) error { return nil } - return c.adoptVariantFile(ctx, path, entry, VariantKey(name)) + return c.reconcileVariantFile(ctx, path, entry, VariantKey(name)) }, ) } -// adoptVariantFile inserts an accounting row for a variant file that -// has none, using the file's size and modification time. -func (c *Cache) adoptVariantFile( +// reconcileVariantFile checks one variant file against its accounting +// row. A file without a row is adopted into the accounting, unless it is +// empty. A file that is empty, or not the size its row records, as a +// crash or a restored snapshot can leave it, is removed with its row and +// a warning naming it, so the variant is converted again. +func (c *Cache) reconcileVariantFile( ctx context.Context, path string, entry fs.DirEntry, cacheKey VariantKey, ) error { - var rowExists int - - err := c.db.QueryRowContext(ctx, - `SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, string(cacheKey), - ).Scan(&rowExists) - if err != nil { - return fmt.Errorf("failed to check variant accounting row: %w", err) - } - - if rowExists > 0 { - return nil - } - info, err := entry.Info() if err != nil { return fmt.Errorf("failed to stat variant file: %w", err) } + var recordedSize int64 + + err = c.db.QueryRowContext(ctx, + `SELECT size_bytes FROM variant_content WHERE cache_key = ?`, string(cacheKey), + ).Scan(&recordedSize) + + switch { + case errors.Is(err, sql.ErrNoRows) && info.Size() > 0: + return c.adoptVariantFile(ctx, path, info, cacheKey) + case err != nil && !errors.Is(err, sql.ErrNoRows): + return fmt.Errorf("failed to check variant accounting row: %w", err) + case info.Size() > 0 && info.Size() == recordedSize: + return nil + } + + // The file is empty, or not the size its row records + c.log.Warn("removing damaged cached variant file", "path", path, + "file_bytes", info.Size(), "recorded_bytes", recordedSize) + + return c.evictVariant(ctx, cacheKey) +} + +// adoptVariantFile inserts an accounting row for a variant file that +// has none, using the file's size and modification time. +func (c *Cache) adoptVariantFile( + ctx context.Context, path string, info fs.FileInfo, cacheKey VariantKey, +) error { modTime := info.ModTime().UTC().Format(sqliteTimestampLayout) contentType := c.variantContentTypeFromSidecar(path) - _, err = c.db.ExecContext(ctx, ` + _, err := c.db.ExecContext(ctx, ` INSERT INTO variant_content (cache_key, size_bytes, content_type, created_at, last_accessed_at) VALUES (?, ?, ?, ?, ?) @@ -753,10 +773,9 @@ func queryStringColumn[T ~string]( return values, nil } -// reconcileSourceFiles walks the source content directory, removing -// blob files the database does not track (they are unreachable: source -// lookups always go through source_metadata) and sweeping stale temp -// files. +// reconcileSourceFiles walks the source content directory, checking +// each blob file against its source_content row (reconcileSourceFile) +// and sweeping stale temp files. func (c *Cache) reconcileSourceFiles(ctx context.Context) error { return filepath.WalkDir( c.srcContent.baseDir, @@ -777,31 +796,55 @@ func (c *Cache) reconcileSourceFiles(ctx context.Context) error { return nil } - return c.removeUntrackedSourceFile(ctx, path, ContentHash(name)) + return c.reconcileSourceFile(ctx, path, entry, ContentHash(name)) }, ) } +// reconcileSourceFile checks one source blob file against its +// source_content row. A file without a row is removed: it is +// unreachable, as source lookups always go through source_metadata. A +// file that is empty, or not the size its row records, as a crash or a +// restored snapshot can leave it, is removed with its rows and a warning +// naming it, so the source is fetched again. +func (c *Cache) reconcileSourceFile( + ctx context.Context, path string, entry fs.DirEntry, contentHash ContentHash, +) error { + info, err := entry.Info() + if err != nil { + return fmt.Errorf("failed to stat source content file: %w", err) + } + + var recordedSize int64 + + err = c.db.QueryRowContext(ctx, + `SELECT size_bytes FROM source_content WHERE content_hash = ?`, + string(contentHash), + ).Scan(&recordedSize) + + switch { + case errors.Is(err, sql.ErrNoRows): + return c.removeUntrackedSourceFile(ctx, path, contentHash) + case err != nil: + return fmt.Errorf("failed to check source content row: %w", err) + case info.Size() > 0 && info.Size() == recordedSize: + return nil + } + + // The file is empty, or not the size its row records + c.log.Warn("removing damaged cached source file", "path", path, + "file_bytes", info.Size(), "recorded_bytes", recordedSize) + + return c.evictSourceBlob(ctx, contentHash) +} + // removeUntrackedSourceFile deletes a source blob file that has no // source_content row. Any source_metadata rows referencing the hash // are removed first so no row ever points at a deleted file. func (c *Cache) removeUntrackedSourceFile( ctx context.Context, path string, contentHash ContentHash, ) error { - var rowExists int - - err := c.db.QueryRowContext(ctx, - `SELECT COUNT(*) FROM source_content WHERE content_hash = ?`, string(contentHash), - ).Scan(&rowExists) - if err != nil { - return fmt.Errorf("failed to check source content row: %w", err) - } - - if rowExists > 0 { - return nil - } - - _, err = c.db.ExecContext(ctx, + _, err := c.db.ExecContext(ctx, `DELETE FROM source_metadata WHERE content_hash = ?`, string(contentHash)) if err != nil { return fmt.Errorf("failed to delete metadata rows for untracked blob: %w", err) diff --git a/internal/imgcache/service.go b/internal/imgcache/service.go index 9ac5d70..52684a9 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 + // Check variant cache first (disk only, no DB) 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 249e2d3..f14781e 100644 --- a/internal/imgcache/storage.go +++ b/internal/imgcache/storage.go @@ -151,24 +151,8 @@ func (s *ContentStorage) Exists(hash ContentHash) bool { return err == nil } -// 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. +// writeIfAbsent writes data with writeFileSynced to the path derived +// from hash, unless a file of the data's size is already there. func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error { // Build path: /// path := s.hashToPath(hash) @@ -179,15 +163,23 @@ func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error { return nil } - // Create directory structure - dir := filepath.Dir(path) - - err := os.MkdirAll(dir, StorageDirPerm) + err := os.MkdirAll(filepath.Dir(path), StorageDirPerm) if err != nil { return fmt.Errorf("failed to create directory: %w", err) } - // Write to temp file first, then rename for atomicity + return writeFileSynced(path, data) +} + +// writeFileSynced writes data to path through a temp file in the same +// directory: the temp file is synced to disk, renamed to path, and the +// directory synced after, so that neither a concurrent reader nor a +// crash finds a partial file under path. When the directory sync fails, +// the file is removed from path, so that a failed write leaves no file +// there. +func writeFileSynced(path string, data []byte) error { + dir := filepath.Dir(path) + tmpFile, err := os.CreateTemp(dir, ".tmp-*") if err != nil { return fmt.Errorf("failed to create temp file: %w", err) @@ -195,18 +187,12 @@ func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error { tmpPath := tmpFile.Name() - // Each failure path below unlinks the temp file explicitly. This - // replaces a deferred cleanup that read a named result, which the - // canonical config does not permit; the set of paths that remove - // tmpPath, and the order relative to Close, is unchanged. This - // mirrors how MetadataStorage.Store and VariantStorage.Store below - // already express the same cleanup. _, err = tmpFile.Write(data) if err != nil { _ = tmpFile.Close() _ = os.Remove(tmpPath) - return fmt.Errorf("failed to write content: %w", err) + return fmt.Errorf("failed to write temp file: %w", err) } err = tmpFile.Sync() @@ -224,7 +210,6 @@ func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error { return fmt.Errorf("failed to close temp file: %w", err) } - // Atomic rename err = os.Rename(filepath.Clean(tmpPath), filepath.Clean(path)) if err != nil { _ = os.Remove(tmpPath) @@ -232,7 +217,14 @@ func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error { return fmt.Errorf("failed to rename temp file: %w", err) } - return syncDirectory(dir) + err = syncDirectory(dir) + if err != nil { + _ = os.Remove(path) + + return err + } + + return nil } // syncDirectory syncs the directory dir to disk, so that a file just @@ -457,7 +449,9 @@ func NewVariantStorage(baseDir string, log *slog.Logger) (*VariantStorage, error return &VariantStorage{baseDir: baseDir, log: log}, nil } -// Store writes content and metadata to storage at the given key. +// Store writes content and metadata to storage at the given key, each +// with writeFileSynced. The .meta file is written first, so that a crash +// cannot leave the content in place without it. func (s *VariantStorage) Store( key VariantKey, r io.Reader, contentType string, ) (int64, error) { @@ -470,61 +464,11 @@ func (s *VariantStorage) Store( path := s.keyToPath(key) metaPath := path + ".meta" - // Create directory structure - dir := filepath.Dir(path) - - err = os.MkdirAll(dir, StorageDirPerm) + err = os.MkdirAll(filepath.Dir(path), StorageDirPerm) if err != nil { return 0, fmt.Errorf("failed to create directory: %w", err) } - // 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) - } - - tmpPath := tmpFile.Name() - - _, err = tmpFile.Write(data) - if err != nil { - _ = tmpFile.Close() - _ = os.Remove(tmpPath) - - 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) - - return 0, fmt.Errorf("failed to close temp file: %w", err) - } - - // Atomic rename content - err = os.Rename(filepath.Clean(tmpPath), filepath.Clean(path)) - if err != nil { - _ = os.Remove(tmpPath) - - 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, Size: size, @@ -536,13 +480,19 @@ func (s *VariantStorage) Store( return 0, fmt.Errorf("failed to marshal metadata: %w", err) } - // Metadata write failure is non-fatal; content is already stored. - err = os.WriteFile(metaPath, metaData, StorageFilePerm) + // Metadata write failure is non-fatal: the content is then served as + // application/octet-stream. + err = writeFileSynced(metaPath, metaData) if err != nil { s.log.Warn("failed to write variant metadata sidecar", "path", metaPath, "error", err) } + err = writeFileSynced(path, data) + if err != nil { + return 0, err + } + return size, nil } @@ -620,20 +570,6 @@ 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 {