fix(content/oci): reclaim the temporary files of interrupted writes

`ingest` names its return values so that the deferred cleanup can remove
the ingest file when a write fails, but its error paths return with
`return "", err`, which clears the named return `path` before the deferred
function runs. The cleanup therefore calls `os.Remove("")` and the ingest
file is left behind, so every failed push costs the disk space of the
content written before the failure. Set the named error and return, as
`ioutil.Ingest` has done since #1185.

Nothing reclaimed those files afterwards either. `GC` sweeps `blobs/` only,
so the residue of interrupted blob and metadata writes accumulates for the
life of the store, which makes the store a contributor to the full file
system that produced the residue in the first place. Remove them in `GC`,
where the other unreachable content of the store is already collected.

Only entries that have not been modified for an hour are reclaimed. A write
removes its own temporary file when it fails, so anything still present is
either the residue of a process that died mid-write or a write that another
`Store` on the same directory is performing right now, which the lock of
this `Store` does not serialize against. The modification time of a write in
progress keeps advancing, so a live write is never eligible while the
residue of a dead one always becomes eligible.

The two directories are otherwise treated differently on purpose. `ingest/`
is created and owned by this package, so any stale entry in it is residue.
The root of the store may hold files that belong to whoever put them there,
which the image layout specification permits, so only the names a temporary
file of this package can actually have are considered there.

Signed-off-by: Chris Wedgwood <cw@f00f.org>

Author: cwedgwood

content/oci/oci.go

@@ -23,11 +23,13 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"io/fs"
 	"maps"
 	"os"
 	"path"
 	"path/filepath"
 	"sync"
+	"time"
 
 	"github.com/opencontainers/go-digest"
 	specs "github.com/opencontainers/image-spec/specs-go"
@@ -530,6 +532,7 @@ func (s *Store) writeIndexFile() error {
 // The garbage to be cleaned are:
 //   - unreferenced (dangling) blobs in Store which have no predecessors
 //   - garbage blobs in the storage whose metadata is not stored in Store
+//   - temporary files left behind by interrupted blob and metadata writes
 func (s *Store) GC(ctx context.Context) error {
 	s.sync.Lock()
 	defer s.sync.Unlock()
@@ -580,6 +583,80 @@ func (s *Store) GC(ctx context.Context) error {
 			}
 		}
 	}
+
+	// clean up the temporary files left behind by interrupted writes
+	if err := s.gcLeftovers(); err != nil {
+		return fmt.Errorf("unable to remove leftover temporary files: %w", err)
+	}
+	return nil
+}
+
+// leftoverExpiry is how long a temporary file must have been untouched before
+// GC reclaims it, so that a write in progress -- possibly by another Store on
+// the same directory -- is never reclaimed from underneath.
+const leftoverExpiry = time.Hour
+
+// gcLeftovers removes the temporary files that interrupted writes leave
+// behind: the ingest files of blob writes that did not complete, and the
+// temporary files of metadata writes that were never renamed into place.
+// Nothing else refers to them, and no other code path removes them.
+func (s *Store) gcLeftovers() error {
+	// A write removes its own temporary file when it fails, so anything still
+	// here is the residue of a process that died mid-write -- or a write that
+	// another Store on the same directory is performing right now, which the
+	// lock of this Store does not serialize against. Only entries that have
+	// not been touched for leftoverExpiry are reclaimed, since the modification
+	// time of a write in progress keeps advancing.
+	stale := olderThan(leftoverExpiry)
+
+	if err := removeFiles(s.storage.ingestRoot, stale); err != nil {
+		return err
+	}
+	// the temporary files of metadata writes are created next to the file that
+	// they replace, in the root of the store.
+	return removeFiles(s.root, func(entry fs.DirEntry) bool {
+		name := entry.Name()
+		if !isTempFileOf(name, ocispec.ImageIndexFile) && !isTempFileOf(name, ocispec.ImageLayoutFile) {
+			return false
+		}
+		return stale(entry)
+	})
+}
+
+// olderThan returns a matcher accepting the directory entries that have not
+// been modified for at least d. An entry whose information cannot be read is
+// not matched: it has either just been removed by someone else, or it is not
+// ours to reason about.
+func olderThan(d time.Duration) func(entry fs.DirEntry) bool {
+	return func(entry fs.DirEntry) bool {
+		fi, err := entry.Info()
+		if err != nil {
+			return false
+		}
+		return time.Since(fi.ModTime()) >= d
+	}
+}
+
+// removeFiles removes the files in dir that are matched by match. Directories
+// and entries that are not matched are left alone, and a directory that does
+// not exist is not an error.
+func removeFiles(dir string, match func(entry fs.DirEntry) bool) error {
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		if errors.Is(err, fs.ErrNotExist) {
+			return nil
+		}
+		return err
+	}
+
+	for _, entry := range entries {
+		if entry.IsDir() || !match(entry) {
+			continue
+		}
+		if err := os.Remove(filepath.Join(dir, entry.Name())); err != nil && !errors.Is(err, fs.ErrNotExist) {
+			return err
+		}
+	}
 	return nil
 }
 

