From f5bf05bbc0480a7edcd098a8e40c6c16af5d5f68 Mon Sep 17 00:00:00 2001 From: Peter Stone Date: Sat, 24 Oct 2026 09:33:49 +0000 Subject: Fix the shadow run: resolve audio paths, and survive a bad file The first real shadow run died on its first batch. Two faults, and the second is the one worth keeping. store.Episode.Path is relative to the audio directory. Detection joins it before handing it to the decoder and this did not, so ffmpeg went looking in the working directory and found nothing. Then that single failure aborted the whole show's learning. Every other per-item loop in this server isolates one item's failure -- the refresh sweep, the poll loop, the detection pass all say so in their own comments -- and this one propagated instead. An unreadable episode is now counted and stepped over. Counted once per episode rather than once per failure: a file that cannot be read fails both the matching pass and the harvesting pass, and reporting it twice would overstate the damage. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_018AwnALZL6bD4VBtATDVv3B --- server/cmd/listend/main.go | 12 +++----- server/cmd/listend/shadow.go | 13 ++++++-- server/internal/signature/shadow.go | 33 ++++++++++++-------- server/internal/signature/shadow_test.go | 52 ++++++++++++++++++++++++++++++++ 4 files changed, 88 insertions(+), 22 deletions(-) diff --git a/server/cmd/listend/main.go b/server/cmd/listend/main.go index a556f5e..a4400d2 100644 --- a/server/cmd/listend/main.go +++ b/server/cmd/listend/main.go @@ -77,7 +77,6 @@ func main() { keep: *keep, mirror: &mirror.Mirror{AudioDir: audioDir, Probe: decoder}, decoder: decoder, - audioDir: audioDir, requests: make(chan request, 64), wakeups: make(chan struct{}, 1), } @@ -154,12 +153,11 @@ type worker struct { // signatures is the library the shadow run learns into. Nothing it holds is published yet. signatures *signature.Library signaturePath string - keep int - mirror *mirror.Mirror - decoder decode.Decoder - audioDir string - requests chan request - wakeups chan struct{} + keep int + mirror *mirror.Mirror + decoder decode.Decoder + requests chan request + wakeups chan struct{} } // wake asks for a poll now. Coalescing, so a burst of subscription updates is one poll. diff --git a/server/cmd/listend/shadow.go b/server/cmd/listend/shadow.go index e9376f1..7531115 100644 --- a/server/cmd/listend/shadow.go +++ b/server/cmd/listend/shadow.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log" + "path/filepath" "sort" "time" @@ -68,7 +69,7 @@ func (w *worker) runShadow(ctx context.Context, show *store.Show, episodes []*st batch.Episodes = append(batch.Episodes, ep) } - report, err := signature.NewShadow(w.signatures, audioFor(w.decoder, episodes)).Run(ctx, batch) + report, err := signature.NewShadow(w.signatures, audioFor(w.decoder, w.mirror.AudioDir, episodes)).Run(ctx, batch) if err != nil { log.Printf("%s: shadow run failed: %v", show.Title, err) return @@ -118,10 +119,16 @@ const shadowMinOccasions = 2 // // The paths come from the batch rather than from a lookup: these are the episodes detection // just finished with, so their files are the ones it read and are certainly still there. -func audioFor(decoder decode.Decoder, episodes []*store.Episode) signature.Audio { +// +// store.Episode.Path is relative to the audio directory, which is what detection joins it with +// before handing it to the decoder. Passing it raw sent ffmpeg looking in the working directory, +// and the first real shadow run died on it. +func audioFor(decoder decode.Decoder, audioDir string, episodes []*store.Episode) signature.Audio { paths := make(map[string]string, len(episodes)) for _, e := range episodes { - paths[e.ID] = e.Path + if e.Path != "" { + paths[e.ID] = filepath.Join(audioDir, e.Path) + } } return episodeAudio{decoder: decoder, pathOf: func(id string) string { return paths[id] }} } diff --git a/server/internal/signature/shadow.go b/server/internal/signature/shadow.go index 9c2fe6b..412597c 100644 --- a/server/internal/signature/shadow.go +++ b/server/internal/signature/shadow.go @@ -1,9 +1,6 @@ package signature -import ( - "context" - "fmt" -) +import "context" // Audio supplies decoded audio at [SampleRate]. type Audio interface { @@ -89,13 +86,17 @@ func NewShadow(library *Library, audio Audio) *Shadow { // they came from, and every run would report itself as a triumph. func (s *Shadow) Run(ctx context.Context, batch ShadowBatch) (Report, error) { var report Report + // Counted per episode, not per failure: an episode that could not be read fails both the + // matching pass and the harvesting pass, and reporting it twice would overstate the damage. + unreadable := map[string]bool{} for _, ep := range batch.Episodes { if ctx.Err() != nil { + report.Skipped = len(unreadable) return report, ctx.Err() } samples, err := s.audio.Episode(ctx, ep.ID) if err != nil || len(samples) == 0 { - report.Skipped++ + unreadable[ep.ID] = true continue } for _, m := range s.library.Match(Take(samples), s.MinSightings) { @@ -112,29 +113,37 @@ func (s *Shadow) Run(ctx context.Context, batch ShadowBatch) (Report, error) { for _, ep := range batch.Episodes { if ctx.Err() != nil { + report.Skipped = len(unreadable) return report, ctx.Err() } for _, br := range ep.Breaks { - if err := s.harvest(ctx, batch, ep, br, &report); err != nil { - return report, err + // A break this cannot read is counted and stepped over. Every other per-item loop + // in this server isolates a single failure the same way, and the first real run + // lost a whole show's learning to one episode whose path was wrong. + if !s.harvest(ctx, batch, ep, br, &report) { + unreadable[ep.ID] = true + break } } } + report.Skipped = len(unreadable) return report, nil } -func (s *Shadow) harvest(ctx context.Context, batch ShadowBatch, ep ShadowEpisode, br ShadowBreak, report *Report) error { +// harvest stores one break's audio as a signature, and reports whether the episode could be +// read at all. A false means give up on this episode, not on the batch. +func (s *Shadow) harvest(ctx context.Context, batch ShadowBatch, ep ShadowEpisode, br ShadowBreak, report *Report) bool { duration := br.EndMs - br.StartMs if duration <= 0 { - return nil + return true } samples, err := s.audio.Range(ctx, ep.ID, br.StartMs, duration) if err != nil { - return fmt.Errorf("decode %s %d-%dms: %w", ep.ID, br.StartMs, br.EndMs, err) + return false } fp := Take(samples) if len(fp.Hashes) == 0 { - return nil + return true } _, isNew := s.library.Observe(Observation{ Fingerprint: fp, @@ -149,7 +158,7 @@ func (s *Shadow) harvest(ctx context.Context, batch ShadowBatch, ep ShadowEpisod } else { report.Reinforced++ } - return nil + return true } // kindFor maps what detection called a break onto what the library stores. diff --git a/server/internal/signature/shadow_test.go b/server/internal/signature/shadow_test.go index 3e6336f..f6c6676 100644 --- a/server/internal/signature/shadow_test.go +++ b/server/internal/signature/shadow_test.go @@ -2,6 +2,7 @@ package signature_test import ( "context" + "errors" "testing" "github.com/terst/listen/server/internal/signature" @@ -14,6 +15,27 @@ func (f fakeAudio) Episode(_ context.Context, id string) ([]float64, error) { return f[id], nil } +// failingAudio serves one episode and refuses the rest, which is what an unreadable mirrored +// file looks like. +type failingAudio struct { + fakeAudio + refuse map[string]bool +} + +func (f failingAudio) Episode(ctx context.Context, id string) ([]float64, error) { + if f.refuse[id] { + return nil, errors.New("no such file") + } + return f.fakeAudio.Episode(ctx, id) +} + +func (f failingAudio) Range(ctx context.Context, id string, startMs, durMs int64) ([]float64, error) { + if f.refuse[id] { + return nil, errors.New("no such file") + } + return f.fakeAudio.Range(ctx, id, startMs, durMs) +} + func (f fakeAudio) Range(_ context.Context, id string, startMs, durMs int64) ([]float64, error) { all := f[id] from := int(startMs) * signature.SampleRate / 1000 @@ -181,3 +203,33 @@ func TestAShadowRunPublishesNothing(t *testing.T) { t.Fatal("the shadow run changed the breaks it was given") } } + +// One unreadable file must not cost the batch its learning. Every other per-item loop in this +// server isolates a single failure, and the first real shadow run aborted a whole show's pass +// because one episode's path was wrong. +func TestOneUnreadableEpisodeDoesNotCostTheBatchItsLearning(t *testing.T) { + ad := tone(42, seconds(20)) + cue := tone(7, seconds(8)) + audio := failingAudio{ + fakeAudio: fakeAudio{"good": planted(1, ad, 300, 60), "bad": planted(2, cue, 300, 30)}, + refuse: map[string]bool{"bad": true}, + } + l := signature.NewLibrary() + + report, err := signature.NewShadow(l, audio).Run(context.Background(), batch( + signature.ShadowEpisode{ID: "bad", + Breaks: []signature.ShadowBreak{{StartMs: 30_000, EndMs: 38_000, Kind: "ad", Partners: 3}}}, + signature.ShadowEpisode{ID: "good", + Breaks: []signature.ShadowBreak{{StartMs: 60_000, EndMs: 80_000, Kind: "ad", Partners: 4}}}, + )) + + if err != nil { + t.Fatalf("one bad file failed the whole run: %v", err) + } + if report.Harvested != 1 { + t.Fatalf("harvested %d, want the one episode that could be read", report.Harvested) + } + if report.Skipped != 1 { + t.Fatalf("skipped %d, want 1 -- an unreadable file should be counted, not silent", report.Skipped) + } +} -- cgit v1.2.3