Go on past cache entries the reconciliation pass cannot read (closes #343) #344

Merged
clawbot merged 2 commits from issue-343-reconcile-walk-goes-on into next 2026-10-27 02:24:15 +01:00
4 changed files with 257 additions and 33 deletions
+5
View File
@@ -256,6 +256,11 @@ Each look also drops the rows of variants whose files are missing. When pixa
cannot read a variant file's size for another reason, it keeps the variant's
row. It logs a warning naming the variant key and the error.
During a look, pixa skips a directory it cannot list and a file whose size it
cannot read. It logs a warning naming the path and the error, and goes on with
the rest of the look. A file renamed or removed after pixa listed its directory
gets no warning.
pixa's target is 1-5k r/s, which has not been measured at that rate (see Load
Test). Toward it, a cache hit opens the variant's file and nothing else, and
makes no database call. The content type comes from the file's first bytes,
+11
View File
@@ -30,6 +30,17 @@ P2: security: per-origin rate limiting
# Completed Steps
- 2026-10-27 the reconciliation pass skips a directory it cannot list and a file
whose info it cannot read, and goes on (closes #343). It stopped before its
later checks at a directory it could not list, or at a source or variant file
whose info it could not read.
- It logs the warning "skipped unreadable entry in cache directory", with
the path and the error.
- A file renamed or removed after its directory was listed gets no warning.
- `reconcileFiles` reads each file's info once, before the temp-file check,
and hands it to `sweepStaleTempFile`, `reconcileVariantFile` or
`reconcileSourceFile`.
- A cancelled context still stops the walk with its error.
- 2026-10-26 the reconciliation pass drops a variant's accounting row only when
its file is missing (closes #340). Any other error reading the file's size
keeps the row. It dropped the row on any error and logged the file as missing.
+40 -33
View File
@@ -515,12 +515,15 @@ func (c *Cache) reconcileAccounting(ctx context.Context) error {
}
// reconcileFiles walks baseDir, one of the cache's storage directories,
// sweeping stale temp files and passing each other file, with its name, to
// reconcileFile. Once ctx is cancelled it stops at the next file and
// returns ctx's error.
// sweeping stale temp files and passing each other file, with its info and
// name, to reconcileFile. It skips a directory it cannot list and a file
// whose info it cannot read, so one such entry does not stop the pass. It
// logs a warning naming the path and the error. A file renamed or removed
// after its directory was listed is skipped with no warning. Once ctx is
// cancelled it stops at the next file and returns ctx's error.
func (c *Cache) reconcileFiles(
ctx context.Context, baseDir string,
reconcileFile func(path string, entry fs.DirEntry, name string) error,
reconcileFile func(path string, info fs.FileInfo, name string) error,
) error {
return filepath.WalkDir(
baseDir,
@@ -529,19 +532,38 @@ func (c *Cache) reconcileFiles(
return ctx.Err()
}
if err != nil || entry.IsDir() {
return err
if err != nil {
c.log.Warn("skipped unreadable entry in cache directory",
"path", path, "error", err)
return nil
}
if entry.IsDir() {
return nil
}
info, err := entry.Info()
if errors.Is(err, fs.ErrNotExist) {
return nil
}
if err != nil {
c.log.Warn("skipped unreadable entry in cache directory",
"path", path, "error", err)
return nil
}
name := entry.Name()
if strings.HasPrefix(name, tempFilePrefix) {
c.sweepStaleTempFile(path, entry)
c.sweepStaleTempFile(path, info)
return nil
}
return reconcileFile(path, entry, name)
return reconcileFile(path, info, name)
},
)
}
@@ -551,8 +573,8 @@ func (c *Cache) reconcileFiles(
// and sweeping stale temp files.
func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
return c.reconcileFiles(ctx, c.variants.baseDir,
func(path string, entry fs.DirEntry, name string) error {
return c.reconcileVariantFile(ctx, path, entry, imagerequest.VariantKey(name))
func(path string, info fs.FileInfo, name string) error {
return c.reconcileVariantFile(ctx, path, info, imagerequest.VariantKey(name))
})
}
@@ -562,16 +584,11 @@ func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
// 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 imagerequest.VariantKey,
ctx context.Context, path string, info fs.FileInfo, cacheKey imagerequest.VariantKey,
) error {
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,
err := c.db.QueryRowContext(ctx,
`SELECT size_bytes FROM variant_content WHERE cache_key = ?`, string(cacheKey),
).Scan(&recordedSize)
@@ -714,8 +731,8 @@ func queryStringColumn[T ~string](
// and sweeping stale temp files.
func (c *Cache) reconcileSourceFiles(ctx context.Context) error {
return c.reconcileFiles(ctx, c.srcContent.baseDir,
func(path string, entry fs.DirEntry, name string) error {
return c.reconcileSourceFile(ctx, path, entry, imagerequest.SourceKey(name))
func(path string, info fs.FileInfo, name string) error {
return c.reconcileSourceFile(ctx, path, info, imagerequest.SourceKey(name))
})
}
@@ -726,16 +743,11 @@ func (c *Cache) reconcileSourceFiles(ctx context.Context) error {
// is removed with its row and a warning naming it, so the source is
// fetched again.
func (c *Cache) reconcileSourceFile(
ctx context.Context, path string, entry fs.DirEntry, key imagerequest.SourceKey,
ctx context.Context, path string, info fs.FileInfo, key imagerequest.SourceKey,
) 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,
err := c.db.QueryRowContext(ctx,
`SELECT size_bytes FROM source_content WHERE source_key = ?`,
string(key),
).Scan(&recordedSize)
@@ -900,17 +912,12 @@ func (c *Cache) correctUsageTotal(
// sweepStaleTempFile removes a temp file left behind by a crashed
// write once it is old enough that no in-flight store can own it.
func (c *Cache) sweepStaleTempFile(path string, entry fs.DirEntry) {
info, err := entry.Info()
if err != nil {
return
}
func (c *Cache) sweepStaleTempFile(path string, info fs.FileInfo) {
if time.Since(info.ModTime()) < staleTempFileAge {
return
}
err = os.Remove(path)
err := os.Remove(path)
if err != nil && !os.IsNotExist(err) {
c.log.Warn("failed to remove stale temp file", "path", path, "error", err)
+201
View File
@@ -9,8 +9,10 @@ import (
"log/slog"
"os"
"path/filepath"
"slices"
"strings"
"sync"
"syscall"
"testing"
"time"
@@ -785,6 +787,205 @@ func TestReconcileVariantRowsDropsOnlyRowsOfMissingFiles(t *testing.T) {
}
}
// openDirectoryNearPathLimit makes a chain of directories inside dir, each
// named with 255 zeros, the longest name Linux allows. Linux refuses a path
// of 4096 bytes or more, and the chain goes as deep as it can below that.
// It returns the chain's deepest directory as an os.Root, closed when the
// test ends. An entry made through it with a 255-byte name has a path even
// the root user cannot read. The os.Root can make that entry because it
// names the entry only by its name, not by the whole path.
func openDirectoryNearPathLimit(t *testing.T, dir string) *os.Root {
t.Helper()
name := strings.Repeat("0", 255)
deepest := dir
for len(filepath.Join(deepest, name)) < 4096 {
deepest = filepath.Join(deepest, name)
}
err := os.MkdirAll(deepest, StorageDirPerm)
if err != nil {
t.Fatalf("failed to make a chain of directories: %v", err)
}
root, err := os.OpenRoot(deepest)
if err != nil {
t.Fatalf("os.OpenRoot(%s) error = %v", deepest, err)
}
t.Cleanup(func() { _ = root.Close() })
return root
}
// makeEntriesTooLongToRead makes a new directory and a new file in the
// deepest directory of the chain that openDirectoryNearPathLimit makes
// inside dir. Even the root user can then neither list the new directory
// nor read the new file's info. It returns the paths of the new directory
// and the new file.
func makeEntriesTooLongToRead(t *testing.T, dir string) []string {
t.Helper()
deepest := openDirectoryNearPathLimit(t, dir)
dirName := strings.Repeat("0", 255)
err := deepest.Mkdir(dirName, StorageDirPerm)
if err != nil {
t.Fatalf("failed to make a directory too deep to list: %v", err)
}
fileName := strings.Repeat("1", 255)
err = deepest.WriteFile(fileName, []byte("unreadable"), StorageFilePerm)
if err != nil {
t.Fatalf("failed to make a file too deep to read: %v", err)
}
return []string{
filepath.Join(deepest.Name(), dirName),
filepath.Join(deepest.Name(), fileName),
}
}
// makeTempFileTooLongToRead makes a temp file in the deepest directory of
// the chain that openDirectoryNearPathLimit makes inside dir. Even the
// root user then cannot read the temp file's info. It returns the temp
// file's path.
func makeTempFileTooLongToRead(t *testing.T, dir string) string {
t.Helper()
deepest := openDirectoryNearPathLimit(t, dir)
name := tempFilePrefix + strings.Repeat("1", 255-len(tempFilePrefix))
err := deepest.WriteFile(name, []byte("partial"), StorageFilePerm)
if err != nil {
t.Fatalf("failed to make a temp file too deep to read: %v", err)
}
return filepath.Join(deepest.Name(), name)
}
// assertSkippedEntryWarning checks that logs hold the warning "skipped
// unreadable entry in cache directory" naming path, with the error
// ENAMETOOLONG for that path.
func assertSkippedEntryWarning(t *testing.T, logs, path string) {
t.Helper()
for line := range strings.Lines(logs) {
if strings.Contains(line, `"level":"WARN"`) &&
strings.Contains(line, `"msg":"skipped unreadable entry in cache directory"`) &&
strings.Contains(line, `"path":"`+path+`"`) &&
strings.Contains(line, path+": "+syscall.ENAMETOOLONG.Error()) {
return
}
}
t.Errorf("no warning %q names %s with the error %q",
"skipped unreadable entry in cache directory", path, syscall.ENAMETOOLONG)
}
// TestReconciliationGoesOnPastEntriesItCannotRead puts, in each storage
// directory, a directory the pass cannot list and a file whose info it
// cannot read. The pass walks each directory in name order, so it reaches
// them before the cache's files, whose keys start with a letter. It must
// log a warning naming each one's path and the error, and still do the
// rest of its work:
// - adopt a variant file that has no row
// - drop the row of a variant whose file is missing
// - remove a source file that has no row
// - correct a wrong total cache usage
func TestReconciliationGoesOnPastEntriesItCannotRead(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
ctx := t.Context()
untrackedVariant := []byte("variant with no row")
err := cache.variants.Write(string(testVariantKeyOne), untrackedVariant)
if err != nil {
t.Fatalf("failed to write a variant file with no row: %v", err)
}
storeEvictionTestVariant(t, cache, testVariantKeyTwo, []byte("variant two"))
err = os.Remove(cache.variants.keyToPath(string(testVariantKeyTwo)))
if err != nil {
t.Fatalf("failed to remove the second variant's file: %v", err)
}
untrackedSourceKey := "ffeeddcc0001"
err = cache.srcContent.Write(untrackedSourceKey, []byte("source with no row"))
if err != nil {
t.Fatalf("failed to write a source file with no row: %v", err)
}
_, err = cache.db.ExecContext(ctx,
`UPDATE cache_usage SET total_size_bytes = 1 WHERE id = 1`)
if err != nil {
t.Fatalf("failed to set a wrong total: %v", err)
}
unreadable := slices.Concat(
makeEntriesTooLongToRead(t, cache.variants.baseDir),
makeEntriesTooLongToRead(t, cache.srcContent.baseDir),
)
logs := reconcileAsAtStartup(t, cache)
for _, path := range unreadable {
assertSkippedEntryWarning(t, logs, path)
}
rowQuery := `SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`
if n := countRows(t, cache, rowQuery, string(testVariantKeyOne)); n != 1 {
t.Errorf("the variant file with no row got %d rows, want 1", n)
}
if n := countRows(t, cache, rowQuery, string(testVariantKeyTwo)); n != 0 {
t.Errorf("the variant with no file kept %d rows, want 0", n)
}
_, err = os.Stat(cache.srcContent.keyToPath(untrackedSourceKey))
if err == nil {
t.Error("the source file with no row is still on disk")
}
usage, err := cache.UsageBytes(ctx)
if err != nil {
t.Fatalf("UsageBytes() error = %v", err)
}
if usage != int64(len(untrackedVariant)) {
t.Errorf("UsageBytes() = %d, want %d, the size of the adopted variant",
usage, len(untrackedVariant))
}
}
// TestReconciliationWarnsOfTempFilesWhoseInfoItCannotRead puts, in each
// storage directory, a temp file whose info the pass cannot read. The
// pass must log a warning naming each one's path and the error, as it
// does for any other file whose info it cannot read.
func TestReconciliationWarnsOfTempFilesWhoseInfoItCannotRead(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
unreadable := []string{
makeTempFileTooLongToRead(t, cache.variants.baseDir),
makeTempFileTooLongToRead(t, cache.srcContent.baseDir),
}
logs := reconcileAsAtStartup(t, cache)
for _, path := range unreadable {
assertSkippedEntryWarning(t, logs, path)
}
}
// TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup proves
// reconciliation is not a one-shot startup-only pass: it must also run
// on the periodic ticker, so a variant file that lands on disk with no