content/oci/oci_test.go

@@ -25,6 +25,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"io/fs"
 	"os"
 	"path"
 	"path/filepath"
@@ -33,6 +34,7 @@ import (
 	"strings"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/opencontainers/go-digest"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
@@ -3187,3 +3189,141 @@ func TestStore_TrailingBytesInMetadataFile(t *testing.T) {
 		t.Fatal("New() error =", err)
 	}
 }
+
+// TestStore_GC_Leftovers ensures that the temporary files of writes that were
+// interrupted are reclaimed by GC, since nothing else removes them, and that
+// the files GC does not own -- including the temporary files of a write that
+// may still be in progress -- are left alone.
+func TestStore_GC_Leftovers(t *testing.T) {
+	tempDir := t.TempDir()
+	s, err := New(tempDir)
+	if err != nil {
+		t.Fatal("New() error =", err)
+	}
+	ctx := context.Background()
+
+	ingestRoot := s.storage.ingestRoot
+	if err := os.MkdirAll(ingestRoot, 0777); err != nil {
+		t.Fatal("error calling MkdirAll(), error =", err)
+	}
+	write := func(path string) {
+		t.Helper()
+		if err := os.WriteFile(path, []byte("whatever"), 0666); err != nil {
+			t.Fatal("error calling WriteFile(), error =", err)
+		}
+	}
+	// residue of writes interrupted long enough ago that no process can still
+	// be working on them
+	stale := []string{
+		filepath.Join(ingestRoot, digest.FromString("whatever").Encoded()+"_1525580788"),
+		filepath.Join(ingestRoot, "not-an-ingest-file"),
+		filepath.Join(tempDir, ocispec.ImageIndexFile+"_"+rand.Text()),
+		filepath.Join(tempDir, ocispec.ImageLayoutFile+"_"+rand.Text()),
+	}
+	for _, path := range stale {
+		write(path)
+		old := time.Now().Add(-2 * leftoverExpiry)
+		if err := os.Chtimes(path, old, old); err != nil {
+			t.Fatal("error calling Chtimes(), error =", err)
+		}
+	}
+	// files that must survive: temporary files recent enough to belong to a
+	// write in progress, and files that only share the prefix of one
+	keep := []string{
+		filepath.Join(ingestRoot, digest.FromString("in progress").Encoded()+"_1525580789"),
+		filepath.Join(tempDir, ocispec.ImageIndexFile+"_"+rand.Text()),
+		filepath.Join(tempDir, ocispec.ImageIndexFile+"_backup"),
+		filepath.Join(tempDir, ocispec.ImageIndexFile+"_"+strings.ToLower(rand.Text())),
+	}
+	for _, path := range keep {
+		write(path)
+	}
+	keep = append(keep,
+		filepath.Join(tempDir, ocispec.ImageIndexFile),
+		filepath.Join(tempDir, ocispec.ImageLayoutFile))
+
+	if err := s.GC(ctx); err != nil {
+		t.Fatal("Store.GC() error =", err)
+	}
+
+	for _, path := range stale {
+		if _, err := os.Stat(path); !errors.Is(err, os.ErrNotExist) {
+			t.Errorf("%s still exists after GC(), error = %v", path, err)
+		}
+	}
+	for _, path := range keep {
+		if _, err := os.Stat(path); err != nil {
+			t.Errorf("error calling Stat() on %s, error = %v", path, err)
+		}
+	}
+}
+
+func Test_olderThan(t *testing.T) {
+	tempDir := t.TempDir()
+	path := filepath.Join(tempDir, "file")
+	if err := os.WriteFile(path, []byte("whatever"), 0666); err != nil {
+		t.Fatal("error calling WriteFile(), error =", err)
+	}
+	entry := func() fs.DirEntry {
+		t.Helper()
+		entries, err := os.ReadDir(tempDir)
+		if err != nil {
+			t.Fatal("error calling ReadDir(), error =", err)
+		}
+		return entries[0]
+	}
+
+	// a file just written belongs to a write that may still be in progress
+	if olderThan(time.Hour)(entry()) {
+		t.Error("olderThan() = true for a file just written, want false")
+	}
+	old := time.Now().Add(-2 * time.Hour)
+	if err := os.Chtimes(path, old, old); err != nil {
+		t.Fatal("error calling Chtimes(), error =", err)
+	}
+	if !olderThan(time.Hour)(entry()) {
+		t.Error("olderThan() = false for an untouched file, want true")
+	}
+}
+
+func Test_removeFiles(t *testing.T) {
+	tempDir := t.TempDir()
+	notADir := filepath.Join(tempDir, "file")
+	if err := os.WriteFile(notADir, []byte("whatever"), 0666); err != nil {
+		t.Fatal("error calling WriteFile(), error =", err)
+	}
+	subDir := filepath.Join(tempDir, "dir")
+	if err := os.Mkdir(subDir, 0777); err != nil {
+		t.Fatal("error calling Mkdir(), error =", err)
+	}
+
+	all := func(fs.DirEntry) bool { return true }
+	named := func(want string) func(fs.DirEntry) bool {
+		return func(entry fs.DirEntry) bool { return entry.Name() == want }
+	}
+
+	// a directory that does not exist holds no leftovers
+	if err := removeFiles(filepath.Join(tempDir, "missing"), all); err != nil {
+		t.Error("removeFiles() error =", err)
+	}
+	// a path that is not a directory cannot be listed
+	if err := removeFiles(notADir, all); err == nil {
+		t.Error("removeFiles() error = nil, wantErr = true")
+	}
+	// directories and unmatched files are left alone
+	if err := removeFiles(tempDir, named("dir")); err != nil {
+		t.Fatal("removeFiles() error =", err)
+	}
+	for _, path := range []string{notADir, subDir} {
+		if _, err := os.Stat(path); err != nil {
+			t.Errorf("error calling Stat() on %s, error = %v", path, err)
+		}
+	}
+	// matched files are removed
+	if err := removeFiles(tempDir, named("file")); err != nil {
+		t.Fatal("removeFiles() error =", err)
+	}
+	if _, err := os.Stat(notADir); !errors.Is(err, os.ErrNotExist) {
+		t.Errorf("Stat() error = %v, want %v", err, os.ErrNotExist)
+	}
+}

