Fix episode skips and add Marvin repair dashboard (0.42.11)

Data/DatabaseManager.ImportCoverage.cs

@@ -17,6 +17,7 @@ public async Task EnsureImportCoverageAsync(CancellationToken ct = default)
         await ExecuteWriteAsync("CREATE TABLE IF NOT EXISTS import_aliases (alias TEXT PRIMARY KEY, identity TEXT NOT NULL);", _ => { }, ct);
         await ExecuteWriteAsync("CREATE TABLE IF NOT EXISTS import_attempts (id TEXT PRIMARY KEY, attempted_at TEXT NOT NULL);", _ => { }, ct);
         await ExecuteWriteAsync("CREATE INDEX IF NOT EXISTS ix_import_attempts_time ON import_attempts(attempted_at);", _ => { }, ct);
+        await ExecuteWriteAsync("CREATE TABLE IF NOT EXISTS import_run_history (id TEXT PRIMARY KEY, finished_at TEXT NOT NULL, payload TEXT NOT NULL);", _ => { }, ct);
         if (GetMetadata("import_schema") != "1") await PersistMetadataAsync("import_schema", "1", ct);
     }
 
@@ -98,6 +99,31 @@ public Task<int> GetRecentImportAttemptsAsync(DateTimeOffset now) => QueryScalar
         "SELECT COUNT(*) FROM import_attempts WHERE attempted_at >= @since;",
         c => BindText(c, "@since", now.AddDays(-1).ToString("o")));
 
