media: make retention and the disk budget true
Put wrote the blob and then the sidecar. A full disk or a crash between the two left bytes on disk with no sidecar, and List walks sidecars, so Prune could never see them: Put returned an error and an image nobody knew about became permanent. The sidecar goes first, a failed write is rolled back, and Prune also collects blob files that have no readable sidecar and are past retention, which picks up whatever an older build leaked. The per-blob cap bounds one call and nothing bounded their sum. Content addressing does not help, because one flipped pixel is a different digest, so 64 MiB per call and an unlimited number of calls fills the disk mavend's database lives on. The store now carries a whole-store budget, seeded from disk at open so a restart does not begin at zero. Found in review of #72.
This commit is contained in:
kami
+167
-6
@@ -11,6 +11,7 @@ import (
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -24,6 +25,19 @@ const DefaultMaxBytes int64 = 64 << 20
|
||||
// short enough that "she has a month of my meetings on disk" is never true.
|
||||
const DefaultRetention = 7 * 24 * time.Hour
|
||||
|
||||
// DefaultMaxTotalBytes — the whole-store budget when one is not configured. The
|
||||
// per-blob cap bounds one call and nothing bounded the sum of them: 64 MiB per
|
||||
// call, an unlimited number of calls, and a seven-day window fills the disk
|
||||
// mavend's database lives on. Content addressing does not help, because one
|
||||
// flipped pixel is a different digest. 4 GiB is roughly sixty meetings or a few
|
||||
// thousand photos inside the window.
|
||||
const DefaultMaxTotalBytes int64 = 4 << 30
|
||||
|
||||
// ErrStoreFull — the store is at its total-bytes budget. Distinct from
|
||||
// ErrTooLarge: the payload is a reasonable size and there is no room for it, so
|
||||
// the answer is to prune or raise the budget, not to send something smaller.
|
||||
var ErrStoreFull = errors.New("media: store is full")
|
||||
|
||||
// Store — a content-addressed blob directory. Zero value is not usable; build
|
||||
// one with Open, which creates the directory 0700. The store holds no lock and
|
||||
// no cache: every operation is a filesystem call, and two writers of the same
|
||||
@@ -31,8 +45,16 @@ const DefaultRetention = 7 * 24 * time.Hour
|
||||
type Store struct {
|
||||
dir string
|
||||
maxBytes int64
|
||||
maxTotal int64
|
||||
retention time.Duration
|
||||
now func() time.Time
|
||||
|
||||
// total is the running sum of stored blob bytes, seeded by Open with a
|
||||
// directory walk and kept up to date by Put, Delete and Prune. It is a
|
||||
// cache of something the filesystem already knows: re-walking on every Put
|
||||
// would be correct too and would make an image intake O(store size).
|
||||
totalMu sync.Mutex
|
||||
total int64
|
||||
}
|
||||
|
||||
// Open prepares a blob store rooted at dir. maxBytes ≤ 0 ⇒ DefaultMaxBytes;
|
||||
@@ -40,6 +62,12 @@ type Store struct {
|
||||
// created later) is 0700: these are recordings of people, and the daemon's user
|
||||
// is the only reader.
|
||||
func Open(dir string, maxBytes int64, retention time.Duration) (*Store, error) {
|
||||
return OpenWithBudget(dir, maxBytes, 0, retention)
|
||||
}
|
||||
|
||||
// OpenWithBudget is Open with the whole-store budget spelled out. maxTotal ≤ 0
|
||||
// ⇒ DefaultMaxTotalBytes.
|
||||
func OpenWithBudget(dir string, maxBytes, maxTotal int64, retention time.Duration) (*Store, error) {
|
||||
if strings.TrimSpace(dir) == "" {
|
||||
return nil, errors.New("media: empty dir")
|
||||
}
|
||||
@@ -53,12 +81,47 @@ func Open(dir string, maxBytes int64, retention time.Duration) (*Store, error) {
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = DefaultMaxBytes
|
||||
}
|
||||
if maxTotal <= 0 {
|
||||
maxTotal = DefaultMaxTotalBytes
|
||||
}
|
||||
if maxTotal < maxBytes {
|
||||
return nil, fmt.Errorf("media: max_total_bytes %d is below the per-blob cap %d", maxTotal, maxBytes)
|
||||
}
|
||||
if retention <= 0 {
|
||||
retention = DefaultRetention
|
||||
}
|
||||
return &Store{dir: abs, maxBytes: maxBytes, retention: retention, now: time.Now}, nil
|
||||
s := &Store{dir: abs, maxBytes: maxBytes, maxTotal: maxTotal, retention: retention, now: time.Now}
|
||||
s.total = s.measure()
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// measure sums what is already on disk, so a restart does not start the budget
|
||||
// over at zero.
|
||||
func (s *Store) measure() int64 {
|
||||
var total int64
|
||||
_ = filepath.WalkDir(s.dir, func(path string, d fs.DirEntry, err error) error {
|
||||
if err != nil || d.IsDir() || strings.HasSuffix(path, ".json") {
|
||||
return nil //nolint:nilerr // an unreadable corner is not worth refusing to boot over
|
||||
}
|
||||
if info, err := d.Info(); err == nil {
|
||||
total += info.Size()
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return total
|
||||
}
|
||||
|
||||
// Total is the number of blob bytes currently stored, and Budget the cap Put
|
||||
// checks it against. Both are exported so the daemon can log how close it is.
|
||||
func (s *Store) Total() int64 {
|
||||
s.totalMu.Lock()
|
||||
defer s.totalMu.Unlock()
|
||||
return s.total
|
||||
}
|
||||
|
||||
// Budget is the whole-store cap.
|
||||
func (s *Store) Budget() int64 { return s.maxTotal }
|
||||
|
||||
// Dir is the store root. Exported for logs and for pointing a subprocess at a
|
||||
// path under it.
|
||||
func (s *Store) Dir() string { return s.dir }
|
||||
@@ -104,10 +167,36 @@ func (s *Store) Put(kind Kind, mime, source string, data []byte) (Blob, error) {
|
||||
b.Created = prev.Created
|
||||
}
|
||||
|
||||
if err := writeFile(blobPath, data); err != nil {
|
||||
// A blob already on disk costs nothing more, so dedupe is checked before
|
||||
// the budget rather than after it.
|
||||
_, already := os.Stat(blobPath)
|
||||
if already != nil {
|
||||
s.totalMu.Lock()
|
||||
room := s.total+b.Size <= s.maxTotal
|
||||
if room {
|
||||
s.total += b.Size
|
||||
}
|
||||
s.totalMu.Unlock()
|
||||
if !room {
|
||||
return Blob{}, fmt.Errorf("%w: %d stored, %d budget, %d more asked for",
|
||||
ErrStoreFull, s.Total(), s.maxTotal, b.Size)
|
||||
}
|
||||
}
|
||||
|
||||
// The sidecar goes first. Written second, a full disk or a crash between
|
||||
// the two left the bytes on disk with no sidecar, and List only sees
|
||||
// sidecars, so Prune could never collect them: Put returned an error and an
|
||||
// image nobody knew about became permanent.
|
||||
if err := writeMeta(metaPath, b); err != nil {
|
||||
return Blob{}, err
|
||||
}
|
||||
if err := writeMeta(metaPath, b); err != nil {
|
||||
if err := writeFile(blobPath, data); err != nil {
|
||||
_ = os.Remove(metaPath)
|
||||
if already != nil {
|
||||
s.totalMu.Lock()
|
||||
s.total -= b.Size
|
||||
s.totalMu.Unlock()
|
||||
}
|
||||
return Blob{}, err
|
||||
}
|
||||
return b, nil
|
||||
@@ -210,10 +299,24 @@ func (s *Store) Delete(id string) error {
|
||||
continue
|
||||
}
|
||||
for _, e := range entries {
|
||||
if strings.HasPrefix(e.Name(), id) {
|
||||
if err := os.Remove(filepath.Join(bucket, e.Name())); err != nil && !errors.Is(err, fs.ErrNotExist) {
|
||||
return fmt.Errorf("media: delete %s: %w", shortID(id), err)
|
||||
if !strings.HasPrefix(e.Name(), id) {
|
||||
continue
|
||||
}
|
||||
path := filepath.Join(bucket, e.Name())
|
||||
var size int64
|
||||
if info, err := e.Info(); err == nil && !strings.HasSuffix(e.Name(), ".json") {
|
||||
size = info.Size()
|
||||
}
|
||||
if err := os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) {
|
||||
return fmt.Errorf("media: delete %s: %w", shortID(id), err)
|
||||
}
|
||||
if size > 0 {
|
||||
s.totalMu.Lock()
|
||||
s.total -= size
|
||||
if s.total < 0 {
|
||||
s.total = 0
|
||||
}
|
||||
s.totalMu.Unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -232,15 +335,70 @@ func (s *Store) Prune() (int, error) {
|
||||
}
|
||||
now := s.now()
|
||||
deleted := 0
|
||||
known := map[string]bool{}
|
||||
for _, b := range blobs {
|
||||
known[b.ID] = true
|
||||
if b.Age(now) <= s.retention {
|
||||
continue
|
||||
}
|
||||
if err := s.Delete(b.ID); err != nil {
|
||||
return deleted, err
|
||||
}
|
||||
delete(known, b.ID)
|
||||
deleted++
|
||||
}
|
||||
n, err := s.pruneOrphans(known, now)
|
||||
return deleted + n, err
|
||||
}
|
||||
|
||||
// pruneOrphans collects blob files with no readable sidecar. List walks
|
||||
// sidecars, so those files were invisible to retention and stayed on disk
|
||||
// forever: audio of people accumulating is the exact failure this package
|
||||
// exists to prevent, and a half-finished Put from an older build is enough to
|
||||
// produce one. They are only collected once they are older than retention, so a
|
||||
// Put racing a Prune does not lose its bytes.
|
||||
func (s *Store) pruneOrphans(known map[string]bool, now time.Time) (int, error) {
|
||||
deleted := 0
|
||||
for _, kind := range []Kind{KindImage, KindAudio} {
|
||||
root := filepath.Join(s.dir, string(kind))
|
||||
err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
if errors.Is(err, fs.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
if d.IsDir() || strings.HasSuffix(path, ".json") {
|
||||
return nil
|
||||
}
|
||||
name := d.Name()
|
||||
id, _, _ := strings.Cut(name, ".")
|
||||
if known[id] {
|
||||
return nil
|
||||
}
|
||||
info, err := d.Info()
|
||||
if err != nil {
|
||||
return nil //nolint:nilerr // gone underneath us is the outcome we wanted
|
||||
}
|
||||
if now.Sub(info.ModTime()) <= s.retention {
|
||||
return nil
|
||||
}
|
||||
if err := os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) {
|
||||
return err
|
||||
}
|
||||
s.totalMu.Lock()
|
||||
s.total -= info.Size()
|
||||
if s.total < 0 {
|
||||
s.total = 0
|
||||
}
|
||||
s.totalMu.Unlock()
|
||||
deleted++
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return deleted, fmt.Errorf("media: prune %s: %w", kind, err)
|
||||
}
|
||||
}
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
@@ -300,6 +458,9 @@ func extFor(mime string, kind Kind) string {
|
||||
case "image/gif":
|
||||
return ".gif"
|
||||
case "image/webp":
|
||||
// Unreachable for images today: SniffImage refuses webp before
|
||||
// anything reaches Put, because this build has no webp decoder. Kept
|
||||
// so the mapping is right on the day one arrives.
|
||||
return ".webp"
|
||||
case "audio/wav", "audio/x-wav", "audio/wave":
|
||||
return ".wav"
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
package media
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
@@ -212,3 +214,129 @@ func TestOpenRejectsEmptyDir(t *testing.T) {
|
||||
t.Error("empty dir accepted")
|
||||
}
|
||||
}
|
||||
|
||||
// A blob whose sidecar is missing was invisible to List, so Prune never saw it
|
||||
// and the bytes stayed on disk forever. Put produced exactly that state, by
|
||||
// writing the blob first and the sidecar second.
|
||||
func TestPruneCollectsASidecarlessBlob(t *testing.T) {
|
||||
s := testStore(t)
|
||||
b, err := s.Put(KindImage, "image/png", "web:upload", []byte("orphan"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
meta := filepath.Join(s.dir, string(KindImage), b.ID[:2], b.ID+".json")
|
||||
if err := os.Remove(meta); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Age the file past retention, the same way a real orphan gets there.
|
||||
old := time.Now().Add(-2 * DefaultRetention)
|
||||
if err := os.Chtimes(b.Path, old, old); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
n, err := s.Prune()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("pruned %d, want the orphan collected", n)
|
||||
}
|
||||
if _, err := os.Stat(b.Path); !errors.Is(err, os.ErrNotExist) {
|
||||
t.Fatalf("the orphaned bytes are still on disk: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// A young orphan is left alone: a Put racing a Prune must not lose its bytes.
|
||||
func TestPruneLeavesAYoungOrphan(t *testing.T) {
|
||||
s := testStore(t)
|
||||
b, err := s.Put(KindImage, "image/png", "web:upload", []byte("fresh"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Remove(filepath.Join(s.dir, string(KindImage), b.ID[:2], b.ID+".json")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n, err := s.Prune(); err != nil || n != 0 {
|
||||
t.Fatalf("prune = %d, %v; want the fresh orphan kept", n, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Put writes the sidecar first, so a failure writing the bytes leaves nothing
|
||||
// at all rather than an uncollectable blob.
|
||||
func TestPutLeavesNothingWhenTheBytesCannotBeWritten(t *testing.T) {
|
||||
s := testStore(t)
|
||||
data := []byte("will not land")
|
||||
sum := sha256.Sum256(data)
|
||||
id := hex.EncodeToString(sum[:])
|
||||
bucket := filepath.Join(s.dir, string(KindImage), id[:2])
|
||||
if err := os.MkdirAll(bucket, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// A directory where the blob file needs to be: rename onto it fails, which
|
||||
// is the same shape as a full disk one step later.
|
||||
if err := os.Mkdir(filepath.Join(bucket, id+".png"), 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := s.Put(KindImage, "image/png", "web:upload", data); err == nil {
|
||||
t.Fatal("put must fail")
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(bucket, id+".json")); !errors.Is(err, os.ErrNotExist) {
|
||||
t.Fatalf("a sidecar was left behind claiming a blob that does not exist: %v", err)
|
||||
}
|
||||
if s.Total() != 0 {
|
||||
t.Errorf("total = %d, want the failed put not counted", s.Total())
|
||||
}
|
||||
}
|
||||
|
||||
// The per-blob cap bounds one call and nothing bounded their sum. 64 MiB per
|
||||
// call times unlimited calls inside a seven-day window fills the disk mavend's
|
||||
// database lives on.
|
||||
func TestPutRefusesPastTheStoreBudget(t *testing.T) {
|
||||
s, err := OpenWithBudget(t.TempDir(), 16, 48, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i, want := range []bool{true, true, true, false} {
|
||||
data := []byte(strings.Repeat(string(rune('a'+i)), 16))
|
||||
_, err := s.Put(KindImage, "image/png", "web:upload", data)
|
||||
if ok := err == nil; ok != want {
|
||||
t.Fatalf("put %d: err = %v, want ok=%v", i, err, want)
|
||||
}
|
||||
if !want && !errors.Is(err, ErrStoreFull) {
|
||||
t.Fatalf("put %d: err = %v, want ErrStoreFull", i, err)
|
||||
}
|
||||
}
|
||||
// The same bytes again cost nothing, so they are not refused.
|
||||
if _, err := s.Put(KindImage, "image/png", "web:upload", []byte(strings.Repeat("a", 16))); err != nil {
|
||||
t.Fatalf("a re-send of stored bytes was refused: %v", err)
|
||||
}
|
||||
// Deleting frees the budget again.
|
||||
blobs, err := s.List(KindImage)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := s.Delete(blobs[0].ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := s.Put(KindImage, "image/png", "web:upload", []byte(strings.Repeat("z", 16))); err != nil {
|
||||
t.Fatalf("budget was not released on delete: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// A restart must not start the budget over at zero.
|
||||
func TestOpenSeedsTheBudgetFromDisk(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
s, err := OpenWithBudget(dir, 16, 48, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := s.Put(KindImage, "image/png", "web:upload", []byte(strings.Repeat("a", 16))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
again, err := OpenWithBudget(dir, 16, 48, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if again.Total() != 16 {
|
||||
t.Fatalf("total after reopen = %d, want 16", again.Total())
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user