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. #252 Model: opus-5-5
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
+73
-11
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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: <basedir>/<ab>/<cd>/<hash>
|
||||
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 {
|
||||
|
||||
Reference in New Issue
Block a user