diff --git a/internal/media/store.go b/internal/media/store.go index 32d3fc4..b931e9f 100644 --- a/internal/media/store.go +++ b/internal/media/store.go @@ -42,6 +42,29 @@ const DefaultRetention = 7 * 24 * time.Hour // thousand photos inside the window. const DefaultMaxTotalBytes int64 = 4 << 30 +const ( + // dirPerm and filePerm: these are recordings of people, so the daemon's + // user is the only reader. + dirPerm fs.FileMode = 0o700 + filePerm fs.FileMode = 0o600 + // metaExt is the sidecar suffix. Anything else under a bucket is blob + // bytes, which is how the walkers tell the two apart. + metaExt = ".json" + // spoolName is the incremental-write directory, held outside the kind + // directories so no walker mistakes a half-written file for a blob. + spoolName = "spool" + // bucketPrefix is how many leading id characters name the subdirectory, so + // one kind is spread over 256 directories rather than one flat listing. + bucketPrefix = 2 + // idLen is the length of a hex sha256, which is the only id shape a path + // is ever built from. + idLen = 64 +) + +// allKinds is every kind a walker has to visit. A store-wide operation covers +// all of them, and this is the one list to extend when a third kind lands. +var allKinds = []Kind{KindImage, KindAudio} + // 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. @@ -85,7 +108,7 @@ func OpenWithBudget(dir string, maxBytes, maxTotal int64, retention time.Duratio if err != nil { return nil, fmt.Errorf("media: resolve dir: %w", err) } - if err := os.MkdirAll(abs, 0o700); err != nil { + if err := os.MkdirAll(abs, dirPerm); err != nil { return nil, fmt.Errorf("media: create dir: %w", err) } if maxBytes <= 0 { @@ -117,14 +140,14 @@ func OpenWithBudget(dir string, maxBytes, maxTotal int64, retention time.Duratio // over at zero. func (s *Store) measure() int64 { var total int64 - spool := filepath.Join(s.dir, "spool") + spool := filepath.Join(s.dir, spoolName) _ = filepath.WalkDir(s.dir, func(path string, d fs.DirEntry, err error) error { if err == nil && d.IsDir() && path == spool { // Spool files are not blobs yet and PutFile counts them when they // become one. Counting them here too would double them. return filepath.SkipDir } - if err != nil || d.IsDir() || strings.HasSuffix(path, ".json") { + if err != nil || d.IsDir() || strings.HasSuffix(path, metaExt) { return nil //nolint:nilerr // an unreadable corner is not worth refusing to boot over } if info, err := d.Info(); err == nil { @@ -153,6 +176,54 @@ func (s *Store) Dir() string { return s.dir } // Retention is the configured age limit Prune enforces. func (s *Store) Retention() time.Duration { return s.retention } +// reserve claims size against the whole-store budget and reports ErrStoreFull +// when there is no room. Claiming before the write means two concurrent Puts +// cannot both pass a check that only one of them fits through. +func (s *Store) reserve(size int64) error { + s.totalMu.Lock() + room := s.total+size <= s.maxTotal + if room { + s.total += size + } + stored := s.total + s.totalMu.Unlock() + if !room { + return fmt.Errorf("%w: %d stored, %d budget, %d more asked for", + ErrStoreFull, stored, s.maxTotal, size) + } + return nil +} + +// release gives size back, for a reservation whose write failed and for bytes +// a delete removed. The floor at zero keeps a miscount from reading as a store +// that owes itself space. +func (s *Store) release(size int64) { + s.totalMu.Lock() + s.total -= size + if s.total < 0 { + s.total = 0 + } + s.totalMu.Unlock() +} + +// bucket is the directory an id lives in. Every path the store builds goes +// through here, so the traversal guard in validID has one place to sit. +func (s *Store) bucket(kind Kind, id string) string { + return filepath.Join(s.dir, string(kind), id[:bucketPrefix]) +} + +// newBlob describes what is about to be stored. A blob already on disk keeps +// its first-seen time: re-sending the same photo every hour must not keep it +// alive past retention. +func (s *Store) newBlob(kind Kind, mime, source, id, blobPath, metaPath string, size int64) Blob { + b := Blob{ID: id, Kind: kind, MIME: mime, Size: size, Source: source, + Created: s.now().UTC(), Path: blobPath} + if prev, err := readMeta(metaPath); err == nil && !prev.Created.IsZero() { + b.Created = prev.Created + } + return b +} + // Put stores data and returns its Blob. The id is the sha256 of data, so // storing the same bytes twice is idempotent: the second call rewrites the // sidecar (keeping the ORIGINAL creation time, so a re-send cannot extend @@ -178,32 +249,19 @@ func (s *Store) Put(kind Kind, mime, source string, data []byte) (Blob, error) { if err != nil { return Blob{}, err } - if err := os.MkdirAll(filepath.Dir(blobPath), 0o700); err != nil { + if err := os.MkdirAll(filepath.Dir(blobPath), dirPerm); err != nil { return Blob{}, fmt.Errorf("media: create bucket: %w", err) } - b := Blob{ID: id, Kind: kind, MIME: mime, Size: int64(len(data)), Source: source, - Created: s.now().UTC(), Path: blobPath} - - // A blob already here keeps its first-seen time. Re-sending the same photo - // every hour must not keep it alive past retention. - if prev, err := readMeta(metaPath); err == nil && !prev.Created.IsZero() { - b.Created = prev.Created - } + b := s.newBlob(kind, mime, source, id, blobPath, metaPath, int64(len(data))) // 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) + _, statErr := os.Stat(blobPath) + fresh := statErr != nil + if fresh { + if err := s.reserve(b.Size); err != nil { + return Blob{}, err } } @@ -216,10 +274,8 @@ func (s *Store) Put(kind Kind, mime, source string, data []byte) (Blob, error) { } if err := writeFile(blobPath, data); err != nil { _ = os.Remove(metaPath) - if already != nil { - s.totalMu.Lock() - s.total -= b.Size - s.totalMu.Unlock() + if fresh { + s.release(b.Size) } return Blob{}, err } @@ -264,43 +320,31 @@ func (s *Store) PutFile(kind Kind, mime, source, src string) (Blob, error) { if err != nil { return Blob{}, err } - if err := os.MkdirAll(filepath.Dir(blobPath), 0o700); err != nil { + if err := os.MkdirAll(filepath.Dir(blobPath), dirPerm); err != nil { return Blob{}, fmt.Errorf("media: create bucket: %w", err) } - b := Blob{ID: id, Kind: kind, MIME: mime, Size: info.Size(), Source: source, - Created: s.now().UTC(), Path: blobPath} - if prev, err := readMeta(metaPath); err == nil && !prev.Created.IsZero() { - b.Created = prev.Created - } - _, 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) + b := s.newBlob(kind, mime, source, id, blobPath, metaPath, info.Size()) + _, statErr := os.Stat(blobPath) + fresh := statErr != nil + if fresh { + if err := s.reserve(b.Size); err != nil { + return Blob{}, err } } if err := writeMeta(metaPath, b); err != nil { return Blob{}, err } - if already == nil { + if !fresh { // Same bytes already here. Drop the spool copy. _ = os.Remove(src) return b, nil } - if err := os.Chmod(src, 0o600); err != nil { + if err := os.Chmod(src, filePerm); err != nil { return Blob{}, fmt.Errorf("media: chmod spool: %w", err) } if err := os.Rename(src, blobPath); err != nil { _ = os.Remove(metaPath) - s.totalMu.Lock() - s.total -= b.Size - s.totalMu.Unlock() + s.release(b.Size) return Blob{}, fmt.Errorf("media: move spool: %w", err) } return b, nil @@ -311,15 +355,15 @@ func (s *Store) PutFile(kind Kind, mime, source, src string) (Blob, error) { // looks at it and List never reports it; PutFile is what turns it into a blob. // The caller owns removing it if it never gets that far. func (s *Store) SpoolFile(prefix string) (*os.File, error) { - dir := filepath.Join(s.dir, "spool") - if err := os.MkdirAll(dir, 0o700); err != nil { + dir := filepath.Join(s.dir, spoolName) + if err := os.MkdirAll(dir, dirPerm); err != nil { return nil, fmt.Errorf("media: create spool: %w", err) } f, err := os.CreateTemp(dir, prefix+"-*") if err != nil { return nil, fmt.Errorf("media: spool: %w", err) } - if err := f.Chmod(0o600); err != nil { + if err := f.Chmod(filePerm); err != nil { f.Close() return nil, fmt.Errorf("media: chmod spool: %w", err) } @@ -346,9 +390,8 @@ func (s *Store) Get(id string) (Blob, error) { if !validID(id) { return Blob{}, ErrBadID } - for _, kind := range []Kind{KindImage, KindAudio} { - metaPath := filepath.Join(s.dir, string(kind), id[:2], id+".json") - b, err := readMeta(metaPath) + for _, kind := range allKinds { + b, err := readMeta(filepath.Join(s.bucket(kind, id), id+metaExt)) if err != nil { continue } @@ -382,7 +425,7 @@ func (s *Store) Read(id string) (Blob, []byte, error) { // inside the retention window) that is cheap, and it means the sidecars are the // single source of truth with no index to fall out of sync. func (s *Store) List(kind Kind) ([]Blob, error) { - kinds := []Kind{KindImage, KindAudio} + kinds := allKinds if kind != "" { if !kind.Valid() { return nil, ErrBadKind @@ -391,17 +434,7 @@ func (s *Store) List(kind Kind) ([]Blob, error) { } var out []Blob for _, k := range kinds { - root := filepath.Join(s.dir, string(k)) - err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { - if err != nil { - if errors.Is(err, fs.ErrNotExist) { - return nil // kind never used; not an error - } - return err - } - if d.IsDir() || !strings.HasSuffix(path, ".json") { - return nil - } + err := s.walkSidecars(k, func(path string, _ fs.DirEntry) error { b, err := readMeta(path) if err != nil { return nil // a corrupt sidecar is skipped, not fatal @@ -431,8 +464,8 @@ func (s *Store) Delete(id string) error { if !validID(id) { return ErrBadID } - for _, kind := range []Kind{KindImage, KindAudio} { - bucket := filepath.Join(s.dir, string(kind), id[:2]) + for _, kind := range allKinds { + bucket := s.bucket(kind, id) entries, err := os.ReadDir(bucket) if err != nil { continue @@ -441,21 +474,17 @@ func (s *Store) Delete(id string) error { if !strings.HasPrefix(e.Name(), id) { continue } - path := filepath.Join(bucket, e.Name()) + // Only the bytes count against the budget, so the sidecar's own + // size is never given back. var size int64 - if info, err := e.Info(); err == nil && !strings.HasSuffix(e.Name(), ".json") { + if info, err := e.Info(); err == nil && !strings.HasSuffix(e.Name(), metaExt) { size = info.Size() } - if err := os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) { + 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 size > 0 { - s.totalMu.Lock() - s.total -= size - if s.total < 0 { - s.total = 0 - } - s.totalMu.Unlock() + s.release(size) } } } @@ -498,20 +527,9 @@ func (s *Store) Prune() (int, error) { // 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, ".") + for _, kind := range allKinds { + err := s.walkBlobFiles(kind, func(path string, d fs.DirEntry) error { + id, _, _ := strings.Cut(d.Name(), ".") if known[id] { return nil } @@ -525,12 +543,7 @@ func (s *Store) pruneOrphans(known map[string]bool, now time.Time) (int, error) 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() + s.release(info.Size()) deleted++ return nil }) @@ -541,13 +554,42 @@ func (s *Store) pruneOrphans(known map[string]bool, now time.Time) (int, error) return deleted, nil } +// walkSidecars visits every sidecar of one kind, and walkBlobFiles every file +// that is not one. The sidecars are the store's index and the rest are the +// bytes, so a walker always wants exactly one of the two. +func (s *Store) walkSidecars(kind Kind, fn func(path string, d fs.DirEntry) error) error { + return s.walkKind(kind, true, fn) +} + +func (s *Store) walkBlobFiles(kind Kind, fn func(path string, d fs.DirEntry) error) error { + return s.walkKind(kind, false, fn) +} + +// walkKind walks one kind's directory tree. A kind that was never used has no +// directory, which is silence rather than an error. +func (s *Store) walkKind(kind Kind, sidecars bool, fn func(path string, d fs.DirEntry) error) error { + root := filepath.Join(s.dir, string(kind)) + return 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, metaExt) != sidecars { + return nil + } + return fn(path, d) + }) +} + // paths returns the blob and sidecar paths for an id. func (s *Store) paths(kind Kind, id, mime string) (blobPath, metaPath string, err error) { if !validID(id) { return "", "", ErrBadID } - bucket := filepath.Join(s.dir, string(kind), id[:2]) - return filepath.Join(bucket, id+extFor(mime, kind)), filepath.Join(bucket, id+".json"), nil + bucket := s.bucket(kind, id) + return filepath.Join(bucket, id+extFor(mime, kind)), filepath.Join(bucket, id+metaExt), nil } // locate finds the stored bytes for an id whose extension we do not know, @@ -556,14 +598,14 @@ func (s *Store) locate(kind Kind, id string) (string, error) { if !validID(id) { return "", ErrBadID } - bucket := filepath.Join(s.dir, string(kind), id[:2]) + bucket := s.bucket(kind, id) entries, err := os.ReadDir(bucket) if err != nil { return "", ErrNotFound } for _, e := range entries { name := e.Name() - if strings.HasPrefix(name, id) && !strings.HasSuffix(name, ".json") { + if strings.HasPrefix(name, id) && !strings.HasSuffix(name, metaExt) { return filepath.Join(bucket, name), nil } } @@ -573,7 +615,7 @@ func (s *Store) locate(kind Kind, id string) (string, error) { // validID guards every path built from an id. Without it a caller-supplied id // is a path traversal: Get("../../etc/passwd") would read outside the store. func validID(id string) bool { - if len(id) != 64 { + if len(id) != idLen { return false } for i := 0; i < len(id); i++ { @@ -621,7 +663,7 @@ func writeFile(path string, data []byte) error { return fmt.Errorf("media: temp: %w", err) } defer os.Remove(tmp.Name()) - if err := tmp.Chmod(0o600); err != nil { + if err := tmp.Chmod(filePerm); err != nil { tmp.Close() return fmt.Errorf("media: chmod: %w", err) }