kami/Maven
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

media: move a file into the store instead of reading it in

Browse Source 
Put takes a []byte, so storing a recording meant the whole recording in
memory. A two hour meeting at 16 kHz mono is about 230 MB of WAV, and
building it from PCM held a second copy of the same size in the process
that also owns the database and the resident model. PutFile stats the
file, hashes it in a stream and renames it into place, so the peak is one
buffer regardless of length. SpoolFile hands out the scratch file it
moves from, under the media dir so it shares the same disk and the same
permissions.

Audio also gets its own per blob cap of 512 MiB. The image cap of 64 MiB
is 35 minutes of audio, which contradicted the two hour session cap: the
long meeting was exactly the one that failed to store.

audio.WAVHeader is split out of WAVFromPCM because a spooled capture
writes a placeholder header first and stamps the real length at the end.

Found in review of #73.
This commit is contained in:
kami
2026-08-01 14:36:17 +04:00
parent 543aefde4b
commit 71b42e31bd
2 changed files with 165 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/audio/pcmwav.go
+23 -4
View File
@@ -94,14 +94,33 @@ func PCMFromWAV(wav []byte) (Format, []byte, error) {
// header so the result can be written to disk and played with `aplay`.
// Used by the reference client to write the TTS reply; not on the wire.
func WAVFromPCM(format Format, pcm []byte) ([]byte, error) {
hdr, err := WAVHeader(format, len(pcm))
if err != nil {
return nil, err
}
out := make([]byte, wavHeaderSize+len(pcm))
copy(out, hdr)
copy(out[wavHeaderSize:], pcm)
return out, nil
}
// WAVHeaderSize is the fixed size of the header WAVHeader writes. A caller
// spooling audio to a file reserves this many bytes up front and rewrites them
// once it knows the length.
const WAVHeaderSize = wavHeaderSize
// WAVHeader builds just the 44-byte canonical header for n bytes of PCM. It
// exists so a long recording can be written straight to a file: holding the
// whole meeting in memory to prepend 44 bytes is what the streaming path is
// avoiding.
func WAVHeader(format Format, n int) ([]byte, error) {
if !format.IsValid() {
return nil, fmt.Errorf("audio: WAVFromPCM: %w: %+v", ErrNotCanonicalPCM, format)
}
out := make([]byte, wavHeaderSize+len(pcm))
copy(out[wavHeaderSize:], pcm)
out := make([]byte, wavHeaderSize)
// RIFF header
copy(out[0:4], []byte("RIFF"))
binary.LittleEndian.PutUint32(out[4:8], uint32(36+len(pcm)))
binary.LittleEndian.PutUint32(out[4:8], uint32(36+n))
copy(out[8:12], []byte("WAVE"))
// fmt chunk
copy(out[12:16], []byte("fmt "))
@@ -116,6 +135,6 @@ func WAVFromPCM(format Format, pcm []byte) ([]byte, error) {
binary.LittleEndian.PutUint16(out[34:36], uint16(format.SampleBits))
// data chunk
copy(out[36:40], []byte("data"))
binary.LittleEndian.PutUint32(out[40:44], uint32(len(pcm)))
binary.LittleEndian.PutUint32(out[40:44], uint32(n))
return out, nil
}
internal/media/store.go
+142 -3
View File
@@ -6,6 +6,7 @@ import (
"encoding/json"
"errors"
"fmt"
"io"
"io/fs"
"os"
"path/filepath"
@@ -20,6 +21,14 @@ import (
// ceiling; a single item bigger than that is a mistake, not a meeting.
const DefaultMaxBytes int64 = 64 << 20
// DefaultMaxAudioBytes — the per-blob cap for audio. Separate from
// DefaultMaxBytes because the two kinds are not the same size of thing: an
// image over 64 MiB is a mistake, and a two-hour meeting at 16 kHz mono is
// about 230 MB of PCM by design. With one shared cap, capture's own
// DefaultMaxDuration of two hours and this store's 64 MiB contradicted each
// other, and the meeting that hit the limit was the one that failed to store.
const DefaultMaxAudioBytes int64 = 512 << 20
// DefaultRetention — how long a blob is kept when no retention is configured.
// Seven days is long enough to re-run a transcription that came out wrong and
// short enough that "she has a month of my meetings on disk" is never true.
@@ -45,6 +54,7 @@ var ErrStoreFull = errors.New("media: store is full")
type Store struct {
dir string
maxBytes int64
maxAudio int64
maxTotal int64
retention time.Duration
now func() time.Time
@@ -81,16 +91,24 @@ func OpenWithBudget(dir string, maxBytes, maxTotal int64, retention time.Duratio
if maxBytes <= 0 {
maxBytes = DefaultMaxBytes
}
maxAudio := DefaultMaxAudioBytes
if maxBytes > maxAudio {
maxAudio = maxBytes
}
if maxTotal <= 0 {
maxTotal = DefaultMaxTotalBytes
}
if maxTotal < maxAudio {
maxAudio = maxTotal
}
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
}
s := &Store{dir: abs, maxBytes: maxBytes, maxTotal: maxTotal, retention: retention, now: time.Now}
s := &Store{dir: abs, maxBytes: maxBytes, maxAudio: maxAudio, maxTotal: maxTotal,
retention: retention, now: time.Now}
s.total = s.measure()
return s, nil
}
@@ -99,7 +117,13 @@ 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")
_ = 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") {
return nil //nolint:nilerr // an unreadable corner is not worth refusing to boot over
}
@@ -144,8 +168,8 @@ func (s *Store) Put(kind Kind, mime, source string, data []byte) (Blob, error) {
if len(data) == 0 {
return Blob{}, ErrEmpty
}
if int64(len(data)) > s.maxBytes {
return Blob{}, fmt.Errorf("%w: %d > %d", ErrTooLarge, len(data), s.maxBytes)
if cap := s.capFor(kind); int64(len(data)) > cap {
return Blob{}, fmt.Errorf("%w: %d > %d", ErrTooLarge, len(data), cap)
}
sum := sha256.Sum256(data)
id := hex.EncodeToString(sum[:])
@@ -202,6 +226,121 @@ func (s *Store) Put(kind Kind, mime, source string, data []byte) (Blob, error) {
return b, nil
}
// capFor is the per-blob cap for a kind. Audio has its own, larger one.
func (s *Store) capFor(kind Kind) int64 {
if kind == KindAudio {
return s.maxAudio
}
return s.maxBytes
}
// PutFile stores a file that is already on disk, by moving it into place rather
// than reading it into memory. It exists for meeting audio: a two-hour capture
// is a couple of hundred megabytes, and Put's []byte means that much heap in
// the process that owns the database, twice over while the WAV is built.
//
// src is consumed: on success it has been renamed into the store, and on a
// duplicate it is removed. On failure it is left where it is, so a caller that
// still needs the bytes can fall back to reading them.
func (s *Store) PutFile(kind Kind, mime, source, src string) (Blob, error) {
if !kind.Valid() {
return Blob{}, ErrBadKind
}
info, err := os.Stat(src)
if err != nil {
return Blob{}, fmt.Errorf("media: stat spool: %w", err)
}
if info.Size() == 0 {
return Blob{}, ErrEmpty
}
if cap := s.capFor(kind); info.Size() > cap {
return Blob{}, fmt.Errorf("%w: %d > %d", ErrTooLarge, info.Size(), cap)
}
id, err := hashFile(src)
if err != nil {
return Blob{}, err
}
blobPath, metaPath, err := s.paths(kind, id, mime)
if err != nil {
return Blob{}, err
}
if err := os.MkdirAll(filepath.Dir(blobPath), 0o700); 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)
}
}
if err := writeMeta(metaPath, b); err != nil {
return Blob{}, err
}
if already == nil {
// Same bytes already here. Drop the spool copy.
_ = os.Remove(src)
return b, nil
}
if err := os.Chmod(src, 0o600); 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()
return Blob{}, fmt.Errorf("media: move spool: %w", err)
}
return b, nil
}
// SpoolFile creates an empty file under the store, outside the kind
// directories, for a caller that is writing a blob incrementally. Prune never
// 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 {
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 {
f.Close()
return nil, fmt.Errorf("media: chmod spool: %w", err)
}
return f, nil
}
// hashFile streams the digest so the id costs one buffer rather than the whole
// file.
func hashFile(path string) (string, error) {
f, err := os.Open(path)
if err != nil {
return "", fmt.Errorf("media: open spool: %w", err)
}
defer f.Close()
h := sha256.New()
if _, err := io.Copy(h, f); err != nil {
return "", fmt.Errorf("media: hash spool: %w", err)
}
return hex.EncodeToString(h.Sum(nil)), nil
}
// Get returns the blob's metadata without reading its bytes.
func (s *Store) Get(id string) (Blob, error) {
if !validID(id) {