summaryrefslogtreecommitdiff
path: root/server/internal
diff options
context:
space:
mode:
Diffstat (limited to 'server/internal')
-rw-r--r--server/internal/api/api.go11
-rw-r--r--server/internal/store/store.go52
-rw-r--r--server/internal/store/store_test.go105
-rw-r--r--server/internal/words/pipeline.go77
-rw-r--r--server/internal/words/pipeline_test.go69
5 files changed, 305 insertions, 9 deletions
diff --git a/server/internal/api/api.go b/server/internal/api/api.go
index 4b7fb2d..6114c1b 100644
--- a/server/internal/api/api.go
+++ b/server/internal/api/api.go
@@ -3,6 +3,7 @@ package api
import (
"encoding/json"
+ "log"
"net/http"
"net/url"
"os"
@@ -178,7 +179,15 @@ func (s *Server) subscriptions(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "expected {\"feeds\":[...]}"})
return
}
- removed := s.state.SetFeeds(body.Feeds)
+ removed, err := s.state.SetFeedsChecked(body.Feeds)
+ if err != nil {
+ // Refused rather than applied: losing most of a library in one push is another device
+ // or a bug. Said out loud, because a silent refusal looks exactly like a silent
+ // acceptance from the phone's end.
+ log.Printf("subscriptions: %v", err)
+ writeJSON(w, http.StatusConflict, map[string]string{"error": err.Error()})
+ return
+ }
// After the list, so a tier for a feed that was just removed is dropped with it.
s.state.SetTiers(body.Tiers)
if s.opts.OnSubscriptions != nil {
diff --git a/server/internal/store/store.go b/server/internal/store/store.go
index f69ff90..2e20f2e 100644
--- a/server/internal/store/store.go
+++ b/server/internal/store/store.go
@@ -360,6 +360,58 @@ func (s *State) Save(path string) error {
return nil
}
+/*
+guardFloor is the library size below which a shrinking push is not questioned.
+
+A new phone's first push legitimately replaces nothing with everything, and a library of three
+can legitimately go to one. Ten is where "most of it vanished" stops being plausible as
+something somebody did on purpose.
+*/
+const guardFloor = 10
+
+/*
+SetFeedsChecked replaces the subscription list unless doing so would empty the library.
+
+This exists because it happened. A test device was pointed at the live server to check
+playback, its own short subscription list replaced the real one, and the server reclaimed every
+show that device had never heard of: 290 mirrored episodes and days of recognition, gone in one
+second.
+
+The list is still replaced wholesale -- the phone owns it, and reconciling two editable copies
+of something changed this rarely would cost more in sync bugs than it saves. What is refused is
+the magnitude. Losing most of a library in a single push is a second device or a bug; it is
+never somebody unsubscribing.
+
+Returns what was removed, or an error and nothing removed.
+*/
+func (s *State) SetFeedsChecked(feeds []string) (removed []string, err error) {
+ s.mu.RLock()
+ existing := len(s.feeds)
+ s.mu.RUnlock()
+
+ if existing >= guardFloor {
+ wanted := map[string]bool{}
+ for _, f := range feeds {
+ wanted[strings.TrimSpace(f)] = true
+ }
+ kept := 0
+ s.mu.RLock()
+ for _, f := range s.feeds {
+ if wanted[f] {
+ kept++
+ }
+ }
+ s.mu.RUnlock()
+ if kept*2 < existing {
+ return nil, fmt.Errorf(
+ "refusing a subscription push that keeps %d of %d feeds: losing most of a "+
+ "library in one push is another device or a bug, not an unsubscribe",
+ kept, existing)
+ }
+ }
+ return s.SetFeeds(feeds), nil
+}
+
// SetFeeds replaces the subscription list and reports the feeds that are no longer wanted.
//
// Wholesale replacement because the phone owns this list; reconciling two editable copies of
diff --git a/server/internal/store/store_test.go b/server/internal/store/store_test.go
index f0c111b..a88fdd4 100644
--- a/server/internal/store/store_test.go
+++ b/server/internal/store/store_test.go
@@ -1,8 +1,10 @@
package store
import (
+ "fmt"
"os"
"path/filepath"
+ "strings"
"testing"
"github.com/terst/listen/server/internal/feed"
@@ -640,3 +642,106 @@ func TestABreakBothDetectorsFoundSaysSo(t *testing.T) {
t.Errorf("origin = %q, want both named", got[0].Origin)
}
}
+
+/*
+A subscription push that would empty the library is refused.
+
+This happened. A test device was pointed at the live server to check playback, its own short
+subscription list replaced the real one, and the server reclaimed every show the test device
+had never heard of -- 290 mirrored episodes and days of recognition, deleted in one second at
+07:07:13 on 2026-10-31.
+
+The list is still wholesale-replaced, because the phone owns it and reconciling two editable
+copies would cost more in sync bugs than it saves. What is refused is the magnitude: losing
+most of a library in one push is a second device or a bug, never somebody unsubscribing.
+
+Below [guardFloor] shows there is nothing to protect -- a new phone's first push legitimately
+replaces nothing with everything, and a small library can legitimately halve.
+*/
+
+func TestAPushThatWouldEmptyTheLibraryIsRefused(t *testing.T) {
+ s := New()
+ many := make([]string, 0, 20)
+ for i := 0; i < 20; i++ {
+ many = append(many, fmt.Sprintf("https://feed%d", i))
+ }
+ s.SetFeeds(many)
+
+ removed, err := s.SetFeedsChecked([]string{"https://feed0", "https://feed1"})
+
+ if err == nil {
+ t.Fatal("accepted a push that drops 18 of 20 shows")
+ }
+ if len(removed) != 0 {
+ t.Errorf("removed %d shows anyway", len(removed))
+ }
+ if len(s.Feeds()) != 20 {
+ t.Errorf("the list is %d long, want it untouched", len(s.Feeds()))
+ }
+}
+
+// Unsubscribing from a few shows is ordinary and must keep working.
+func TestAnOrdinaryUnsubscribeIsAccepted(t *testing.T) {
+ s := New()
+ many := make([]string, 0, 20)
+ for i := 0; i < 20; i++ {
+ many = append(many, fmt.Sprintf("https://feed%d", i))
+ }
+ s.SetFeeds(many)
+
+ // Merged so there are shows to reclaim; removed names shows with audio, not feed URLs.
+ for _, url := range many {
+ s.Merge(url, "A Show", nil)
+ }
+
+ removed, err := s.SetFeedsChecked(many[:17])
+
+ if err != nil {
+ t.Fatalf("refused an ordinary unsubscribe: %v", err)
+ }
+ if len(removed) != 3 {
+ t.Errorf("removed %d, want 3", len(removed))
+ }
+ if len(s.Feeds()) != 17 {
+ t.Errorf("the list is %d long, want 17", len(s.Feeds()))
+ }
+}
+
+// A first push replaces nothing with everything, which must not look like a catastrophe.
+func TestAFirstPushIsAccepted(t *testing.T) {
+ s := New()
+
+ if _, err := s.SetFeedsChecked([]string{"https://a", "https://b"}); err != nil {
+ t.Fatalf("refused a first push: %v", err)
+ }
+}
+
+// A small library can legitimately halve; there is nothing worth protecting yet.
+func TestASmallLibraryIsNotGuarded(t *testing.T) {
+ s := New()
+ s.SetFeeds([]string{"https://a", "https://b", "https://c"})
+
+ if _, err := s.SetFeedsChecked([]string{"https://a"}); err != nil {
+ t.Fatalf("refused a small library shrinking: %v", err)
+ }
+}
+
+// Clearing the list deliberately is still possible, because an empty push is unambiguous in a
+// way a short one is not -- but only from empty, so it cannot happen by accident here.
+func TestTheGuardNamesWhatItRefused(t *testing.T) {
+ s := New()
+ many := make([]string, 0, 12)
+ for i := 0; i < 12; i++ {
+ many = append(many, fmt.Sprintf("https://feed%d", i))
+ }
+ s.SetFeeds(many)
+
+ _, err := s.SetFeedsChecked(nil)
+
+ if err == nil {
+ t.Fatal("accepted an empty push over a full library")
+ }
+ if !strings.Contains(err.Error(), "12") {
+ t.Errorf("error %q does not say how much it would have lost", err)
+ }
+}
diff --git a/server/internal/words/pipeline.go b/server/internal/words/pipeline.go
index b37efa3..2ae2c4e 100644
--- a/server/internal/words/pipeline.go
+++ b/server/internal/words/pipeline.go
@@ -68,9 +68,20 @@ type Pipeline struct {
// Save persists the state. Called as each episode finishes rather than at the end,
// because these passes run for tens of minutes and a restart in the middle should not
// lose the ones already done.
+ //
+ // Optional: a nil Save means the caller is not persisting, which is what every test wants.
+ // It must be optional rather than assumed, because assuming it shipped a server that
+ // panicked on its first recognised episode.
Save func()
}
+// save persists the state if the caller asked for persistence.
+func (p *Pipeline) save() {
+ if p.Save != nil {
+ p.Save()
+ }
+}
+
/*
Forget drops everything known about an episode's words.
@@ -221,10 +232,10 @@ func (p *Pipeline) Recognise(ctx context.Context) {
log.Printf("recognition abandoned after %d failures in a row -- "+
"giving those attempts back, since this is the engine and not the episodes",
breaker.Run())
- p.Save()
+ p.save()
return
}
- p.Save()
+ p.save()
continue
}
@@ -244,10 +255,10 @@ func (p *Pipeline) Recognise(ctx context.Context) {
log.Printf("recognition abandoned after %d failures in a row -- "+
"giving those attempts back, since this is the engine and not the episodes",
breaker.Run())
- p.Save()
+ p.save()
return
}
- p.Save()
+ p.save()
continue
}
e.TranscriptSource = string(transcript.SourceASR)
@@ -262,7 +273,7 @@ func (p *Pipeline) Recognise(ctx context.Context) {
log.Printf("transcribed %q: %d lines, %s, took %s",
e.Title, len(stored.Cues), adAccounting(e, stored.Ads),
time.Since(started).Round(time.Second))
- p.Save()
+ p.save()
}
if done > 0 || failed > 0 {
@@ -569,10 +580,10 @@ func (p *Pipeline) SecondCopy(ctx context.Context) {
}
log.Printf("refetching abandoned after %d failures in a row -- giving those "+
"attempts back, since this is the machine and not the episodes", breaker.Run())
- p.Save()
+ p.save()
return
}
- p.Save()
+ p.save()
continue
}
breaker.Success()
@@ -583,7 +594,7 @@ func (p *Pipeline) SecondCopy(ctx context.Context) {
// then nothing for subtraction to see.
log.Printf("second copy of %q: %s against the mirrored %s, %d lines, %s",
e.Title, clock(secondMs), clock(e.DurationMs), e.MasterCues, adAccounting(e, found))
- p.Save()
+ p.save()
}
if done > 0 || failed > 0 {
log.Printf("second copies -- %d done, %d failed, %d still waiting",
@@ -704,3 +715,53 @@ func (p *Pipeline) pendingRecognition() []*store.Episode {
})
return pending
}
+
+/*
+Reclaim restores what the transcript files on disk say about episodes the state has forgotten.
+
+Recognition is the expensive thing this server does -- about five minutes of CPU an episode --
+and its output is a file keyed by the episode's id, which is derived from the feed and the
+guid. So a transcript survives an episode record being lost and matches again the moment that
+episode is mirrored afresh.
+
+That is not hypothetical. A test device pointed at the live server replaced the subscription
+list with its own, the server reclaimed every show the device had never heard of, and 290
+mirrored episodes went with it -- while their transcripts stayed on disk. Without this, every
+one of them would be recognised again from scratch.
+
+Only where the file genuinely matches the episode. A transcript whose last cue runs past the
+end of the audio describes a different recording, which is the same check transcript.Fits
+makes of a publisher's.
+*/
+func (p *Pipeline) Reclaim() {
+ restored := 0
+ for _, show := range p.State.Shows() {
+ for _, e := range show.Episodes {
+ if !e.Mirrored() || e.HasTranscript() || !p.Transcripts.Has(e.ID) {
+ continue
+ }
+ held, err := p.Transcripts.Get(e.ID)
+ if err != nil || len(held.Cues) == 0 {
+ continue
+ }
+ if e.DurationMs > 0 && held.LastEndMs() > e.DurationMs {
+ // Describes a longer recording than the one on disk, so it is not this copy.
+ continue
+ }
+ e.TranscriptSource = string(held.Source)
+ e.TranscriptCues = len(held.Cues)
+ e.TranscriptNote = ""
+ if p.Transcripts.HasMaster(e.ID) {
+ if master, err := p.Transcripts.Master(e.ID); err == nil && len(master.Cues) > 0 {
+ e.MasterCues = len(master.Cues)
+ e.MasterSource = string(master.Source)
+ }
+ }
+ p.publishWordBreaks(e, held.Ads)
+ restored++
+ }
+ }
+ if restored > 0 {
+ log.Printf("recovered the words of %d episode(s) from transcripts already on disk", restored)
+ }
+}
diff --git a/server/internal/words/pipeline_test.go b/server/internal/words/pipeline_test.go
index 180b807..c010c8e 100644
--- a/server/internal/words/pipeline_test.go
+++ b/server/internal/words/pipeline_test.go
@@ -135,3 +135,72 @@ func TestRecurringNeedsEnoughEpisodesToMeanAnything(t *testing.T) {
}
}
}
+
+/*
+Recovering recognition that outlived its episode record.
+
+A transcript is keyed by the episode's id, which comes from the feed and the guid, so it
+survives the record being lost and matches again when the episode is mirrored afresh. Without
+this, 290 episodes whose records were destroyed would each be recognised again -- about twenty
+hours of CPU for text already sitting on disk.
+*/
+func TestReclaimRestoresWordsFromDisk(t *testing.T) {
+ dir := t.TempDir()
+ held := transcript.Store{Dir: dir}
+ if err := held.Put(transcript.Transcript{
+ EpisodeID: "e1", Source: transcript.SourceASR,
+ Cues: []transcript.Cue{{StartMs: 0, EndMs: 1000, Text: "words"}},
+ }); err != nil {
+ t.Fatalf("Put: %v", err)
+ }
+
+ state := store.New()
+ state.Merge("https://f", "A Show", nil)
+ show := state.Show("https://f")
+ show.Episodes = []*store.Episode{{ID: "e1", Path: "e1.mp3", DurationMs: 60_000}}
+
+ (&Pipeline{State: state, Transcripts: held}).Reclaim()
+
+ got := show.Episodes[0]
+ if got.TranscriptCues != 1 || got.TranscriptSource != string(transcript.SourceASR) {
+ t.Errorf("got %+v, want the words recovered", got)
+ }
+}
+
+// A transcript longer than the audio describes a different recording, which is the same check
+// made of a publisher's transcript.
+func TestReclaimRefusesATranscriptOfAnotherRecording(t *testing.T) {
+ dir := t.TempDir()
+ held := transcript.Store{Dir: dir}
+ if err := held.Put(transcript.Transcript{
+ EpisodeID: "e1", Source: transcript.SourceASR,
+ Cues: []transcript.Cue{{StartMs: 0, EndMs: 900_000, Text: "a much longer episode"}},
+ }); err != nil {
+ t.Fatalf("Put: %v", err)
+ }
+
+ state := store.New()
+ state.Merge("https://f", "A Show", nil)
+ show := state.Show("https://f")
+ show.Episodes = []*store.Episode{{ID: "e1", Path: "e1.mp3", DurationMs: 60_000}}
+
+ (&Pipeline{State: state, Transcripts: held}).Reclaim()
+
+ if show.Episodes[0].TranscriptCues != 0 {
+ t.Error("took a transcript that outlasts the audio")
+ }
+}
+
+// An episode that already has its words is left alone.
+func TestReclaimLeavesWhatIsAlreadyThere(t *testing.T) {
+ state := store.New()
+ state.Merge("https://f", "A Show", nil)
+ show := state.Show("https://f")
+ show.Episodes = []*store.Episode{{ID: "e1", Path: "e1.mp3", TranscriptCues: 99}}
+
+ (&Pipeline{State: state, Transcripts: transcript.Store{Dir: t.TempDir()}}).Reclaim()
+
+ if show.Episodes[0].TranscriptCues != 99 {
+ t.Error("overwrote words that were already there")
+ }
+}
generated by cgit v1.2.3 (git 2.39.1) at 2026-11-14 22:36:21 +0000