+    public async Task<DateTimeOffset?> GetNextImportCreditAsync(DateTimeOffset now, int allowance)
+    {
+        var value = await QuerySingleAsync(@"SELECT attempted_at FROM import_attempts
+            WHERE attempted_at>=@since AND (SELECT COUNT(*) FROM import_attempts WHERE attempted_at>=@since)>=@limit
+            ORDER BY attempted_at LIMIT 1 OFFSET MAX(0,(SELECT COUNT(*) FROM import_attempts WHERE attempted_at>=@since)-@limit);",
+            c => { BindText(c, "@since", now.AddDays(-1).ToString("o")); BindInt(c, "@limit", Math.Max(1, allowance)); }, r => r.GetString(0));
+        return DateTimeOffset.TryParse(value, out var at) ? at.AddDays(1) : null;
+    }
+
+    public async Task SaveImportRunReportAsync(string id, DateTimeOffset finished, string report, CancellationToken ct)
+    {
+        await ExecuteWriteAsync("INSERT OR REPLACE INTO import_run_history(id,finished_at,payload) VALUES(@id,@at,@report);",
+            c => { BindText(c, "@id", id); BindText(c, "@at", finished.ToString("o")); BindText(c, "@report", report); }, ct);
+        // Analytics-only retention. The independent attempt ledger is never reset.
+        await ExecuteWriteAsync("DELETE FROM import_run_history WHERE finished_at<@at;",
+            c => BindText(c, "@at", finished.AddDays(-7).ToString("o")), ct);
+    }
+    public async Task<List<string>> GetImportRunHistoryAsync(int limit = 12)
+    {
+        if (await QueryScalarIntAsync("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='import_run_history';", _ => { }) == 0)
+            return new();
+        return await QueryListAsync("SELECT payload FROM import_run_history ORDER BY finished_at DESC,id LIMIT @limit;",
+            c => BindInt(c, "@limit", Math.Clamp(limit, 1, 48)), r => r.GetString(0));
+    }
+
     public async Task RecordImportAttemptAsync(string id, DateTimeOffset now, CancellationToken ct)
     {
         await ExecuteWriteAsync("INSERT INTO import_attempts(id,attempted_at) VALUES(@id,@at);",

InfiniteDrive.csproj

@@ -2,9 +2,9 @@
 
   <PropertyGroup>
     <TargetFramework>net8.0</TargetFramework>
-    <AssemblyVersion>0.42.10.0</AssemblyVersion>
-    <FileVersion>0.42.10.0</FileVersion>
-    <Version>0.42.10.0</Version>
+    <AssemblyVersion>0.42.11.0</AssemblyVersion>
+    <FileVersion>0.42.11.0</FileVersion>
+    <Version>0.42.11.0</Version>
     <Nullable>enable</Nullable>
     <RootNamespace>InfiniteDrive</RootNamespace>
     <AssemblyName>InfiniteDrive</AssemblyName>

Models/ImportCoverage.cs

@@ -14,6 +14,7 @@ public sealed class ImportCoverage
     public List<string> CatalogIds { get; set; } = new();
     public string SnapshotStatus { get; set; } = "unknown";
     public string ProviderStatus { get; set; } = "not_checked";
+    public int InventoryPolicyVersion { get; set; }
     public DateTimeOffset? SnapshotAt { get; set; }
     public DateTimeOffset? CheckedAt { get; set; }
     public DateTimeOffset? MetadataRetryAt { get; set; }

Models/ImportRunTelemetry.cs

@@ -0,0 +1,101 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading;
+
+namespace InfiniteDrive.Models;
+
+public sealed record ImportStageTiming(int Count, double TotalSeconds, double MeanSeconds,
+    double P95Seconds, double MaxSeconds, int SampleCount);
+public sealed record ImportRunSnapshot(string Id, DateTimeOffset StartedAt, DateTimeOffset ReadAt,
+    DateTimeOffset? FinishedAt, string Status, string Phase, string Title, string EpisodeKey,
+    double ElapsedSeconds, double PhaseSeconds, int EpisodesChecked, int MissingAttempts,
+    int RefreshAttempts, int ActiveLookups, int Matched, int EmptyResults, int TransportFailures,
+    int LookupDeadlines, int Http429, int ProviderConfigurationFailures, int CancelledLookups, int Published, int Refreshed,
+    int PublicationFailures, IReadOnlyDictionary<string, ImportStageTiming> Timings);
+
+/// <summary>One bounded, thread-safe run observation. Contains no target URLs or exception messages.</summary>
+public sealed class ImportRunTelemetry
+{
+    private static ImportRunTelemetry? _current;
+    public static ImportRunSnapshot? Current => Volatile.Read(ref _current)?.Snapshot();
+    private readonly object _gate = new();
+    private readonly string _id;
+    private readonly DateTimeOffset _started = DateTimeOffset.UtcNow;
+    private readonly Stopwatch _elapsed = Stopwatch.StartNew();
+    private readonly Stopwatch _phaseTime = Stopwatch.StartNew();
+    private readonly Dictionary<string, Stage> _timings = new(StringComparer.Ordinal);
+    private DateTimeOffset? _finished;
+    private string _status = "running", _phase = "inventory", _title = "", _episode = "";
+    private int _checked, _missing, _refresh, _active, _matched, _empty, _transport, _deadlines,
+        _http429, _providerConfiguration, _cancelled, _published, _refreshed, _publicationFailures;
+    public ImportRunTelemetry(string id) { _id = id; }
+    public static ImportRunTelemetry Start(string id)
+    {
+        var value = new ImportRunTelemetry(id); Interlocked.Exchange(ref _current, value); return value;
+    }
+    public void Work(string phase, string title = "", string episode = "")
+    {
+        lock (_gate) { _phase = phase; _title = title; _episode = episode; _phaseTime.Restart(); }
+    }
+    public void Checked() { lock (_gate) _checked++; }
+    public void Attempt(bool refresh) { lock (_gate) { if (refresh) _refresh++; else _missing++; } }
+    public void LookupStarted() { lock (_gate) _active++; }
+    public void LookupFinished(string failure)
+    {
+        lock (_gate)
+        {
+            _active--;
+            switch (failure)
+            {
+                case "": _matched++; break;
+                case "source_unavailable": _empty++; break;
+                case "transport_failure": _transport++; break;
+                case "lookup_deadline": _deadlines++; break;
+                case "http_429": _http429++; break;
+                case "provider_configuration": _providerConfiguration++; break;
+                case "cancelled": _cancelled++; break;
+            }
+        }
+    }
+    public void Published(bool refresh) { lock (_gate) { _published++; if (refresh) _refreshed++; } }
+    public void PublicationFailed() { lock (_gate) _publicationFailures++; }
+    public void RecordTiming(string stage, double seconds)
+    {
+        if (stage is not ("metadata" or "observation" or "resolution" or "publication" or "checkpoint") ||
+            !double.IsFinite(seconds) || seconds < 0) throw new ArgumentOutOfRangeException(nameof(stage));
+        lock (_gate)
+        {
+            if (!_timings.TryGetValue(stage, out var value)) _timings[stage] = value = new();
+            value.Count++; value.Total += seconds; value.Maximum = Math.Max(value.Maximum, seconds);
+            // The latest 512 timings bound memory, even for a long or abnormal run.
+            if (value.Samples.Count == 512) value.Samples.Dequeue();
+            value.Samples.Enqueue(seconds);
+        }
+    }
+    public void Finish(string status)
+    {
+        lock (_gate) { _status = status; _finished = DateTimeOffset.UtcNow; _elapsed.Stop(); _phaseTime.Stop(); _phase = "finished"; }
+    }
+    public ImportRunSnapshot Snapshot()
+    {
+        lock (_gate)
+        {
+            var times = _timings.ToDictionary(x => x.Key, x =>
+            {
+                var v = x.Value; var samples = v.Samples.OrderBy(n => n).ToArray();
+                return new ImportStageTiming(v.Count, Math.Round(v.Total, 3), Math.Round(v.Total / v.Count, 3),
+                    Math.Round(samples[(int)Math.Ceiling(samples.Length * .95) - 1], 3), Math.Round(v.Maximum, 3), samples.Length);
+            }, StringComparer.Ordinal);
+            return new(_id, _started, DateTimeOffset.UtcNow, _finished, _status, _phase, _title, _episode,
+                Math.Round(_elapsed.Elapsed.TotalSeconds, 3), Math.Round(_phaseTime.Elapsed.TotalSeconds, 3),
+                _checked, _missing, _refresh, _active, _matched, _empty, _transport, _deadlines, _http429,
+                _providerConfiguration, _cancelled, _published, _refreshed, _publicationFailures, times);
+        }
+    }
+    private sealed class Stage
+    {
+        public int Count; public double Total, Maximum; public Queue<double> Samples = new();
+    }
+}

README.md

@@ -9,7 +9,7 @@ imports titles and repairs missing files.
 
 A stream will appear. Probably.
 
-**Current release: 0.42.10**, built and tested against **Emby 4.10.0.40**.
+**Current release: 0.42.11**, built and tested against **Emby 4.10.0.40**.
 
 ## How we run it
 
@@ -203,14 +203,32 @@ Our Marvin task runs every ten minutes; a large library takes many passes.
 For a big refresh, engage **Infinite Improbability Drive** on the Marvin page.
 The switch runs up to 64 lookups together, paced to two starts a second. Each
 pass gets eight minutes, up to 4,096 attempts shared by missing-file repairs and
-existing-file refreshes, and a rolling daily ceiling of 40,000. Existing managed
-titles get priority. Completed lookups publish while inventory checks continue.
+existing-file refreshes, and a rolling daily ceiling of 40,000. Missing-file fills and existing-file refreshes alternate first priority across
+native runs; either can use capacity left by the other. Completed lookups publish while inventory checks continue.
+It isolates disputed episode numbering so confirmed siblings can proceed.
 It remembers refreshed episodes,
 keeps provider backoff and media protections, and switches itself off after seven
 days. Disengage it any time: **Normality has been restored.** Actual speed depends
 on your sources; the ceiling is not a completion estimate.
 See [Import recovery](docs/import-reconciliation.md) for limits and exclusions.
 
+### Marvin dashboard
+
+Open **Plugins → InfiniteDrive → Marvin** and use **Refresh dashboard**.
+It shows a dated snapshot of the current run: current step/title, lookups in
+flight, missing-file versus refresh attempts, matching sources, empty results,
+transport failures, lookup deadlines, actual HTTP 429 responses and publication
+outcomes. The rolling allowance is an attempt budget, not a library count.
+When it is full, the dashboard shows when the earliest usable credit ages out.
+
+Timings cover metadata, file/native observations, source resolution, publication
+and observation checkpoints. Parallel request totals overlap; they do not add
+up to elapsed time. Resolution includes dispatch pacing and provider wait.
+Mean/max/count cover the run; p95 uses the latest 512 samples per stage. Recent
+completed passes persist for seven days. Timing history starts with 0.42.11;
+older reports keep their original counters. A server restart ends the live
+sample, but preserves saved reports. Reading the dashboard starts no work.
+
 ## Set it up
 
 You'll need Emby Server **4.10.0.40**, an AIOStreams manifest and a compatible

Services/Api/ImportHealthService.cs

@@ -36,13 +36,14 @@ public async Task<object> Get(ImportHealthRequest request)
         if (deny != null) return deny;
         var db = Plugin.Instance.DatabaseManager;
         // No schema creation, provider calls, notifications, or task launch from GET.
-        if (db.GetMetadata("import_last_run") == null) return new { Status = "not_observed", Mode = Plugin.Instance.Configuration.ImportRecoveryMode.ToString() };
+        if (db.GetMetadata("import_last_run") == null) return new { Status = "not_observed", Mode = Plugin.Instance.Configuration.ImportRecoveryMode.ToString(), PipelinePhase = Plugin.Pipeline.Current, CurrentRun = ImportRunTelemetry.Current };
         var speed = ImportWorkBudget.For(Plugin.Instance.Configuration, DateTimeOffset.UtcNow);
         var used = await db.GetRecentImportAttemptsAsync(DateTimeOffset.UtcNow);
         var page = await db.GetImportCoveragePageAsync(request.Offset, request.Limit);
         return new { Mode = Plugin.Instance.Configuration.ImportRecoveryMode.ToString(),
             Speed = speed.IsCatchUp ? "catch_up" : "normal", speed.CatchUpUntil, speed.AttemptsPerDay, AttemptsUsed = used,
-            LastRun = db.GetMetadata("import_last_run"), CollectionHealth = db.GetMetadata("import_collection_health"), NextOffset = request.Offset + page.Count,
+            PipelinePhase = Plugin.Pipeline.Current, CurrentRun = ImportRunTelemetry.Current, NextCreditAt = await db.GetNextImportCreditAsync(DateTimeOffset.UtcNow, speed.AttemptsPerDay),
+            RecentRuns = await db.GetImportRunHistoryAsync(), LastRun = db.GetMetadata("import_last_run"), CollectionHealth = db.GetMetadata("import_collection_health"), NextOffset = request.Offset + page.Count,
             Items = page.Select(x => new { x.Identity, x.Title, Complete = x.Complete && x.SnapshotAt >= DateTimeOffset.UtcNow.AddHours(-6), x.SnapshotAt, x.CheckedAt,
                 x.SnapshotStatus, x.ProviderStatus, x.Exclusion,
                 Stale = x.SnapshotAt == null || x.SnapshotAt < DateTimeOffset.UtcNow.AddHours(-6),

Services/ImportInventory.cs

@@ -112,6 +112,7 @@ public async Task<ImportSnapshot> FetchAsync(CatalogItem item, CancellationToken
                     && string.IsNullOrEmpty(item.TvdbId) && string.IsNullOrEmpty(item.TmdbId) ? "unverified" : "aired"
             }).ToList();
         }
+        expected = NormalizeSourceNumbering(expected);
         var providerStatus = "not_applicable";
         if (IsSeries(item))
         {
@@ -125,21 +126,11 @@ public async Task<ImportSnapshot> FetchAsync(CatalogItem item, CancellationToken
                     var results = await _providers.GetAllEpisodes(series, _library.GetLibraryOptions(series), ct);
                     if (results.Length > 0)
                     {
-                        var providerKeys = results.Where(x => x.ParentIndexNumber.HasValue && x.IndexNumber.HasValue)
-                            .Select(x => $"aired:{x.ParentIndexNumber}:{x.IndexNumber}").ToHashSet(StringComparer.Ordinal);
-                        var common = expected.Count(x => providerKeys.Contains(x.Key));
-                        var compatible = expected.All(x => x.Numbering == "aired") && common == expected.Count
-                            && results.Count(x => x.ParentIndexNumber.HasValue && x.IndexNumber.HasValue) == providerKeys.Count;
-                        providerStatus = compatible ? "success" : "numbering_conflict";
-                        // Only accept provider-only keys when the inventories establish matching numbering.
-                        if (compatible)
-                            foreach (var result in results.Where(x => x.ParentIndexNumber.HasValue && x.IndexNumber.HasValue))
-                            {
-                                var key = $"aired:{result.ParentIndexNumber}:{result.IndexNumber}";
-                                if (expected.All(x => x.Key != key)) expected.Add(new ImportEpisode
-                                { Key = key, Season = result.ParentIndexNumber, Episode = result.IndexNumber,
-                                    Released = result.PremiereDate });
-                            }
+                        var providerEpisodes = results.Where(x => x.ParentIndexNumber.HasValue && x.IndexNumber.HasValue)
+                            .Select(x => new ImportEpisode
+                            { Key = $"aired:{x.ParentIndexNumber}:{x.IndexNumber}", Season = x.ParentIndexNumber,
+                                Episode = x.IndexNumber, Released = x.PremiereDate }).ToList();
+                        providerStatus = ReconcileNumbering(expected, providerEpisodes);
                     }
                 }
                 catch (OperationCanceledException) { throw; }
@@ -149,6 +140,31 @@ public async Task<ImportSnapshot> FetchAsync(CatalogItem item, CancellationToken
         return new(expected, providerStatus);
     }
 
+    public static List<ImportEpisode> NormalizeSourceNumbering(IEnumerable<ImportEpisode> episodes) =>
+        episodes.GroupBy(x => x.Key, StringComparer.Ordinal).Select(group =>
+        {
+            var episode = group.First();
+            if (group.Count() > 1) episode.Numbering = "conflict";
+            return episode;
+        }).ToList();
+
+    // Provider coverage is checked per key. A future/special/missing provider row
+    // does not invalidate siblings whose aired numbering agrees. No episode is
+    // mapped to another season or an absolute number to manufacture agreement.
+    public static string ReconcileNumbering(List<ImportEpisode> source, IReadOnlyList<ImportEpisode> provider)
+    {
+        var keys = provider.GroupBy(x => x.Key).ToDictionary(x => x.Key, x => x.ToList(), StringComparer.Ordinal);
+        foreach (var episode in source.Where(x => x.Numbering == "aired"))
+            episode.Numbering = !keys.TryGetValue(episode.Key, out var rows) ? "unconfirmed"
+                : rows.Count != 1 ? "conflict" : "aired";
+        var completeAgreement = source.All(x => x.Numbering == "aired");
+        // Provider-only additions retain the earlier strict agreement requirement.
+        if (completeAgreement)
+            foreach (var rows in keys.Values.Where(x => x.Count == 1))
+                if (source.All(x => x.Key != rows[0].Key)) source.Add(rows[0]);
+        return completeAgreement && keys.Values.All(x => x.Count == 1) ? "success" : "partial_numbering";
+    }
+
     public static DateTimeOffset? ParseRelease(string? value) => DateTimeOffset.TryParse(value,
         CultureInfo.InvariantCulture, DateTimeStyles.AssumeUniversal | DateTimeStyles.AdjustToUniversal, out var date) ? date : null;
 
@@ -282,6 +298,7 @@ private async Task<List<SelectedVersion>> ResolveCoreAsync(CatalogItem item, Imp
             await Plugin.Instance!.DatabaseManager.PersistMetadataAsync("import_provider_pause_" + client.ConfigurationFingerprint, DateTimeOffset.UtcNow.ToString("o"), ct);
             throw new ImportProviderConfigurationException();
         }
+        if (client.LastHttpStatus == 429) throw new ImportHttpRateLimitException();
         if (response == null) throw new IOException("stream_transport_unavailable");
         return VersionSelectorService.SelectBestVersions(StreamParser.ParseAll(response.Streams),
             _config.DesiredVersions, RuntimePolicy.EmbyVersionLimit, _config);
@@ -322,3 +339,5 @@ public void Notify(CatalogItem item)
 }
 
 public sealed class ImportProviderConfigurationException : Exception { }
+
+public sealed class ImportHttpRateLimitException : IOException { }