content/oci/storage.go

@@ -155,12 +155,14 @@ func (s *Storage) ingest(expected ocispec.Descriptor, content io.Reader) (path s
 	buf := bufPool.Get().(*[]byte)
 	defer bufPool.Put(buf)
 	if err := ioutil.CopyBuffer(fp, content, *buf, expected); err != nil {
-		return "", fmt.Errorf("failed to ingest: %w", err)
+		ingestErr = fmt.Errorf("failed to ingest: %w", err)
+		return
 	}
 
 	// change to readonly
 	if err := os.Chmod(path, 0444); err != nil {
-		return "", fmt.Errorf("failed to make readonly: %w", err)
+		ingestErr = fmt.Errorf("failed to make readonly: %w", err)
+		return
 	}
 
 	return

content/oci/storage_test.go

@@ -413,3 +413,34 @@ func TestStorage_Delete(t *testing.T) {
 		t.Fatalf("got error = %v, want %v", err, errdef.ErrNotFound)
 	}
 }
+
+// TestStorage_BadPush_NoIngestFileLeftBehind ensures that the ingest file of a
+// push that failed is removed. Nothing else removes it, so a leaked ingest file
+// occupies its space until the store is deleted.
+func TestStorage_BadPush_NoIngestFileLeftBehind(t *testing.T) {
+	content := []byte("hello world")
+	desc := ocispec.Descriptor{
+		MediaType: "test",
+		Digest:    digest.FromBytes(content),
+		Size:      int64(len(content)),
+	}
+
+	tempDir := t.TempDir()
+	s, err := NewStorage(tempDir)
+	if err != nil {
+		t.Fatal("New() error =", err)
+	}
+	ctx := context.Background()
+
+	if err := s.Push(ctx, desc, strings.NewReader("foobar")); err == nil {
+		t.Fatal("Storage.Push() error = nil, wantErr = true")
+	}
+
+	entries, err := os.ReadDir(s.ingestRoot)
+	if err != nil {
+		t.Fatal("error calling ReadDir(), error =", err)
+	}
+	if len(entries) != 0 {
+		t.Errorf("len(ingest entries) = %v, want %v", len(entries), 0)
+	}
+}