Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
aee645faf8 | ||
|
|
ae74b46f1a | ||
|
|
4f8e304aee |
@@ -153,6 +153,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
|
||||
|
||||
@@ -30,6 +30,15 @@ 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.
|
||||
- 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) {
|
||||
|
||||
@@ -0,0 +1,359 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// getImage calls Get for req and returns the image it answered with and
|
||||
// its cache status.
|
||||
func getImage(t *testing.T, svc *Service, req *ImageRequest) ([]byte, CacheStatus) {
|
||||
t.Helper()
|
||||
|
||||
resp, err := svc.Get(t.Context(), req)
|
||||
if err != nil {
|
||||
t.Fatalf("Get() of quality %d error = %v", req.Quality, err)
|
||||
}
|
||||
|
||||
defer func() { _ = resp.Content.Close() }()
|
||||
|
||||
body, err := io.ReadAll(resp.Content)
|
||||
if err != nil {
|
||||
t.Fatalf("reading the image of quality %d: %v", req.Quality, err)
|
||||
}
|
||||
|
||||
return body, resp.CacheStatus
|
||||
}
|
||||
|
||||
// fileSize returns the size of the file at path.
|
||||
func fileSize(t *testing.T, path string) int64 {
|
||||
t.Helper()
|
||||
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
t.Fatalf("stat %s: %v", path, err)
|
||||
}
|
||||
|
||||
return info.Size()
|
||||
}
|
||||
|
||||
// truncateFile cuts the file at path down to size bytes.
|
||||
func truncateFile(t *testing.T, path string, size int64) {
|
||||
t.Helper()
|
||||
|
||||
t.Logf("cutting %s from %d to %d bytes",
|
||||
filepath.Base(path), fileSize(t, path), size)
|
||||
|
||||
err := os.Truncate(path, size)
|
||||
if err != nil {
|
||||
t.Fatalf("truncating %s: %v", path, err)
|
||||
}
|
||||
}
|
||||
|
||||
// cachedSourcePath returns the path of the cached source file req uses.
|
||||
func cachedSourcePath(t *testing.T, svc *Service, req *ImageRequest) string {
|
||||
t.Helper()
|
||||
|
||||
contentHash, _, err := svc.cache.LookupSource(t.Context(), req)
|
||||
if err != nil || contentHash == "" {
|
||||
t.Fatalf("LookupSource() = %q, %v; want the cached source", contentHash, err)
|
||||
}
|
||||
|
||||
return svc.cache.srcContent.hashToPath(contentHash)
|
||||
}
|
||||
|
||||
// reconcileAsAtStartup runs the cache's reconciliation pass, as pixa does
|
||||
// when it starts, and returns what the pass logged.
|
||||
func reconcileAsAtStartup(t *testing.T, cache *Cache) string {
|
||||
t.Helper()
|
||||
|
||||
var logBuf bytes.Buffer
|
||||
|
||||
cache.log = slog.New(slog.NewJSONHandler(&logBuf, nil))
|
||||
|
||||
err := cache.reconcileAccounting(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("reconcileAccounting() error = %v", err)
|
||||
}
|
||||
|
||||
t.Logf("reconciliation log: %s", logBuf.String())
|
||||
|
||||
return logBuf.String()
|
||||
}
|
||||
|
||||
// assertWarningNames checks that logs hold a warning naming the file at
|
||||
// path.
|
||||
func assertWarningNames(t *testing.T, logs, path string) {
|
||||
t.Helper()
|
||||
|
||||
for line := range strings.Lines(logs) {
|
||||
if strings.Contains(line, `"level":"WARN"`) &&
|
||||
strings.Contains(line, `"path":"`+path+`"`) {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
t.Errorf("reconciliation logged no warning naming %s", path)
|
||||
}
|
||||
|
||||
// TestService_Get_FetchesAgainForEmptyCachedSource empties the cached source
|
||||
// file, as a crash or a restored snapshot could, and runs the reconciliation
|
||||
// pass, which must remove it with a warning naming it. It then asks for two
|
||||
// new sizes of the source: the first must fetch it again and cache it whole,
|
||||
// so that the second is made from the cache without a third fetch.
|
||||
func TestService_Get_FetchesAgainForEmptyCachedSource(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures := SetupTestService(t)
|
||||
|
||||
getImage(t, svc, photoVariant(fixtures, 85, FitCover))
|
||||
|
||||
path := cachedSourcePath(t, svc, photoVariant(fixtures, 85, FitCover))
|
||||
wholeSize := fileSize(t, path)
|
||||
|
||||
truncateFile(t, path, 0)
|
||||
assertWarningNames(t, reconcileAsAtStartup(t, svc.cache), path)
|
||||
|
||||
getImage(t, svc, photoVariant(fixtures, 40, FitCover))
|
||||
getImage(t, svc, photoVariant(fixtures, 30, FitCover))
|
||||
|
||||
fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
|
||||
t.Logf("upstream fetches = %d", fetches)
|
||||
|
||||
if fetches != 2 {
|
||||
t.Errorf("upstream fetches = %d, want 2: the empty cached source "+
|
||||
"was kept and fetched again on every new size", fetches)
|
||||
}
|
||||
|
||||
size := fileSize(t, path)
|
||||
if size != wholeSize {
|
||||
t.Errorf("cached source is %d bytes, want %d", size, wholeSize)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Get_FetchesAgainForPartialCachedSource cuts the cached source
|
||||
// file to half its size, as a crash or a restored snapshot could, and runs
|
||||
// the reconciliation pass, which must remove it with a warning naming it.
|
||||
// It then asks for two new sizes of the source: the first must fetch it
|
||||
// again rather than fail to decode the partial file, and the second is
|
||||
// made from the cache.
|
||||
func TestService_Get_FetchesAgainForPartialCachedSource(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures := SetupTestService(t)
|
||||
|
||||
getImage(t, svc, photoVariant(fixtures, 85, FitCover))
|
||||
|
||||
path := cachedSourcePath(t, svc, photoVariant(fixtures, 85, FitCover))
|
||||
wholeSize := fileSize(t, path)
|
||||
|
||||
truncateFile(t, path, wholeSize/2)
|
||||
assertWarningNames(t, reconcileAsAtStartup(t, svc.cache), path)
|
||||
|
||||
getImage(t, svc, photoVariant(fixtures, 40, FitCover))
|
||||
getImage(t, svc, photoVariant(fixtures, 30, FitCover))
|
||||
|
||||
fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
|
||||
t.Logf("upstream fetches = %d", fetches)
|
||||
|
||||
if fetches != 2 {
|
||||
t.Errorf("upstream fetches = %d, want 2", fetches)
|
||||
}
|
||||
|
||||
size := fileSize(t, path)
|
||||
if size != wholeSize {
|
||||
t.Errorf("cached source is %d bytes, want %d", size, wholeSize)
|
||||
}
|
||||
}
|
||||
|
||||
// assertConvertedAgain caches a variant of the photo, cuts its file to size
|
||||
// bytes and runs the reconciliation pass, which must remove it with a
|
||||
// warning naming it. It then checks that Get converts the image again
|
||||
// instead of serving the file, and that the next Get serves the whole
|
||||
// variant from the cache.
|
||||
func assertConvertedAgain(t *testing.T, size int64) {
|
||||
t.Helper()
|
||||
|
||||
svc, fixtures := SetupTestService(t)
|
||||
req := photoVariant(fixtures, 85, FitCover)
|
||||
|
||||
want, _ := getImage(t, svc, req)
|
||||
|
||||
path := svc.cache.variants.keyToPath(CacheKey(req))
|
||||
|
||||
truncateFile(t, path, size)
|
||||
assertWarningNames(t, reconcileAsAtStartup(t, svc.cache), path)
|
||||
|
||||
got, status := getImage(t, svc, req)
|
||||
t.Logf("after the cut: %d bytes, %s", len(got), status)
|
||||
|
||||
if status != CacheMiss || !bytes.Equal(got, want) {
|
||||
t.Errorf("after the cut: Get() = %d bytes, %s; want the %d bytes "+
|
||||
"converted again, %s", len(got), status, len(want), CacheMiss)
|
||||
}
|
||||
|
||||
got, status = getImage(t, svc, req)
|
||||
t.Logf("next request: %d bytes, %s", len(got), status)
|
||||
|
||||
if status != CacheHit || !bytes.Equal(got, want) {
|
||||
t.Errorf("next request: Get() = %d bytes, %s; want the %d bytes, %s",
|
||||
len(got), status, len(want), CacheHit)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Get_ConvertsAgainForEmptyCachedVariant empties a cached
|
||||
// variant file, as a crash or a restored snapshot could, and checks that it
|
||||
// is converted again rather than served empty.
|
||||
func TestService_Get_ConvertsAgainForEmptyCachedVariant(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
assertConvertedAgain(t, 0)
|
||||
}
|
||||
|
||||
// TestService_Get_ConvertsAgainForPartialCachedVariant cuts a cached
|
||||
// variant file to ten bytes, as a crash or a restored snapshot could, and
|
||||
// checks that it is converted again rather than served cut short.
|
||||
func TestService_Get_ConvertsAgainForPartialCachedVariant(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
assertConvertedAgain(t, 10)
|
||||
}
|
||||
|
||||
// TestReconciliationRemovesEmptyVariantFileWithoutRow checks that the
|
||||
// reconciliation pass removes an empty variant file that has no
|
||||
// accounting row, with a warning naming it, instead of adding a row for
|
||||
// it that would have it served empty.
|
||||
func TestReconciliationRemovesEmptyVariantFileWithoutRow(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
|
||||
_, err := cache.variants.Store(testVariantKeyOne, bytes.NewReader(nil),
|
||||
testContentTypeWebP)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to store an empty variant file: %v", err)
|
||||
}
|
||||
|
||||
logs := reconcileAsAtStartup(t, cache)
|
||||
|
||||
assertWarningNames(t, logs, cache.variants.keyToPath(testVariantKeyOne))
|
||||
|
||||
if cache.variants.Exists(testVariantKeyOne) {
|
||||
t.Error("the empty variant file is still on disk")
|
||||
}
|
||||
|
||||
rows := countRows(t, cache,
|
||||
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`,
|
||||
string(testVariantKeyOne))
|
||||
if rows != 0 {
|
||||
t.Errorf("the empty variant file got %d accounting rows, want 0", rows)
|
||||
}
|
||||
}
|
||||
|
||||
// TestVariantStorage_StoreReplacesMetaFile stores a variant over one
|
||||
// already stored and checks that Store puts a new .meta file in place of
|
||||
// the old one, as it does the image, rather than writing into the old
|
||||
// file, which a crash could leave empty. The old .meta file is given a
|
||||
// second name before the store, so that it can be told apart from the new
|
||||
// one after it.
|
||||
func TestVariantStorage_StoreReplacesMetaFile(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
storage, key := storeTestVariant(t)
|
||||
metaPath := storage.keyToPath(key) + variantMetaSuffix
|
||||
oldMetaPath := metaPath + ".old"
|
||||
|
||||
err := os.Link(metaPath, oldMetaPath)
|
||||
if err != nil {
|
||||
t.Fatalf("giving the .meta file a second name: %v", err)
|
||||
}
|
||||
|
||||
_, err = storage.Store(key, bytes.NewReader([]byte("new variant data")),
|
||||
testContentTypeJPEG)
|
||||
if err != nil {
|
||||
t.Fatalf("Store() error = %v", err)
|
||||
}
|
||||
|
||||
oldMeta, err := os.Stat(oldMetaPath)
|
||||
if err != nil {
|
||||
t.Fatalf("stat of the old .meta file: %v", err)
|
||||
}
|
||||
|
||||
newMeta, err := os.Stat(metaPath)
|
||||
if err != nil {
|
||||
t.Fatalf("stat of the new .meta file: %v", err)
|
||||
}
|
||||
|
||||
if os.SameFile(oldMeta, newMeta) {
|
||||
t.Error("Store() wrote into the old .meta file instead of putting " +
|
||||
"a new one in its place")
|
||||
}
|
||||
|
||||
reader, _, contentType, err := storage.LoadWithMeta(key)
|
||||
if err != nil {
|
||||
t.Fatalf("LoadWithMeta() error = %v", err)
|
||||
}
|
||||
|
||||
_ = reader.Close()
|
||||
|
||||
t.Logf("content type after the store: %s", contentType)
|
||||
|
||||
if contentType != testContentTypeJPEG {
|
||||
t.Errorf("content type = %q, want %q", contentType, testContentTypeJPEG)
|
||||
}
|
||||
}
|
||||
|
||||
// TestContentStorage_StoreHashedReplacesFileOfAnotherSize checks that a
|
||||
// store over a file of another size at the content's path, such as a
|
||||
// partial file a crash left, replaces it with the content.
|
||||
func TestContentStorage_StoreHashedReplacesFileOfAnotherSize(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
storage, err := NewContentStorage(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatalf("NewContentStorage() error = %v", err)
|
||||
}
|
||||
|
||||
content := []byte("the whole source image")
|
||||
sum := sha256.Sum256(content)
|
||||
hash := ContentHash(hex.EncodeToString(sum[:]))
|
||||
path := storage.hashToPath(hash)
|
||||
|
||||
err = os.MkdirAll(filepath.Dir(path), StorageDirPerm)
|
||||
if err != nil {
|
||||
t.Fatalf("creating the content's directory: %v", err)
|
||||
}
|
||||
|
||||
err = os.WriteFile(path, content[:5], StorageFilePerm)
|
||||
if err != nil {
|
||||
t.Fatalf("writing a partial file: %v", err)
|
||||
}
|
||||
|
||||
_, err = storage.StoreHashed(hash, content)
|
||||
if err != nil {
|
||||
t.Fatalf("StoreHashed() error = %v", err)
|
||||
}
|
||||
|
||||
stored, err := storage.Load(hash)
|
||||
if err != nil {
|
||||
t.Fatalf("Load() error = %v", err)
|
||||
}
|
||||
|
||||
defer func() { _ = stored.Close() }()
|
||||
|
||||
got, err := io.ReadAll(stored)
|
||||
if err != nil {
|
||||
t.Fatalf("reading the stored content: %v", err)
|
||||
}
|
||||
|
||||
if !bytes.Equal(got, content) {
|
||||
t.Errorf("stored file = %q, want %q", got, content)
|
||||
}
|
||||
}
|
||||
@@ -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