From 1202b52eb8f73c6d342b7ce3b548b0c711722cb1 Mon Sep 17 00:00:00 2001 From: cplieger <917744+cplieger@users.noreply.github.com> Date: Wed, 2 Sep 2026 11:53:31 +0200 Subject: [PATCH] refactor: move the deep-analysis last-run record out of the cache The deep-analysis safety net now keeps its last-run record in a dedicated file, /config/.plex-language-sync-last-run, instead of a last_scheduler_run field in state.json. The restart behavior is unchanged: a startup pass runs only when no record exists or the last completed pass is older than DEEP_SCAN_INTERVAL, and the replay look-back window still extends from the previous completed pass, capped at 30 days. state.json no longer carries the last_scheduler_run key. Files written by earlier versions still load; the retired key is ignored on read. On the first start after this upgrade the record file does not exist yet, so one initial deep-analysis pass runs even if the previous version completed one recently, and that pass uses the 24-hour look-back floor. The pass records its completion and the schedule is back to normal from then on. One boundary detail changed: a record aged exactly DEEP_SCAN_INTERVAL now counts as due, where it previously had to be strictly older. --- README.md | 2 +- internal/cache/cache.go | 20 +- internal/cache/cache_test.go | 49 ++-- internal/cache/contract.go | 30 -- internal/cache/scheduler.go | 28 -- internal/cache/scheduler_test.go | 34 --- internal/deepscan/deepscan.go | 70 +++-- internal/deepscan/deepscan_test.go | 303 +++++++++++++-------- internal/deepscan/deps.go | 7 +- internal/testsupport/fakeapi/cache.go | 20 +- internal/testsupport/fakeapi/cache_test.go | 11 - main.go | 8 + 12 files changed, 263 insertions(+), 319 deletions(-) delete mode 100644 internal/cache/scheduler.go delete mode 100644 internal/cache/scheduler_test.go diff --git a/README.md b/README.md index 0685dc80..923830e9 100644 --- a/README.md +++ b/README.md @@ -155,7 +155,7 @@ Pick the configuration that matches your Plex server: | Mount | Description | | --- | --- | -| `/config` | Persistent cache (`profiles.json` learned language profiles, `tokens.json` encrypted shared-user tokens, `state.json` sync state). Mount a named volume or host path to preserve data across restarts. A corrupt file resets only its own section; a `cache.json` from an earlier version migrates automatically on first start. | +| `/config` | Persistent cache (`profiles.json` learned language profiles, `tokens.json` encrypted shared-user tokens, `state.json` sync state, `.plex-language-sync-last-run` completion record for the deep-scan safety net). Mount a named volume or host path to preserve data across restarts. A corrupt file resets only its own section; a `cache.json` from an earlier version migrates automatically on first start. | ## Graceful shutdown diff --git a/internal/cache/cache.go b/internal/cache/cache.go index 692a556b..dfa4d244 100644 --- a/internal/cache/cache.go +++ b/internal/cache/cache.go @@ -1,13 +1,12 @@ // Package cache is the on-disk persistence layer for processed-episode -// deduplication, per-user language profiles, shared-user tokens, and the -// scheduler's last-run marker. +// deduplication, per-user language profiles, and shared-user tokens. // // On-disk layout: state is split across three files in the cache dir by // retention class, so one corrupt file never costs another class's state: // -// profiles.json language_profiles irreplaceable learned state -// tokens.json user_tokens re-fetchable encrypted secrets -// state.json processed_episodes, last_scheduler_run disposable operational state +// profiles.json language_profiles irreplaceable learned state +// tokens.json user_tokens re-fetchable encrypted secrets +// state.json processed_episodes disposable operational state // // Field names, types, and JSON tags within each file are an inviolate // read-forward / write-back contract across deploys — any change is a @@ -55,8 +54,8 @@ const ( ) // Data is the in-memory state shape. It doubles as the decode target for -// the legacy pre-split cache.json union schema, whose field names and JSON -// tags it preserves verbatim (read-forward contract). +// the legacy pre-split cache.json union schema; a legacy key with no field +// here (last_scheduler_run) is ignored on decode. type Data struct { // ProcessedEpisodes tracks recently processed episode keys to avoid // re-processing the same episode on rapid successive events. @@ -77,9 +76,6 @@ type Data struct { // UserTokens maps userID → accessToken for shared users. Persisted in // tokens.json, encrypted at the disk boundary when a key is set. UserTokens map[string]string `json:"user_tokens"` - // LastSchedulerRun is the unix timestamp of the last scheduler run. - // Persisted in state.json. - LastSchedulerRun int64 `json:"last_scheduler_run"` } // profilesData is the profiles.json schema (irreplaceable learned state). @@ -96,7 +92,6 @@ type tokensData struct { // stateData is the state.json schema (disposable operational state). type stateData struct { ProcessedEpisodes map[string]int64 `json:"processed_episodes"` - LastSchedulerRun int64 `json:"last_scheduler_run"` } // Cache is the concurrent-safe persistent cache. The zero value is usable; @@ -242,7 +237,6 @@ func (c *Cache) applyLegacyLocked(legacy *Data) { c.data.UserTokens = legacy.UserTokens c.decryptTokensLocked() } - c.data.LastSchedulerRun = legacy.LastSchedulerRun } // overlayProfilesLocked loads profiles.json over the baseline. Returns 1 @@ -319,7 +313,6 @@ func (c *Cache) overlayStateLocked(dir string, errs *[]error) int { if c.data.ProcessedEpisodes == nil { c.data.ProcessedEpisodes = make(map[string]int64) } - c.data.LastSchedulerRun = sd.LastSchedulerRun return 1 } @@ -475,7 +468,6 @@ func (c *Cache) encodeAllForSave() (profiles, tokens, state []byte, err error) { } if state, err = json.MarshalIndent(&stateData{ ProcessedEpisodes: c.data.ProcessedEpisodes, - LastSchedulerRun: c.data.LastSchedulerRun, }, "", " "); err != nil { return nil, nil, nil, fmt.Errorf("marshal %s: %w", stateFile, err) } diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index ee1c6bda..603941d1 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -57,7 +57,6 @@ func TestCacheSaveLoadRoundTrip(t *testing.T) { "1": {"jpn": "eng", "eng": ""}, } orig.data.UserTokens = map[string]string{"2": "t2"} - orig.data.LastSchedulerRun = 1700000000 if err := orig.Save(dir); err != nil { t.Fatalf("Save() error = %v", err) @@ -83,8 +82,27 @@ func TestCacheSaveLoadRoundTrip(t *testing.T) { if loaded.data.UserTokens["2"] != "t2" { t.Errorf("UserTokens[2] = %q, want t2", loaded.data.UserTokens["2"]) } - if loaded.data.LastSchedulerRun != 1700000000 { - t.Errorf("LastSchedulerRun = %d, want 1700000000", loaded.data.LastSchedulerRun) +} + +// TestCacheLoadIgnoresRetiredMarkerKey pins the read-forward contract for +// files written by earlier versions: state.json (and the legacy cache.json) +// used to carry a last_scheduler_run key, and a file still holding it must +// load its live sections cleanly. Fails if the load path ever opts into +// strict decoding (DisallowUnknownFields). +func TestCacheLoadIgnoresRetiredMarkerKey(t *testing.T) { + t.Parallel() + dir := t.TempDir() + state := `{"processed_episodes":{"timeline:9":1700000000},"last_scheduler_run":1700000000}` + if err := os.WriteFile(filepath.Join(dir, stateFile), []byte(state), 0o600); err != nil { + t.Fatal(err) + } + + c := New() + if err := c.Load(dir); err != nil { + t.Errorf("Load() with a retired key in state.json = %v, want nil", err) + } + if len(c.data.ProcessedEpisodes) != 1 { + t.Errorf("ProcessedEpisodes = %v, want the seeded entry loaded beside the retired key", c.data.ProcessedEpisodes) } } @@ -211,7 +229,6 @@ func TestCacheLoadCorruptSectionIsolation(t *testing.T) { orig.data.ProcessedEpisodes = map[string]int64{"streams:1:100:1:2": time.Now().Unix()} orig.data.LanguageProfiles = map[string]map[string]string{"1": {"jpn": "eng"}} orig.data.UserTokens = map[string]string{"2": "t2"} - orig.data.LastSchedulerRun = 1700000000 if err := orig.Save(dir); err != nil { t.Fatalf("seed Save() error = %v", err) } @@ -242,8 +259,7 @@ func TestCacheLoadCorruptSectionIsolation(t *testing.T) { gotProfiles := loaded.data.LanguageProfiles["1"]["jpn"] == "eng" gotTokens := loaded.data.UserTokens["2"] == "t2" - gotState := loaded.data.LastSchedulerRun == 1700000000 && - len(loaded.data.ProcessedEpisodes) == 1 + gotState := len(loaded.data.ProcessedEpisodes) == 1 if want := tc.corrupt != profilesFile; gotProfiles != want { t.Errorf("profiles survived = %v, want %v", gotProfiles, want) @@ -283,7 +299,6 @@ func TestCacheLoadMigratesLegacyCacheJSON(t *testing.T) { ProcessedEpisodes: map[string]int64{"timeline:9": time.Now().Unix()}, LanguageProfiles: map[string]map[string]string{"1": {"jpn": "eng"}}, UserTokens: map[string]string{"2": "plain-tok"}, - LastSchedulerRun: 1700000000, }) key, err := DeriveKey("admin-token") @@ -297,7 +312,7 @@ func TestCacheLoadMigratesLegacyCacheJSON(t *testing.T) { } if c.data.LanguageProfiles["1"]["jpn"] != "eng" || c.data.UserTokens["2"] != "plain-tok" || - c.data.LastSchedulerRun != 1700000000 || len(c.data.ProcessedEpisodes) != 1 { + len(c.data.ProcessedEpisodes) != 1 { t.Errorf("migrated state incomplete: %+v", c.data) } for _, name := range splitFiles { @@ -343,14 +358,12 @@ func TestCacheLoadSplitWinsOverStaleLegacy(t *testing.T) { split := New() split.data.LanguageProfiles = map[string]map[string]string{"1": {"jpn": "eng"}} split.data.UserTokens = map[string]string{"2": "new-tok"} - split.data.LastSchedulerRun = 2000000000 if err := split.Save(dir); err != nil { t.Fatalf("Save() error = %v", err) } legacyWrite(t, dir, Data{ LanguageProfiles: map[string]map[string]string{"1": {"jpn": "STALE"}}, UserTokens: map[string]string{"2": "STALE"}, - LastSchedulerRun: 1, }) loaded := New() @@ -365,9 +378,6 @@ func TestCacheLoadSplitWinsOverStaleLegacy(t *testing.T) { t.Errorf("tokens = %q, want the split file's value to win over stale legacy", loaded.data.UserTokens["2"]) } - if loaded.data.LastSchedulerRun != 2000000000 { - t.Errorf("LastSchedulerRun = %d, want the split file's value", loaded.data.LastSchedulerRun) - } if _, err := os.Stat(filepath.Join(dir, legacyCacheFile)); !os.IsNotExist(err) { t.Errorf("stale legacy cache.json not removed (stat err = %v)", err) } @@ -383,7 +393,6 @@ func TestCacheLoadLegacyFillsMissingSection(t *testing.T) { legacyWrite(t, dir, Data{ LanguageProfiles: map[string]map[string]string{"1": {"jpn": "LEGACY"}}, UserTokens: map[string]string{"2": "legacy-tok"}, - LastSchedulerRun: 1700000000, }) pd, err := json.Marshal(&profilesData{ LanguageProfiles: map[string]map[string]string{"1": {"jpn": "SPLIT"}}, @@ -405,9 +414,6 @@ func TestCacheLoadLegacyFillsMissingSection(t *testing.T) { if loaded.data.UserTokens["2"] != "legacy-tok" { t.Errorf("tokens = %q, want the legacy value for the missing section", loaded.data.UserTokens["2"]) } - if loaded.data.LastSchedulerRun != 1700000000 { - t.Errorf("LastSchedulerRun = %d, want the legacy value", loaded.data.LastSchedulerRun) - } if _, err := os.Stat(filepath.Join(dir, legacyCacheFile)); !os.IsNotExist(err) { t.Errorf("legacy cache.json not removed after completing migration (stat err = %v)", err) } @@ -535,7 +541,7 @@ func TestCacheLoadLeavesAuthoritativeSplitFilesUntouched(t *testing.T) { LanguageProfiles: map[string]map[string]string{"1": {"jpn": "eng"}}, }), tokensFile: mustCompactJSON(t, &tokensData{UserTokens: map[string]string{"2": "new-tok"}}), - stateFile: mustCompactJSON(t, &stateData{LastSchedulerRun: 2000000000}), + stateFile: mustCompactJSON(t, &stateData{ProcessedEpisodes: map[string]int64{"timeline:9": 2000000000}}), } for name, raw := range written { if err := os.WriteFile(filepath.Join(dir, name), raw, 0o600); err != nil { @@ -581,7 +587,7 @@ func mustCompactJSON(t *testing.T, v any) []byte { return raw } -// --- PBT: JSON round-trip preserves LastSchedulerRun + map lengths --- +// --- PBT: JSON round-trip preserves the map lengths --- func TestCacheDataJSONRoundTrip(t *testing.T) { rapid.Check(t, func(t *rapid.T) { @@ -596,7 +602,6 @@ func TestCacheDataJSONRoundTrip(t *testing.T) { ProcessedEpisodes: processed, LanguageProfiles: make(map[string]map[string]string), UserTokens: make(map[string]string), - LastSchedulerRun: int64(rapid.IntRange(0, 2000000000).Draw(t, "last_run")), } data, err := json.Marshal(&original) @@ -613,10 +618,6 @@ func TestCacheDataJSONRoundTrip(t *testing.T) { t.Errorf("ProcessedEpisodes length: got %d, want %d", len(decoded.ProcessedEpisodes), len(original.ProcessedEpisodes)) } - if decoded.LastSchedulerRun != original.LastSchedulerRun { - t.Errorf("LastSchedulerRun: got %d, want %d", - decoded.LastSchedulerRun, original.LastSchedulerRun) - } }) } diff --git a/internal/cache/contract.go b/internal/cache/contract.go index 3054822f..615e33ce 100644 --- a/internal/cache/contract.go +++ b/internal/cache/contract.go @@ -3,7 +3,6 @@ package cache import ( "sync" "testing" - "time" "github.com/cplieger/plex-language-sync/internal/streams" ) @@ -24,8 +23,6 @@ type Contract interface { IntentFor(userID, showKey string) (streams.Intent, bool) UserTokens() map[string]string SetUserTokens(tokens map[string]string) - LastSchedulerRun() time.Time - SetLastSchedulerRun(t time.Time) } // RunContract exercises the persisted-cache contract against any @@ -80,36 +77,9 @@ func RunContract(t *testing.T, c Contract) { wg.Wait() }) - schedulerRunContract(t, c) checkAndMarkContract(t, c) } -// schedulerRunContract exercises the scheduler-run portion of the Contract -// contract: the last-scheduler-run marker round-trips a whole-second value and -// resets to the zero time. Split out of RunCacheContract to keep that -// function's cognitive complexity under the gate. -func schedulerRunContract(t *testing.T, c Contract) { - t.Helper() - - t.Run("scheduler_run_roundtrip", func(t *testing.T) { - // Whole-second value: internal/cache persists the marker as a unix - // int64 (time.Unix truncation), so the shared contract is pinned at - // second granularity that both implementations honour. - want := time.Unix(1700000000, 0) - c.SetLastSchedulerRun(want) - if got := c.LastSchedulerRun(); !got.Equal(want) { - t.Errorf("LastSchedulerRun = %v, want %v", got, want) - } - }) - - t.Run("scheduler_run_zero", func(t *testing.T) { - c.SetLastSchedulerRun(time.Time{}) - if got := c.LastSchedulerRun(); !got.IsZero() { - t.Errorf("LastSchedulerRun after zero set = %v, want zero", got) - } - }) -} - // Language-code literals shared by the contract subtests. const ( langJPN = "jpn" diff --git a/internal/cache/scheduler.go b/internal/cache/scheduler.go deleted file mode 100644 index 577c331b..00000000 --- a/internal/cache/scheduler.go +++ /dev/null @@ -1,28 +0,0 @@ -package cache - -import "time" - -// LastSchedulerRun returns the timestamp of the last deep-analysis run. -// A zero time.Time indicates the scheduler has never run (fresh install -// or cache reset). -func (c *Cache) LastSchedulerRun() time.Time { - c.mu.Lock() - defer c.mu.Unlock() - if c.data.LastSchedulerRun == 0 { - return time.Time{} - } - return time.Unix(c.data.LastSchedulerRun, 0) -} - -// SetLastSchedulerRun records the supplied timestamp as the most recent -// scheduler run. Stored as a unix int64 on disk (contract item 7 — -// persisted schema is frozen). -func (c *Cache) SetLastSchedulerRun(t time.Time) { - c.mu.Lock() - defer c.mu.Unlock() - if t.IsZero() { - c.data.LastSchedulerRun = 0 - return - } - c.data.LastSchedulerRun = t.Unix() -} diff --git a/internal/cache/scheduler_test.go b/internal/cache/scheduler_test.go deleted file mode 100644 index 8c1ec06b..00000000 --- a/internal/cache/scheduler_test.go +++ /dev/null @@ -1,34 +0,0 @@ -package cache - -import ( - "testing" - "time" -) - -func TestLastSchedulerRunZeroReturnsZeroTime(t *testing.T) { - t.Parallel() - c := New() - if got := c.LastSchedulerRun(); !got.IsZero() { - t.Errorf("LastSchedulerRun() on fresh cache = %v, want zero time", got) - } -} - -func TestSetLastSchedulerRunRoundTrips(t *testing.T) { - t.Parallel() - c := New() - want := time.Unix(1700000000, 0) - c.SetLastSchedulerRun(want) - if got := c.LastSchedulerRun(); !got.Equal(want) { - t.Errorf("LastSchedulerRun() = %v, want %v", got, want) - } -} - -func TestSetLastSchedulerRunZeroClears(t *testing.T) { - t.Parallel() - c := New() - c.SetLastSchedulerRun(time.Unix(1700000000, 0)) - c.SetLastSchedulerRun(time.Time{}) - if got := c.LastSchedulerRun(); !got.IsZero() { - t.Errorf("LastSchedulerRun() after zero set = %v, want zero", got) - } -} diff --git a/internal/deepscan/deepscan.go b/internal/deepscan/deepscan.go index ce92fdc9..37bf7e1a 100644 --- a/internal/deepscan/deepscan.go +++ b/internal/deepscan/deepscan.go @@ -16,11 +16,12 @@ // - Fan out per-item work across a bounded worker pool // with a circuit breaker that aborts the // pass after a threshold of consecutive per-item failures. -// - Persist the last-run marker through runLedger so a cold restart -// does not double-run the analysis. +// - Persist the last-run record in a scheduler.Stamp file on the +// persistent volume so a cold restart does not double-run the +// analysis. // // Stable contracts preserved (keep these exact: Loki alerts grep the log -// strings and the on-disk cache schema depends on the field names): +// strings): // - WARN slog keys ("scheduler: aborting history processing after // consecutive failures", "scheduler: failed to fetch history", // "scheduler: failed to fetch sections", "scheduler: deep analysis @@ -29,10 +30,6 @@ // starting", "running initial deep analysis", "deep analysis // completed", "scheduler: processing recently added episode", // "scheduler stopped") identical. -// - On-disk cache schema unchanged (LastSchedulerRun lives in -// state.json since the 2026-07 retention split) — reads and writes go -// through runLedger and are tagged by the concrete internal/cache -// package. // // Consumer note: every collaborator is an interface THIS package declares (see // deps.go) — plexReader, EpisodeReader, runLedger, skipChecker and Syncer, each @@ -101,6 +98,7 @@ type Scheduler struct { plex plexReader cache runLedger sync Syncer + stamp *scheduler.Stamp dedup singleflight.Group userClient UserClientFunc saveCache CacheSaver @@ -132,8 +130,11 @@ func (s *Scheduler) workerCount() int { type Deps struct { // Plex reads history and library metadata for the sweep. Plex plexReader - // Cache holds the run marker and the recorded intents the pass re-applies. + // Cache holds the dedup keys and the recorded intents the pass re-applies. Cache runLedger + // Stamp is the persisted last-run record: it gates the startup pass and + // anchors the replay look-back window. + Stamp *scheduler.Stamp // UserClient returns the per-user write client for a username. UserClient UserClientFunc // Sync applies the recorded intent to an episode. @@ -149,6 +150,7 @@ func New(cfg Config, deps Deps) *Scheduler { cfg: cfg, plex: deps.Plex, cache: deps.Cache, + stamp: deps.Stamp, userClient: deps.UserClient, sync: deps.Sync, saveCache: deps.SaveCache, @@ -156,10 +158,10 @@ func New(cfg Config, deps Deps) *Scheduler { } // Run is the outer scheduler loop: it runs a deep-analysis pass at -// startup when the cache's LastSchedulerRun marker is absent or older -// than one Interval, then runs one every Interval via a time.Ticker. -// Returns when the context is cancelled. A disabled scheduler -// (Enable=false or Interval<=0) returns immediately. +// startup when the last-run stamp is absent or older than one Interval, +// then runs one every Interval via a time.Ticker. Returns when the +// context is cancelled. A disabled scheduler (Enable=false or +// Interval<=0) returns immediately. func (s *Scheduler) Run(ctx context.Context) { defer slog.Info("scheduler stopped") @@ -173,9 +175,9 @@ func (s *Scheduler) Run(ctx context.Context) { // Run immediately when never run before or the last run is older than // one interval, so a container restarting more often than the interval // is never starved of a safety-net pass, while one that ran recently - // does not double-run on restart. - lastRun := s.cache.LastSchedulerRun() - if lastRun.IsZero() || time.Since(lastRun) > s.cfg.Interval { + // does not double-run on restart. CountFailed because only complete + // passes are recorded and the ticker owns the retry. + if s.stamp.Due(s.cfg.Interval, time.Now(), scheduler.CountFailed) { slog.Info("running initial deep analysis") s.deepAnalysis(ctx) } @@ -183,7 +185,7 @@ func (s *Scheduler) Run(ctx context.Context) { // Fixed-interval scheduling via scheduler.RunLoop (the fleet // docker-*-scheduler convention). FireOnStart is false: the conditional // startup pass above already handled the immediate run (RunLoop's - // unconditional FireOnStart would ignore the last-run marker and double-run + // unconditional FireOnStart would ignore the last-run stamp and double-run // on a recent restart). Overlapping ticks collapse via the singleflight in // deepAnalysis, RunLoop is sequential, so no wall-clock slot-dedup is needed // and no local wall-clock time is read. @@ -226,22 +228,18 @@ func (s *Scheduler) deepAnalysis(ctx context.Context) { func (s *Scheduler) deepAnalysisCore(ctx context.Context) { complete := false defer func() { - // Advance the run marker ONLY when this pass fully covered its - // look-back window. History and RecentlyAdded are swept - // newest-first, so ANY early exit — graceful-shutdown cancellation, - // the history circuit breaker tripping, a fetch/overflow error, or a - // per-section fetch failure — leaves the OLDER end of the window - // unprocessed. Advancing the marker then would move the next run's - // look-back origin past those unswept events and drop them - // permanently (the WebSocket listener is usually also down during the - // Plex degradation that triggers these exits). Leaving the marker put - // makes the next run re-sweep the same window; the sweep is - // idempotent (recently-added is dedup-guarded, history re-applies the - // same selection), so the re-work is harmless. This uniform - // completeness gate subsumes the earlier ctx-cancel-only guard and - // closes the breaker-abort and overflow variants of the same bug. + // Record the run ONLY when this pass fully covered its look-back + // window. The sweeps run newest-first, so any early exit (cancel, + // breaker abort, fetch/overflow error, per-section failure) leaves + // the OLDER end unprocessed; recording then would move the next + // run's look-back origin past those unswept events and drop them + // permanently. An unrecorded pass re-sweeps the same window, which + // is idempotent. The outcome bit is informational: the startup gate + // reads the record under CountFailed, which considers age alone. if complete { - s.cache.SetLastSchedulerRun(time.Now()) + if err := s.stamp.Record(true); err != nil { + slog.Warn("last-run stamp write failed", "error", err) + } } if s.saveCache != nil { if err := s.saveCache(); err != nil { @@ -252,12 +250,12 @@ func (s *Scheduler) deepAnalysisCore(ctx context.Context) { // Look back to the previous completed run so no window is missed when // DEEP_SCAN_INTERVAL exceeds 24h; floor at 24h so a frequent interval - // (or a zero last-run marker on first boot) still replays a full - // recent day. LastSchedulerRun() reads the PREVIOUS run's timestamp - // here because this run's marker is only written in the deferred - // SetLastSchedulerRun above, after the body completes. + // (or no record on first boot) still replays a full recent day. The + // stamp still holds the PREVIOUS run here because this run is only + // recorded in the deferred Record above, after the body completes. lookback := 24 * time.Hour - if last := s.cache.LastSchedulerRun(); !last.IsZero() { + rec, _ := s.stamp.Last() + if last := rec.Time; !last.IsZero() { if since := time.Since(last); since > lookback { lookback = since } diff --git a/internal/deepscan/deepscan_test.go b/internal/deepscan/deepscan_test.go index f7d860da..a28686a0 100644 --- a/internal/deepscan/deepscan_test.go +++ b/internal/deepscan/deepscan_test.go @@ -5,6 +5,8 @@ import ( "context" "errors" "log/slog" + "os" + "path/filepath" "runtime" "strconv" "strings" @@ -17,12 +19,40 @@ import ( "github.com/cplieger/plex-language-sync/internal/plex" "github.com/cplieger/plex-language-sync/internal/streams" "github.com/cplieger/plex-language-sync/internal/testsupport/fakeapi" + "github.com/cplieger/scheduler/v4" ) // --------------------------------------------------------------------------- // Fakes // --------------------------------------------------------------------------- +// testStamp returns a Stamp backed by a fresh temp-dir file with no record: +// the "never run" state. +func testStamp(t *testing.T) *scheduler.Stamp { + t.Helper() + return scheduler.NewStamp(filepath.Join(t.TempDir(), "last-run")) +} + +// stampAt returns a Stamp seeded with a completed run dated at. Record always +// stamps the current time, so backdating writes the file format directly. +func stampAt(t *testing.T, at time.Time) *scheduler.Stamp { + t.Helper() + path := filepath.Join(t.TempDir(), "last-run") + line := at.UTC().Format(time.RFC3339Nano) + " ok\n" + if err := os.WriteFile(path, []byte(line), 0o600); err != nil { + t.Fatalf("seed stamp: %v", err) + } + return scheduler.NewStamp(path) +} + +// stampTime returns the recorded run time, or the zero time when no run was +// ever recorded — the same collapse the production look-back read uses. +func stampTime(t *testing.T, st *scheduler.Stamp) time.Time { + t.Helper() + rec, _ := st.Last() + return rec.Time +} + type fakeSyncer struct { changeCalls atomic.Int64 processCalls atomic.Int64 @@ -444,26 +474,27 @@ func TestProcessRecentlyAdded_HonorsIgnoreLibraries(t *testing.T) { } // --------------------------------------------------------------------------- -// deepAnalysisCore — persistence (last-run marker + cache flush) +// deepAnalysisCore — persistence (last-run stamp + cache flush) // --------------------------------------------------------------------------- // TestDeepAnalysisCore_SetsLastRunAndFlushesCache pins the deferred persistence in -// deepAnalysisCore: it records the last-run marker (the documented cold-restart +// deepAnalysisCore: it records the last-run stamp (the documented cold-restart // idempotency guard) and flushes the cache exactly once via saveCache. A nil // saveCache must be a no-op, not a panic. func TestDeepAnalysisCore_SetsLastRunAndFlushesCache(t *testing.T) { t.Parallel() plx := &fakeapi.Plex{} - c := fakeapi.NewCache() - if !c.LastSchedulerRun().IsZero() { - t.Fatal("precondition: fresh cache must report zero last-run") + st := testStamp(t) + if _, known := st.Last(); known { + t.Fatal("precondition: fresh stamp must report no recorded run") } var saveCalls atomic.Int64 sched := New( Config{Enable: true}, Deps{ Plex: plx, - Cache: c, + Cache: fakeapi.NewCache(), + Stamp: st, UserClient: func(_ string) EpisodeReader { return plx }, Sync: &fakeSyncer{}, SaveCache: func() error { saveCalls.Add(1); return nil }, @@ -472,8 +503,8 @@ func TestDeepAnalysisCore_SetsLastRunAndFlushesCache(t *testing.T) { sched.deepAnalysisCore(t.Context()) - if c.LastSchedulerRun().IsZero() { - t.Error("deepAnalysisCore did not record the last-run marker") + if _, known := st.Last(); !known { + t.Error("deepAnalysisCore did not record the last-run stamp") } if got := saveCalls.Load(); got != 1 { t.Errorf("saveCache called %d times; want 1", got) @@ -485,6 +516,7 @@ func TestDeepAnalysisCore_SetsLastRunAndFlushesCache(t *testing.T) { Deps{ Plex: plx, Cache: fakeapi.NewCache(), + Stamp: testStamp(t), UserClient: func(_ string) EpisodeReader { return plx }, Sync: &fakeSyncer{}, SaveCache: nil, @@ -507,6 +539,7 @@ func TestRun_DisabledReturnsImmediately(t *testing.T) { Deps{ Plex: plx, Cache: fakeapi.NewCache(), + Stamp: testStamp(t), UserClient: func(_ string) EpisodeReader { return plx }, Sync: &fakeSyncer{}, SaveCache: nil, @@ -541,6 +574,7 @@ func TestRun_NonPositiveIntervalIsDisabled(t *testing.T) { Deps{ Plex: plx, Cache: fakeapi.NewCache(), + Stamp: testStamp(t), UserClient: func(_ string) EpisodeReader { return plx }, Sync: &fakeSyncer{}, SaveCache: nil, @@ -556,24 +590,24 @@ func TestRun_NonPositiveIntervalIsDisabled(t *testing.T) { } // TestRun_RunsInitialAnalysisWhenNeverRun verifies the "run immediately when the -// last-run marker is absent" branch. A pre-cancelled context makes the scheduling +// last-run record is absent" branch. A pre-cancelled context makes the scheduling // loop return as soon as the initial pass is dispatched, so the timer wait // (timing-bound, intentionally untested) is never entered. Because the initial // pass therefore runs under an already-cancelled ctx, the h-f1 watermark-on-cancel -// guard means it does NOT advance the last-run marker; the proof that the initial +// guard means it does NOT record the run; the proof that the initial // branch fired is the Plex History + ShowSections queries below (fetched at the -// top of the pass before any ctx short-circuit). Marker advancement on a COMPLETED +// top of the pass before any ctx short-circuit). Recording on a COMPLETED // pass is pinned separately by // TestDeepAnalysisCore_CancelledPassLeavesWatermarkUnchanged. func TestRun_RunsInitialAnalysisWhenNeverRun(t *testing.T) { t.Parallel() plx := &fakeapi.Plex{} - c := fakeapi.NewCache() // zero last-run -> initial pass should fire sched := New( Config{Enable: true, Interval: 24 * time.Hour}, Deps{ Plex: plx, - Cache: c, + Cache: fakeapi.NewCache(), + Stamp: testStamp(t), // no record -> initial pass should fire UserClient: func(_ string) EpisodeReader { return plx }, Sync: &fakeSyncer{}, SaveCache: nil, @@ -598,29 +632,31 @@ func TestRun_RunsInitialAnalysisWhenNeverRun(t *testing.T) { } } -// TestRun_InitialPassDecisionFromMarker pins Run's initial-pass gate -// (lastRun.IsZero() || time.Since(lastRun) > s.cfg.Interval). The existing Run -// tests cover only the disabled short-circuit and the zero-marker arm; neither -// sets a non-zero marker, so the documented cold-restart idempotency guard (a -// marker newer than one interval must NOT double-run the analysis on restart) -// and its stale-marker catch-up companion have no coverage -- and are not -// covered transitively, since the deepAnalysisCore-level tests call -// deepAnalysisCore directly and bypass this gate. A pre-cancelled context makes -// Run return on ctx.Done() right after the gate, so the timing-bound ticker -// loop is never entered (the technique TestRun_RunsInitialAnalysisWhenNeverRun -// uses). -func TestRun_InitialPassDecisionFromMarker(t *testing.T) { +// TestRun_InitialPassDecisionFromStamp pins Run's initial-pass gate +// (stamp.Due(Interval, now, CountFailed)). The other Run tests cover only the +// disabled short-circuit and the no-record arm; neither stages an existing +// record, so the documented cold-restart idempotency guard (a record newer +// than one interval must NOT double-run the analysis on restart) and its +// stale-record catch-up companion have no coverage -- and are not covered +// transitively, since the deepAnalysisCore-level tests call deepAnalysisCore +// directly and bypass this gate. A pre-cancelled context makes Run return on +// ctx.Done() right after the gate, so the timing-bound ticker loop is never +// entered (the technique TestRun_RunsInitialAnalysisWhenNeverRun uses). +func TestRun_InitialPassDecisionFromStamp(t *testing.T) { t.Parallel() - t.Run("recent marker skips the initial pass", func(t *testing.T) { + t.Run("recent record skips the initial pass", func(t *testing.T) { t.Parallel() plx := &fakeapi.Plex{} - c := fakeapi.NewCache() - c.SetLastSchedulerRun(time.Now()) // ran within the last interval + st := testStamp(t) + if err := st.Record(true); err != nil { // ran within the last interval + t.Fatalf("seed stamp: %v", err) + } sched := New( Config{Enable: true, Interval: 24 * time.Hour}, Deps{ Plex: plx, - Cache: c, + Cache: fakeapi.NewCache(), + Stamp: st, UserClient: func(_ string) EpisodeReader { return plx }, Sync: &fakeSyncer{}, SaveCache: nil, @@ -630,19 +666,19 @@ func TestRun_InitialPassDecisionFromMarker(t *testing.T) { cancel() // pre-cancelled on purpose; Background, not t.Context() sched.Run(ctx) if got := plx.Calls.Load(); got != 0 { - t.Errorf("Run made %d Plex calls with a recent last-run marker; want 0 (the cold-restart guard must skip the initial pass, not re-run a full sweep)", got) + t.Errorf("Run made %d Plex calls with a recent last-run record; want 0 (the cold-restart guard must skip the initial pass, not re-run a full sweep)", got) } }) - t.Run("stale marker runs a catch-up pass", func(t *testing.T) { + t.Run("stale record runs a catch-up pass", func(t *testing.T) { t.Parallel() plx := &fakeapi.Plex{} - c := fakeapi.NewCache() - c.SetLastSchedulerRun(time.Now().Add(-72 * time.Hour)) // older than the 24h interval + st := stampAt(t, time.Now().Add(-72*time.Hour)) // older than the 24h interval sched := New( Config{Enable: true, Interval: 24 * time.Hour}, Deps{ Plex: plx, - Cache: c, + Cache: fakeapi.NewCache(), + Stamp: st, UserClient: func(_ string) EpisodeReader { return plx }, Sync: &fakeSyncer{}, SaveCache: nil, @@ -662,7 +698,7 @@ func TestRun_InitialPassDecisionFromMarker(t *testing.T) { } } if !sawHistory || !sawSections { - t.Errorf("stale marker did not trigger a catch-up pass (calls=%v); want History and ShowSections", names) + t.Errorf("stale record did not trigger a catch-up pass (calls=%v); want History and ShowSections", names) } }) } @@ -902,6 +938,7 @@ func TestDeepAnalysis_ConcurrentCallCollapsesAndWarnsOnce(t *testing.T) { Deps{ Plex: plx, Cache: fakeapi.NewCache(), + Stamp: testStamp(t), UserClient: func(_ string) EpisodeReader { return plx.Plex }, Sync: &fakeSyncer{}, SaveCache: nil, @@ -1047,18 +1084,19 @@ func TestProcessHistoryItem_NilPerUserClientSkips(t *testing.T) { } // TestDeepAnalysisCore_SaveCacheErrorWarns pins the deferred cache-flush error -// branch: a failing saveCache still records the last-run marker and logs the +// branch: a failing saveCache still records the last-run stamp and logs the // "cache save failed" WARN rather than swallowing the error. Companion to // TestDeepAnalysisCore_SetsLastRunAndFlushesCache. Not parallel: captureSlog // mutates the global logger. func TestDeepAnalysisCore_SaveCacheErrorWarns(t *testing.T) { plx := &fakeapi.Plex{} - c := fakeapi.NewCache() + st := testStamp(t) sched := New( Config{Enable: true}, Deps{ Plex: plx, - Cache: c, + Cache: fakeapi.NewCache(), + Stamp: st, UserClient: func(_ string) EpisodeReader { return plx }, Sync: &fakeSyncer{}, SaveCache: func() error { return errors.New("disk full") }, @@ -1070,8 +1108,42 @@ func TestDeepAnalysisCore_SaveCacheErrorWarns(t *testing.T) { if !strings.Contains(out, "cache save failed") { t.Errorf("missing cache-save-failure WARN when saveCache errors; log: %q", out) } - if c.LastSchedulerRun().IsZero() { - t.Error("last-run marker must still be recorded even when the cache flush fails") + if _, known := st.Last(); !known { + t.Error("last-run stamp must still be recorded even when the cache flush fails") + } +} + +// TestDeepAnalysisCore_StampRecordFailureWarnsOnly pins the deferred stamp +// write's error branch: a Record that cannot write (its directory is gone) +// logs a WARN and never fails the pass — the completion line still logs and +// the cache flush still runs. Not parallel: captureSlog mutates the global +// logger. +func TestDeepAnalysisCore_StampRecordFailureWarnsOnly(t *testing.T) { + plx := &fakeapi.Plex{} + st := scheduler.NewStamp(filepath.Join(t.TempDir(), "missing-dir", "last-run")) + var saveCalls atomic.Int64 + sched := New( + Config{Enable: true}, + Deps{ + Plex: plx, + Cache: fakeapi.NewCache(), + Stamp: st, + UserClient: func(_ string) EpisodeReader { return plx }, + Sync: &fakeSyncer{}, + SaveCache: func() error { saveCalls.Add(1); return nil }, + }, + ) + out := captureSlog(t, func() { + sched.deepAnalysisCore(t.Context()) + }) + if !strings.Contains(out, "last-run stamp write failed") { + t.Errorf("missing stamp-write-failure WARN; log: %q", out) + } + if !strings.Contains(out, "deep analysis completed") { + t.Errorf("pass did not run to completion despite the stamp write failing; log: %q", out) + } + if got := saveCalls.Load(); got != 1 { + t.Errorf("saveCache called %d times after a failed stamp write; want 1 (the flush must still run)", got) } } @@ -1288,22 +1360,21 @@ func (p *sinceCapturePlex) History(ctx context.Context, since int64) ([]plex.His // > 24h, or a restart after a long downtime), the replay look-back extends to // the full since-last-run gap instead of the fixed 24h floor, so the span // between 24h and the interval is not silently skipped by the safety net. -// deepAnalysisCore reads LastSchedulerRun() (the PREVIOUS run -- its own marker -// is only written in the deferred SetLastSchedulerRun after the body), extends -// lookback to max(24h, time.Since(last)), and feeds the resulting sinceUnix to -// History. The two existing deepAnalysisCore tests use a fresh (zero last-run) -// cache, so they exercise only the 24h floor; the extend branch and the -// resulting window value were unasserted. +// deepAnalysisCore reads stamp.Last() (the PREVIOUS run -- this run is only +// recorded in the deferred Record after the body), extends lookback to +// max(24h, time.Since(last)), and feeds the resulting sinceUnix to History. +// The two older deepAnalysisCore tests use a fresh (no record) stamp, so they +// exercise only the 24h floor; the extend branch and the resulting window +// value were unasserted. func TestDeepAnalysisCore_ExtendsLookbackBeyond24hFromLastRun(t *testing.T) { t.Parallel() plx := &sinceCapturePlex{Plex: &fakeapi.Plex{}} - c := fakeapi.NewCache() - c.SetLastSchedulerRun(time.Now().Add(-72 * time.Hour)) // previous run 72h ago sched := New( Config{Enable: true}, Deps{ Plex: plx, - Cache: c, + Cache: fakeapi.NewCache(), + Stamp: stampAt(t, time.Now().Add(-72*time.Hour)), // previous run 72h ago UserClient: func(_ string) EpisodeReader { return plx.Plex }, Sync: &fakeSyncer{}, SaveCache: nil, @@ -1330,19 +1401,19 @@ func TestDeepAnalysisCore_ExtendsLookbackBeyond24hFromLastRun(t *testing.T) { // TestDeepAnalysisCore_LookbackFloorIsOneDayOnFirstRun pins the 24h floor at // the other end of the same window calculation: with no previous run recorded — -// first boot, or a state.json that was reset — the replay still covers a full +// first boot, or a deleted record file — the replay still covers a full // recent day. This is the arm the two older deepAnalysisCore tests exercise but // never assert: a floor that collapsed would leave the safety-net pass fetching // an empty window and repairing nothing, silently, forever. func TestDeepAnalysisCore_LookbackFloorIsOneDayOnFirstRun(t *testing.T) { t.Parallel() plx := &sinceCapturePlex{Plex: &fakeapi.Plex{}} - c := fakeapi.NewCache() // zero last-run marker -> the floor decides the window sched := New( Config{Enable: true}, Deps{ Plex: plx, - Cache: c, + Cache: fakeapi.NewCache(), + Stamp: testStamp(t), // no record -> the floor decides the window UserClient: func(_ string) EpisodeReader { return plx.Plex }, Sync: &fakeSyncer{}, SaveCache: nil, @@ -1369,32 +1440,31 @@ func TestDeepAnalysisCore_LookbackFloorIsOneDayOnFirstRun(t *testing.T) { // --------------------------------------------------------------------------- // TestDeepAnalysisCore_CancelledPassLeavesWatermarkUnchanged pins the h-f1 -// watermark-on-cancel guard: deepAnalysisCore's deferred SetLastSchedulerRun -// advances the last-run marker ONLY when the pass completed (ctx.Err()==nil). -// A pass cancelled mid-flight (graceful shutdown) fetches history and -// recently-added newest-first and may leave the OLDER end of its window -// unprocessed, so advancing the marker would make the next run's dynamic -// look-back (l-f11) start past those unprocessed events -- permanently -// skipping them. The marker must therefore stay at the previous completed -// run's value on a cancelled pass and advance on a completed one. The -// saveCache flush stays UNGUARDED (persisting partial learning is harmless; -// main.go re-saves on shutdown), which the cancelled subtest's saveCalls -// check keeps pinned so a regression that wrongly guards the whole defer body -// is caught too. +// watermark-on-cancel guard: deepAnalysisCore's deferred stamp.Record runs +// ONLY when the pass completed (ctx.Err()==nil). A pass cancelled mid-flight +// (graceful shutdown) fetches history and recently-added newest-first and may +// leave the OLDER end of its window unprocessed, so recording it would make +// the next run's dynamic look-back (l-f11) start past those unprocessed +// events -- permanently skipping them. The record must therefore stay at the +// previous completed run's value on a cancelled pass and advance on a +// completed one. The saveCache flush stays UNGUARDED (persisting partial +// learning is harmless; main.go re-saves on shutdown), which the cancelled +// subtest's saveCalls check keeps pinned so a regression that wrongly guards +// the whole defer body is caught too. func TestDeepAnalysisCore_CancelledPassLeavesWatermarkUnchanged(t *testing.T) { t.Parallel() - t.Run("cancelled pass does not advance the marker", func(t *testing.T) { + t.Run("cancelled pass does not advance the record", func(t *testing.T) { t.Parallel() plx := &fakeapi.Plex{} - c := fakeapi.NewCache() - prev := time.Now().Add(-72 * time.Hour) // previous COMPLETED run's watermark - c.SetLastSchedulerRun(prev) + prev := time.Now().Add(-72 * time.Hour) // previous COMPLETED run's record + st := stampAt(t, prev) var saveCalls atomic.Int64 sched := New( Config{Enable: true}, Deps{ Plex: plx, - Cache: c, + Cache: fakeapi.NewCache(), + Stamp: st, UserClient: func(_ string) EpisodeReader { return plx }, Sync: &fakeSyncer{}, SaveCache: func() error { saveCalls.Add(1); return nil }, @@ -1405,8 +1475,8 @@ func TestDeepAnalysisCore_CancelledPassLeavesWatermarkUnchanged(t *testing.T) { sched.deepAnalysisCore(ctx) - if got := c.LastSchedulerRun(); !got.Equal(prev) { - t.Errorf("cancelled pass advanced the watermark to %v; want it unchanged at the previous completed run %v (advancing would skip the unprocessed older window)", + if got := stampTime(t, st); !got.Equal(prev) { + t.Errorf("cancelled pass advanced the record to %v; want it unchanged at the previous completed run %v (advancing would skip the unprocessed older window)", got.UTC(), prev.UTC()) } // saveCache stays UNGUARDED: a cancelled pass still flushes partial @@ -1416,17 +1486,17 @@ func TestDeepAnalysisCore_CancelledPassLeavesWatermarkUnchanged(t *testing.T) { t.Errorf("saveCache called %d times on a cancelled pass; want 1 (the flush is intentionally unguarded)", got) } }) - t.Run("completed pass advances the marker", func(t *testing.T) { + t.Run("completed pass advances the record", func(t *testing.T) { t.Parallel() plx := &fakeapi.Plex{} - c := fakeapi.NewCache() prev := time.Now().Add(-72 * time.Hour) - c.SetLastSchedulerRun(prev) + st := stampAt(t, prev) sched := New( Config{Enable: true}, Deps{ Plex: plx, - Cache: c, + Cache: fakeapi.NewCache(), + Stamp: st, UserClient: func(_ string) EpisodeReader { return plx }, Sync: &fakeSyncer{}, SaveCache: nil, @@ -1436,33 +1506,33 @@ func TestDeepAnalysisCore_CancelledPassLeavesWatermarkUnchanged(t *testing.T) { sched.deepAnalysisCore(t.Context()) - got := c.LastSchedulerRun() + got := stampTime(t, st) if !got.After(prev) { - t.Errorf("completed pass did not advance the watermark: got %v, still <= previous run %v", got.UTC(), prev.UTC()) + t.Errorf("completed pass did not advance the record: got %v, still <= previous run %v", got.UTC(), prev.UTC()) } if got.Before(before) { - t.Errorf("completed pass set the watermark to %v, before the run started %v", got.UTC(), before.UTC()) + t.Errorf("completed pass set the record to %v, before the run started %v", got.UTC(), before.UTC()) } }) } // TestDeepAnalysisCore_IncompletePassLeavesWatermarkUnchanged pins the uniform // completeness gate that generalises the h-f1 ctx-cancel guard: deepAnalysisCore -// advances the last-run marker ONLY when the pass fully covered its look-back -// window. Because History/RecentlyAdded are swept newest-first, ANY early exit -// leaves the OLDER end of the window unprocessed, so advancing the marker would -// make the next run's dynamic look-back start past those unswept events and drop -// them permanently. The three incomplete-pass triggers below (in addition to the -// ctx-cancel case, which has its own test) must all leave the marker unchanged: +// records the run ONLY when the pass fully covered its look-back window. +// Because History/RecentlyAdded are swept newest-first, ANY early exit leaves +// the OLDER end of the window unprocessed, so recording it would make the next +// run's dynamic look-back start past those unswept events and drop them +// permanently. The three incomplete-pass triggers below (in addition to the +// ctx-cancel case, which has its own test) must all leave the record unchanged: // a history circuit-breaker abort, a History fetch/overflow error, and a // recently-added per-section fetch failure. func TestDeepAnalysisCore_IncompletePassLeavesWatermarkUnchanged(t *testing.T) { t.Parallel() - // prev is the previous COMPLETED run's marker; every subtest asserts it is + // prev is the previous COMPLETED run's record; every subtest asserts it is // left untouched by an incomplete pass. - newSched := func(plx plexReader, reader func(string) EpisodeReader, c runLedger) *Scheduler { - return New(Config{Enable: true}, Deps{Plex: plx, Cache: c, UserClient: reader, Sync: &fakeSyncer{}}) + newSched := func(plx plexReader, reader func(string) EpisodeReader, st *scheduler.Stamp) *Scheduler { + return New(Config{Enable: true}, Deps{Plex: plx, Cache: fakeapi.NewCache(), Stamp: st, UserClient: reader, Sync: &fakeSyncer{}}) } t.Run("history circuit-breaker abort", func(t *testing.T) { @@ -1475,32 +1545,30 @@ func TestDeepAnalysisCore_IncompletePassLeavesWatermarkUnchanged(t *testing.T) { items[i] = plex.HistoryItem{AccountID: 1, RatingKey: strconv.Itoa(1000 + i), Type: plex.TypeEpisode} } plx := &fakeapi.Plex{HistoryItems: items, EpisodeErr: errors.New("fetch boom")} - c := fakeapi.NewCache() prev := time.Now().Add(-72 * time.Hour) - c.SetLastSchedulerRun(prev) - sched := newSched(plx, func(_ string) EpisodeReader { return plx }, c) + st := stampAt(t, prev) + sched := newSched(plx, func(_ string) EpisodeReader { return plx }, st) sched.deepAnalysisCore(t.Context()) - if got := c.LastSchedulerRun(); !got.Equal(prev) { - t.Errorf("breaker-abort pass advanced the marker to %v; want unchanged at %v (older window unswept)", got.UTC(), prev.UTC()) + if got := stampTime(t, st); !got.Equal(prev) { + t.Errorf("breaker-abort pass advanced the record to %v; want unchanged at %v (older window unswept)", got.UTC(), prev.UTC()) } }) t.Run("history fetch/overflow error", func(t *testing.T) { t.Parallel() // A History error models the 10MB-overflow case (errBodyOverCap): zero - // items replayed, so the marker must not advance. + // items replayed, so the record must not advance. plx := &fetchErrPlex{Plex: &fakeapi.Plex{}, historyErr: errors.New("body over cap")} - c := fakeapi.NewCache() prev := time.Now().Add(-72 * time.Hour) - c.SetLastSchedulerRun(prev) - sched := newSched(plx, func(_ string) EpisodeReader { return plx.Plex }, c) + st := stampAt(t, prev) + sched := newSched(plx, func(_ string) EpisodeReader { return plx.Plex }, st) sched.deepAnalysisCore(t.Context()) - if got := c.LastSchedulerRun(); !got.Equal(prev) { - t.Errorf("history-error pass advanced the marker to %v; want unchanged at %v", got.UTC(), prev.UTC()) + if got := stampTime(t, st); !got.Equal(prev) { + t.Errorf("history-error pass advanced the record to %v; want unchanged at %v", got.UTC(), prev.UTC()) } }) @@ -1510,34 +1578,32 @@ func TestDeepAnalysisCore_IncompletePassLeavesWatermarkUnchanged(t *testing.T) { // fetch, leaving that section's window unswept → pass incomplete. base := &fakeapi.Plex{Sections: []plex.Section{{Key: "1", Title: "TV"}}} plx := &recentlyAddedErrPlex{Plex: base, failSections: map[string]bool{"1": true}} - c := fakeapi.NewCache() prev := time.Now().Add(-72 * time.Hour) - c.SetLastSchedulerRun(prev) - sched := newSched(plx, func(_ string) EpisodeReader { return plx.Plex }, c) + st := stampAt(t, prev) + sched := newSched(plx, func(_ string) EpisodeReader { return plx.Plex }, st) sched.deepAnalysisCore(t.Context()) - if got := c.LastSchedulerRun(); !got.Equal(prev) { - t.Errorf("section-failure pass advanced the marker to %v; want unchanged at %v", got.UTC(), prev.UTC()) + if got := stampTime(t, st); !got.Equal(prev) { + t.Errorf("section-failure pass advanced the record to %v; want unchanged at %v", got.UTC(), prev.UTC()) } }) } // TestDeepAnalysisCore_CapsLookback pins the maxDeepAnalysisLookback cap: a -// marker far older than the cap (a long outage or a very large +// record far older than the cap (a long outage or a very large // DEEP_SCAN_INTERVAL) must not grow the non-paginated History/RecentlyAdded // window without bound. The look-back is clamped to ~30 days regardless of how // old the previous run is. func TestDeepAnalysisCore_CapsLookback(t *testing.T) { t.Parallel() plx := &sinceCapturePlex{Plex: &fakeapi.Plex{}} - c := fakeapi.NewCache() - c.SetLastSchedulerRun(time.Now().Add(-60 * 24 * time.Hour)) // 60 days ago, well past the 30d cap sched := New( Config{Enable: true}, Deps{ Plex: plx, - Cache: c, + Cache: fakeapi.NewCache(), + Stamp: stampAt(t, time.Now().Add(-60*24*time.Hour)), // 60 days ago, well past the 30d cap UserClient: func(_ string) EpisodeReader { return plx.Plex }, Sync: &fakeSyncer{}, SaveCache: nil, @@ -1550,7 +1616,7 @@ func TestDeepAnalysisCore_CapsLookback(t *testing.T) { windowStart := time.Unix(plx.historySince.Load(), 0) lookback := after.Sub(windowStart) if lookback > 31*24*time.Hour { - t.Errorf("look-back %v exceeds the 30d cap; a 60d-old marker must be clamped (window start %v)", lookback, windowStart.UTC()) + t.Errorf("look-back %v exceeds the 30d cap; a 60d-old record must be clamped (window start %v)", lookback, windowStart.UTC()) } if lookback < 29*24*time.Hour { t.Errorf("look-back %v is below the 30d cap; expected the window clamped to ~30d (window start %v)", lookback, windowStart.UTC()) @@ -1560,7 +1626,7 @@ func TestDeepAnalysisCore_CapsLookback(t *testing.T) { // TestDeepAnalysisCore_ScatteredHistoryFailuresBelowBreakerStillAdvanceMarker // pins the completeness gate's accepted-loss boundary: scattered per-user // Episode() fetch failures that stay BELOW the circuit-breaker threshold do NOT -// block completion, so the pass is complete and the last-run marker advances. +// block completion, so the pass is complete and the run is recorded. // processRecentHistory keys completeness on fedAll (every item fed, breaker did // not abort), NOT on totalErrors==0 -- its comment documents this as "the // design's accepted skip-and-continue loss". The breaker-ABORT side is pinned by @@ -1568,8 +1634,8 @@ func TestDeepAnalysisCore_CapsLookback(t *testing.T) { // complement. Without it a regression coupling completeness to the error count // (return fedAll && totalErrors.Load()==0 && ctx.Err()==nil) survives every test // -- statement coverage of processRecentHistory/deepAnalysisCore is already -// 100% -- yet makes any pass with a single transient item failure never advance -// the marker, growing the look-back to the 30d cap and re-sweeping it every tick. +// 100% -- yet makes any pass with a single transient item failure never record, +// growing the look-back to the 30d cap and re-sweeping it every tick. func TestDeepAnalysisCore_ScatteredHistoryFailuresBelowBreakerStillAdvanceMarker(t *testing.T) { t.Parallel() // 3 episode items whose per-user Episode fetch always fails: 3 consecutive @@ -1577,21 +1643,21 @@ func TestDeepAnalysisCore_ScatteredHistoryFailuresBelowBreakerStillAdvanceMarker // feedHistory feeds every item (fedAll=true), and the pass is complete // despite the 3 accepted-loss failures. workers=1 keeps the count // deterministic. Sections are empty, so the recently-added leg is trivially - // complete and the marker's advance is governed solely by the history leg. + // complete and the record's advance is governed solely by the history leg. items := []plex.HistoryItem{ {AccountID: 1, RatingKey: "1", Type: plex.TypeEpisode}, {AccountID: 1, RatingKey: "2", Type: plex.TypeEpisode}, {AccountID: 1, RatingKey: "3", Type: plex.TypeEpisode}, } plx := &fakeapi.Plex{HistoryItems: items, EpisodeErr: errors.New("fetch boom")} - c := fakeapi.NewCache() - prev := time.Now().Add(-72 * time.Hour) // previous COMPLETED run's marker - c.SetLastSchedulerRun(prev) + prev := time.Now().Add(-72 * time.Hour) // previous COMPLETED run's record + st := stampAt(t, prev) sched := New( Config{Enable: true}, Deps{ Plex: plx, - Cache: c, + Cache: fakeapi.NewCache(), + Stamp: st, UserClient: func(_ string) EpisodeReader { return plx }, Sync: &fakeSyncer{}, SaveCache: nil, @@ -1601,8 +1667,8 @@ func TestDeepAnalysisCore_ScatteredHistoryFailuresBelowBreakerStillAdvanceMarker sched.deepAnalysisCore(t.Context()) - if got := c.LastSchedulerRun(); !got.After(prev) { - t.Errorf("pass with scattered below-breaker item failures did not advance the marker: got %v, still <= previous run %v (accepted-loss failures must not block completion)", got.UTC(), prev.UTC()) + if got := stampTime(t, st); !got.After(prev) { + t.Errorf("pass with scattered below-breaker item failures did not advance the record: got %v, still <= previous run %v (accepted-loss failures must not block completion)", got.UTC(), prev.UTC()) } } @@ -1631,6 +1697,7 @@ func TestDeepAnalysis_CleanPassRaisesNoIncompleteWarning(t *testing.T) { Deps{ Plex: plx, Cache: fakeapi.NewCache(), + Stamp: testStamp(t), UserClient: func(_ string) EpisodeReader { return plx }, Sync: &fakeSyncer{}, SaveCache: nil, diff --git a/internal/deepscan/deps.go b/internal/deepscan/deps.go index 94310b80..4d523c3d 100644 --- a/internal/deepscan/deps.go +++ b/internal/deepscan/deps.go @@ -2,7 +2,6 @@ package deepscan import ( "context" - "time" "github.com/cplieger/plex-language-sync/internal/plex" "github.com/cplieger/plex-language-sync/internal/streams" @@ -33,12 +32,10 @@ type EpisodeReader interface { // be built. Nil means skip the history item for that user. type UserClientFunc func(userID string) EpisodeReader -// runLedger is the persistence the pass needs: the dedup gate plus the last-run -// watermark that keeps a cold restart from double-running the analysis. +// runLedger is the persistence the pass needs: the dedup gate that keeps a +// recently-added episode from being processed twice. type runLedger interface { CheckAndMark(key string) bool - LastSchedulerRun() time.Time - SetLastSchedulerRun(t time.Time) } // skipChecker is the ignore decision. Two methods: the library-only check for diff --git a/internal/testsupport/fakeapi/cache.go b/internal/testsupport/fakeapi/cache.go index a02d5dd0..4957d64f 100644 --- a/internal/testsupport/fakeapi/cache.go +++ b/internal/testsupport/fakeapi/cache.go @@ -24,14 +24,13 @@ import ( // // The surface is exactly cache.Contract, and cache.RunContract is run against // it: consumers assert on the fake through the same readers production code -// uses (WasRecentlyProcessed, IntentFor, UserTokens, LastSchedulerRun) rather -// than through fake-only inspectors. +// uses (WasRecentlyProcessed, IntentFor, UserTokens) rather than through +// fake-only inspectors. type Cache struct { processed map[string]time.Time profiles map[string]map[string]string intents map[string]map[string]streams.Intent tokens map[string]string - lastRun time.Time recentWindow time.Duration mu sync.Mutex } @@ -161,18 +160,3 @@ func (c *Cache) SetUserTokens(tokens map[string]string) { maps.Copy(next, tokens) c.tokens = next } - -// LastSchedulerRun returns the recorded last-run timestamp. Zero value -// indicates "never run". -func (c *Cache) LastSchedulerRun() time.Time { - c.mu.Lock() - defer c.mu.Unlock() - return c.lastRun -} - -// SetLastSchedulerRun records the supplied timestamp. -func (c *Cache) SetLastSchedulerRun(t time.Time) { - c.mu.Lock() - defer c.mu.Unlock() - c.lastRun = t -} diff --git a/internal/testsupport/fakeapi/cache_test.go b/internal/testsupport/fakeapi/cache_test.go index c6782eed..0ae5f290 100644 --- a/internal/testsupport/fakeapi/cache_test.go +++ b/internal/testsupport/fakeapi/cache_test.go @@ -2,7 +2,6 @@ package fakeapi import ( "testing" - "time" "github.com/cplieger/plex-language-sync/internal/streams" ) @@ -52,14 +51,4 @@ func TestCacheRoundTrip(t *testing.T) { if c.UserTokens()["u1"] != "t1" { t.Error("SetUserTokens should defensive-copy its input") } - - // Scheduler run marker. - if !c.LastSchedulerRun().IsZero() { - t.Error("fresh cache should have zero LastSchedulerRun") - } - now := time.Now() - c.SetLastSchedulerRun(now) - if !c.LastSchedulerRun().Equal(now) { - t.Errorf("LastSchedulerRun = %v, want %v", c.LastSchedulerRun(), now) - } } diff --git a/main.go b/main.go index 947674a5..749ec5ae 100644 --- a/main.go +++ b/main.go @@ -24,6 +24,7 @@ import ( "net/http" "os" "os/signal" + "path/filepath" "sync" "syscall" "time" @@ -38,6 +39,7 @@ import ( "github.com/cplieger/plex-language-sync/internal/streams" "github.com/cplieger/plex-language-sync/internal/tracksync" "github.com/cplieger/plex-language-sync/internal/users" + "github.com/cplieger/scheduler/v4" ) // Compile-time assertion that the real client satisfies the per-user surface @@ -50,6 +52,11 @@ var _ tracksync.PlexReadWriter = (*plex.Client)(nil) // migrated on first load). Frozen by inviolate contract item 7 (file paths). const cacheDir = "/config" +// lastRunStampName is the deep-analysis last-run record (a scheduler.Stamp +// file) inside cacheDir, so it survives restarts on the same volume as the +// cache. +const lastRunStampName = ".plex-language-sync-last-run" + // shutdownWaitBudget bounds how long run() waits for background loops // (user-token refresh + scheduler) to join before persisting the cache // on shutdown. If the budget is exceeded the cache is saved anyway — a @@ -220,6 +227,7 @@ func run() int { deepscan.Deps{ Plex: client, Cache: c, + Stamp: scheduler.NewStamp(filepath.Join(cacheDir, lastRunStampName)), UserClient: scanUserClient, Sync: syncer, SaveCache: func() error { return c.Save(cacheDir) },