Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
36c9e1408f | ||
|
|
59063c3e26 | ||
|
|
36a4ce22c3 | ||
|
|
a1fc42b80e | ||
|
|
88e1b1cdfa |
@@ -26,6 +26,11 @@ make docker
|
||||
docker run -p 8080:8080 -e PIXA_SIGNING_KEY="$(openssl rand -base64 32)" pixa:latest
|
||||
```
|
||||
|
||||
The example config allowlists no host (`allowlist_hosts`), so pixa serves an
|
||||
image only for a signed or encrypted URL. Every image on an allowlisted host is
|
||||
served to anyone without a signature, at any size and in any format, so
|
||||
allowlist only hosts you control, where nobody else can put images.
|
||||
|
||||
A container takes its settings from environment variables (see Configuration
|
||||
below for the list). Only `PIXA_SIGNING_KEY` is required; if it is unset the
|
||||
container exits at startup naming the variable. Everything else has a built-in
|
||||
@@ -115,7 +120,9 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
|
||||
- `PIXA_SIGNING_KEY` (required): secret for signed and encrypted URLs and
|
||||
login, 32+ characters, for example from `openssl rand -base64 32`
|
||||
- `PIXA_ALLOWLIST_HOSTS`: upstream hosts served without a signature,
|
||||
comma-separated
|
||||
comma-separated; default none. Every image on a listed host is served to
|
||||
anyone without a signature, so list only hosts you control, where nobody
|
||||
else can put images.
|
||||
- `PIXA_CACHE_MAX_BYTES`: disk cache limit in bytes; `0` disables it;
|
||||
default 75% of (free space + what the cache holds)
|
||||
- the rest are in the table under Configuration below
|
||||
@@ -153,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
|
||||
|
||||
@@ -31,14 +31,25 @@ 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 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.
|
||||
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
|
||||
anyone without a signature, at any size and in any format.
|
||||
`configs/config.example.yml` now leaves `allowlist_hosts` out, with one
|
||||
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-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
|
||||
|
||||
@@ -45,15 +45,14 @@ state_dir: ./data
|
||||
signing_key: "CHANGE_ME_generate_with_openssl_rand_base64_32"
|
||||
|
||||
# Hosts that don't require signatures (default: none)
|
||||
# Every image on an allowlisted host is served to anyone without a
|
||||
# signature, at any size and in any format, so list only hosts you
|
||||
# control, where nobody else can put images.
|
||||
# Use "." prefix for wildcard subdomain matching (e.g., ".example.com" matches "cdn.example.com")
|
||||
# An entry that is neither a host name nor an IP address (IPv6 without
|
||||
# brackets), such as one with a port or a "*." wildcard, aborts startup.
|
||||
allowlist_hosts:
|
||||
- s3.sneak.cloud
|
||||
- static.sneak.cloud
|
||||
- sneak.berlin
|
||||
- github.com
|
||||
- user-images.githubusercontent.com
|
||||
# allowlist_hosts:
|
||||
# - images.example.com
|
||||
|
||||
# Hosts whose pages may not show pixa's images, written as for
|
||||
# allowlist_hosts. A request to /v1/image/ or /v1/e/ whose Referer header
|
||||
|
||||
@@ -92,6 +92,32 @@ func TestGetStringSlice_Empty(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestExampleConfigAllowlistsNoHost loads configs/config.example.yml
|
||||
// with a real signing key, as Getting Started in README.md has it copied
|
||||
// and edited, and checks that it allowlists no host: every image on an
|
||||
// allowlisted host is served to anyone without a signature.
|
||||
func TestExampleConfigAllowlistsNoHost(t *testing.T) {
|
||||
t.Setenv("PIXA_SIGNING_KEY", validTestSigningKey)
|
||||
|
||||
examplePath := filepath.Join("..", "..", "configs", "config.example.yml")
|
||||
|
||||
sc, err := loadTestConfig(examplePath)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to load %s: %v", examplePath, err)
|
||||
}
|
||||
|
||||
c, err := newFromSmartConfig(sc)
|
||||
if err != nil {
|
||||
t.Fatalf("%s is not a valid config: %v", examplePath, err)
|
||||
}
|
||||
|
||||
t.Logf("allowlist_hosts in %s: %v", examplePath, c.AllowlistHosts)
|
||||
|
||||
if len(c.AllowlistHosts) != 0 {
|
||||
t.Errorf("%s allowlists %v, want no host", examplePath, c.AllowlistHosts)
|
||||
}
|
||||
}
|
||||
|
||||
// loadTestConfig is a helper to load a config file for testing.
|
||||
func loadTestConfig(path string) (*smartconfig.Config, error) {
|
||||
return smartconfig.NewFromConfigPath(path)
|
||||
|
||||
+15
-77
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
+36
-100
@@ -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: <basedir>/<ab>/<cd>/<hash>
|
||||
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 {
|
||||
|
||||
Reference in New Issue
Block a user