From 7238b809364afe1b8d42d53c3e590f7d025bcea6 Mon Sep 17 00:00:00 2001 From: OneHotTake Date: Wed, 30 Sep 2026 16:55:40 -0500 Subject: [PATCH 1/3] Fix per-episode numbering gates and rotate catch-up work priority --- InfiniteDrive.csproj | 6 +- Models/ImportCoverage.cs | 1 + README.md | 7 +- Services/ImportInventory.cs | 46 ++++++--- Services/ImportReconciliationService.cs | 92 +++++++++++++++--- Tests/ImportReconciliationTests.cs | 122 +++++++++++++++++++++++- docs/README.md | 2 +- docs/import-reconciliation-qa.md | 21 ++++ docs/import-reconciliation.md | 34 ++++++- plugin.json | 2 +- tools/ImportQa/ImportLabService.cs | 13 ++- tools/ImportQa/README.md | 7 ++ 12 files changed, 307 insertions(+), 46 deletions(-) diff --git a/InfiniteDrive.csproj b/InfiniteDrive.csproj index 0bf1910..50f46b1 100755 --- a/InfiniteDrive.csproj +++ b/InfiniteDrive.csproj @@ -2,9 +2,9 @@ net8.0 - 0.42.10.0 - 0.42.10.0 - 0.42.10.0 + 0.42.11.0 + 0.42.11.0 + 0.42.11.0 enable InfiniteDrive InfiniteDrive diff --git a/Models/ImportCoverage.cs b/Models/ImportCoverage.cs index 9a77808..f1f5f56 100644 --- a/Models/ImportCoverage.cs +++ b/Models/ImportCoverage.cs @@ -14,6 +14,7 @@ public sealed class ImportCoverage public List 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; } diff --git a/README.md b/README.md index 9b30a9a..e2a926c 100755 --- a/README.md +++ b/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,8 +203,9 @@ 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 diff --git a/Services/ImportInventory.cs b/Services/ImportInventory.cs index 4810e5a..5c0dd37 100644 --- a/Services/ImportInventory.cs +++ b/Services/ImportInventory.cs @@ -112,6 +112,7 @@ public async Task 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 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 FetchAsync(CatalogItem item, CancellationToken return new(expected, providerStatus); } + public static List NormalizeSourceNumbering(IEnumerable 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 source, IReadOnlyList 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; diff --git a/Services/ImportReconciliationService.cs b/Services/ImportReconciliationService.cs index 7e0ad50..1cbcbe5 100644 --- a/Services/ImportReconciliationService.cs +++ b/Services/ImportReconciliationService.cs @@ -57,6 +57,14 @@ public async Task RunAsync(CancellationToken ct, IReadOnlyList? sel var runId = Guid.NewGuid().ToString("N"); var status = "success"; var pending = new List(); + var deferred = new List(); + // Rotate the first claim on scarce credits across native slices. The other + // lane can borrow unused credits after inventory discovery; no new quota. + var priority = allowance.IsCatchUp && _db.GetMetadata("import_catch_up_last_priority") == "refresh" + ? "missing" : "refresh"; + var sourceFailures = 0; + var transportFailures = 0; + var refreshed = 0; async Task ResolveAsync(CatalogItem item, ImportEpisode episode) { @@ -105,12 +113,15 @@ async Task CompleteAsync(ResolutionWork work) await _db.SaveImportCoverageAsync(live, token); await RegisterPathsAsync(live, work.Item, episode, result.Versions, token); published++; + if (work.Upgrade) refreshed++; } catch (OperationCanceledException) when (token.IsCancellationRequested) { throw; } catch { failure = "publication_failed"; } } if (failure.Length > 0) { + if (failure == "source_unavailable") sourceFailures++; + if (failure == "transport_failure") transportFailures++; episode.State = work.Upgrade ? work.PreviousState : "retrying"; episode.Failure = failure; episode.NextAttempt = failure == "provider_configuration" ? null : @@ -144,9 +155,52 @@ async Task DrainCompletedAsync() await CompleteAsync(work); } } + async Task DispatchAsync(ResolutionCandidate candidate) + { + token.ThrowIfCancellationRequested(); + var now = _clock(); + if (_mode() != ImportMode.Repair || attempts >= allowance.AttemptsPerSlice || + candidate.Upgrade && upgrades >= allowance.UpgradesPerSlice || + _inventory.ProviderPaused || _cooldown() > now || + await _db.GetRecentImportAttemptsAsync(now) >= allowance.AttemptsPerDay) return; + if (allowance.IsCatchUp && _workBudget().CatchUpStartedAt != allowance.CatchUpStartedAt) + { budget.Cancel(); token.ThrowIfCancellationRequested(); } + // Deferred work may have waited behind earlier results. Recheck its + // durable authorization, generation, lease and backoff before a call. + Func start; + await MutationGate.WaitAsync(token); + try + { + var live = await _db.GetImportCoverageAsync(candidate.Coverage.Identity); + var episode = live?.Items.FirstOrDefault(x => x.Key == candidate.Episode.Key); + if (live == null || episode == null || live.Generation != candidate.Generation || + live.SnapshotStatus != "success" || !episode.Expected || !episode.Eligible || + episode.NextAttempt > now || episode.LeaseUntil > now || + !await IsAuthorizedAsync(live, candidate.Item, token) || _inventory.IsOwned(candidate.Item)) return; + if (candidate.Upgrade && !allowance.NeedsRefresh(episode, now)) return; + attempts++; + if (candidate.Upgrade) upgrades++; + var lease = Guid.NewGuid().ToString("N"); + episode.Lease = lease; episode.LeaseUntil = now.AddMinutes(10); + if (!candidate.Upgrade) episode.InitialFailure = true; + episode.Attempts++; + await _db.RecordImportAttemptAsync(lease, now, token); + await _db.SaveImportCoverageAsync(live, token); + var index = candidate.Coverage.Items.FindIndex(x => x.Key == episode.Key); + if (index >= 0) candidate.Coverage.Items[index] = episode; + start = () => new(candidate.Coverage, candidate.Item, episode, live.Generation, lease, + candidate.Upgrade, episode.State, ResolveAsync(candidate.Item, episode)); + } + finally { MutationGate.Release(); } + pending.Add(start()); + await DrainCompletedAsync(); + if (pending.Count >= allowance.Parallelism) await DrainOneAsync(); + } try { await _db.EnsureImportCoverageAsync(token); + if (allowance.IsCatchUp && _mode() == ImportMode.Repair) + await _db.PersistMetadataAsync("import_catch_up_last_priority", priority, token); var cursorKey = allowance.IsCatchUp ? "import_catch_up_scan_cursor" : "import_scan_cursor"; if (selected == null && allowance.IsCatchUp && _db.GetMetadata("import_catch_up_scan_window") != allowance.CatchUpStartedAt!.Value.ToString("o")) @@ -173,6 +227,7 @@ async Task DrainCompletedAsync() page = (await _db.GetDueImportCatalogAsync(_clock())).Concat(catchUp).Concat(page) .DistinctBy(x => x.Id).ToList(); } + var mayHaveRefresh = page.Any(x => !string.IsNullOrEmpty(x.StrmPath)); var seen = new HashSet(StringComparer.Ordinal); foreach (var sourceItem in page) { @@ -214,10 +269,13 @@ async Task DrainCompletedAsync() await SaveObservedAsync(coverage, token); continue; } - if (metadata < allowance.MetadataPerSlice && (coverage.MetadataRetryAt <= now || + var legacyNumberingGate = coverage.InventoryPolicyVersion == 0 && + coverage.SnapshotStatus == "identity_conflict" && coverage.ProviderStatus == "numbering_conflict"; + if (metadata < allowance.MetadataPerSlice && (legacyNumberingGate || coverage.MetadataRetryAt <= now || !coverage.MetadataRetryAt.HasValue && (!coverage.SnapshotAt.HasValue || coverage.SnapshotAt <= now.AddHours(-6)))) { metadata++; + coverage.InventoryPolicyVersion = 1; try { using var deadline = CancellationTokenSource.CreateLinkedTokenSource(token); @@ -269,7 +327,12 @@ async Task DrainCompletedAsync() observation.Conflict, now); episode.ObservedAt = now; if (episode.State == "indexed") - { episode.LastSuccess = now; episode.Failure = ""; episode.Lease = null; episode.LeaseUntil = null; } + { + episode.LastSuccess = now; + // Native indexing establishes observation success, not + // success of a later source refresh. Preserve its failure. + if (episode.LeaseUntil <= now) { episode.Lease = null; episode.LeaseUntil = null; } + } coverage.Cursor = episode.Key; coverage.CheckedAt = now; await SaveObservedAsync(coverage, token); @@ -296,17 +359,11 @@ async Task DrainCompletedAsync() if ((episode.State != "missing" && !upgrade) || coverage.SnapshotStatus != "success" || _inventory.ProviderPaused || attempts >= allowance.AttemptsPerSlice || _cooldown() > now || await _db.GetRecentImportAttemptsAsync(now) >= allowance.AttemptsPerDay) continue; - attempts++; - if (upgrade) upgrades++; - var lease = Guid.NewGuid().ToString("N"); - episode.Lease = lease; episode.LeaseUntil = now.AddMinutes(10); - if (!upgrade) episode.InitialFailure = true; episode.Attempts++; - await _db.RecordImportAttemptAsync(lease, now, token); - await SaveObservedAsync(coverage, token); - pending.Add(new(coverage, item, episode, coverage.Generation, lease, upgrade, - episode.State, ResolveAsync(item, episode))); - await DrainCompletedAsync(); - if (pending.Count >= allowance.Parallelism) await DrainOneAsync(); + var candidate = new ResolutionCandidate(coverage, item, episode, coverage.Generation, upgrade); + if (allowance.IsCatchUp && (upgrade ? "refresh" : "missing") != priority && + (priority != "refresh" || mayHaveRefresh)) + deferred.Add(candidate); + else await DispatchAsync(candidate); } scanned++; if (coverage.SnapshotStatus == "success" && coverage.Items.Any(x => x.ObservedAt.HasValue)) @@ -325,6 +382,11 @@ async Task DrainCompletedAsync() await _db.PersistMetadataAsync(cursorKey, sourceItem.Id, token); } } + foreach (var candidate in deferred) + { + await DrainCompletedAsync(); + await DispatchAsync(candidate); + } while (pending.Count > 0) await DrainOneAsync(); } catch (OperationCanceledException) { status = ct.IsCancellationRequested ? "cancelled" : "budget_deferred"; } @@ -336,13 +398,15 @@ async Task DrainCompletedAsync() await Task.WhenAll(pending.Select(x => x.Resolution)); try { await _db.PersistMetadataAsync("import_last_run", System.Text.Json.JsonSerializer.Serialize(new { Id = runId, Status = status, FinishedAt = _clock(), Scanned = scanned, Attempts = attempts, Metadata = metadata, Published = published, - Upgrades = upgrades, ElapsedSeconds = elapsed.Elapsed.TotalSeconds, + Upgrades = upgrades, Refreshed = refreshed, SourceFailures = sourceFailures, + TransportFailures = transportFailures, Priority = allowance.IsCatchUp ? priority : "normal", ElapsedSeconds = elapsed.Elapsed.TotalSeconds, Speed = allowance.IsCatchUp ? "catch_up" : "normal", allowance.AttemptsPerDay, allowance.Parallelism, allowance.CatchUpStartedAt, allowance.CatchUpUntil }), CancellationToken.None); } finally { RunGate.Release(); } } } + private sealed record ResolutionCandidate(ImportCoverage Coverage, CatalogItem Item, ImportEpisode Episode, long Generation, bool Upgrade); private sealed record ResolutionResult(List? Versions, string Failure); private sealed record ResolutionWork(ImportCoverage Coverage, CatalogItem Item, ImportEpisode Episode, long Generation, string Lease, bool Upgrade, string PreviousState, Task Resolution); diff --git a/Tests/ImportReconciliationTests.cs b/Tests/ImportReconciliationTests.cs index cfdf2e1..070fc79 100644 --- a/Tests/ImportReconciliationTests.cs +++ b/Tests/ImportReconciliationTests.cs @@ -491,6 +491,116 @@ [Fact] public async Task ScheduledCatchUpResetsItsCursorAndScansExistingFilesBef await worker.RunAsync(default); Assert.Equal(2, h.Inventory.Published); } + [Fact] public void MissingFutureAndSpecialProviderRowsDoNotBlockConfirmedSiblings() + { + var source = new List { Ep(), Ep(2), + new() { Key = "aired:2:1", Season = 2, Episode = 1, Released = Now.AddDays(1) }, + new() { Key = "aired:0:1", Season = 0, Episode = 1, Released = Now.AddDays(-1) } }; + Assert.Equal("partial_numbering", ImportInventory.ReconcileNumbering(source, new[] { Ep() })); + Assert.Equal("eligible", ImportCoveragePolicy.Eligibility(source[0], false, Now, TimeZoneInfo.Utc)); + Assert.All(source.Skip(1), e => Assert.NotEqual("eligible", ImportCoveragePolicy.Eligibility(e, false, Now, TimeZoneInfo.Utc))); + Assert.Equal(4, source.Count); + } + [Fact] public void DuplicateProviderEpisodeBlocksOnlyThatKeyAndDoesNotInventASeason() + { + var source = new List { Ep(), Ep(2) }; + ImportInventory.ReconcileNumbering(source, new[] { Ep(), Ep(), Ep(2), Ep(3) }); + Assert.Equal("conflict", source[0].Numbering); + Assert.Equal("aired", source[1].Numbering); + Assert.Equal(2, source.Count); // No provider-only expansion on partial agreement. + } + [Fact] public void DuplicateSourceKeysRemainExplicitlyConflictedWithConfirmedSiblingsUsable() + { + var source = ImportInventory.NormalizeSourceNumbering(new[] { Ep(), Ep(), Ep(2) }); + Assert.True(ImportCoveragePolicy.AcceptSnapshot(new List(), source)); + ImportInventory.ReconcileNumbering(source, new[] { Ep(), Ep(2) }); + Assert.Equal("conflict", source[0].Numbering); Assert.Equal("aired", source[1].Numbering); + } + [Fact] public void UnverifiedAnimeAndDisjointInventoriesNeverForceMatches() + { + var source = new List { Ep() }; source[0].Numbering = "unverified"; + ImportInventory.ReconcileNumbering(source, new[] { Ep() }); + Assert.Equal("unverified", source[0].Numbering); + source = new() { Ep() }; ImportInventory.ReconcileNumbering(source, new[] { Ep(2) }); + Assert.Single(source); Assert.Equal("unconfirmed", source[0].Numbering); + } + [Fact] public void ProviderOnlyEpisodesRequireCompleteSourceAgreement() + { + var source = new List { Ep() }; + Assert.Equal("success", ImportInventory.ReconcileNumbering(source, new[] { Ep(), Ep(2) })); + Assert.Equal(new[] { 1, 2 }, source.Select(x => x.Episode!.Value)); + } + [Fact] public async Task PartialNumberingPublishesConfirmedEpisodesAndKeepsDisputedOnesExcluded() + { + using var h = new Harness(); await h.Seed(); h.Inventory.DisputedEpisode = 2; + await h.Run(); Assert.Equal(1, h.Inventory.Published); + var state = await h.State(); Assert.Equal("success", state.SnapshotStatus); + Assert.Equal("partial_numbering", state.ProviderStatus); + Assert.Equal("numbering_conflict", state.Items.Single(x => x.Episode == 2).Eligibility); + } + [Fact] public async Task LegacyWholeSeriesGateIsRecheckedWithoutResettingRetriesOrIdentityExclusions() + { + using var h = new Harness(); await h.Seed(); await h.Run(ImportMode.Observe); + var state = await h.State(); state.SnapshotStatus = "identity_conflict"; + state.ProviderStatus = "numbering_conflict"; state.InventoryPolicyVersion = 0; + state.MetadataRetryAt = h.Now.AddHours(6); + state.Items[0].NextAttempt = h.Now.AddHours(2); state.Items[0].Attempts = 3; + await h.Db.SaveImportCoverageAsync(state); h.Inventory.DisputedEpisode = 2; + await h.Run(); state = await h.State(); Assert.Equal("success", state.SnapshotStatus); + Assert.Equal(1, state.InventoryPolicyVersion); Assert.Equal(3, state.Items[0].Attempts); + Assert.Equal(h.Now.AddHours(2), state.Items[0].NextAttempt); + Assert.Equal(0, h.Inventory.Resolutions); + } + [Fact] public async Task GenuineIdentityConflictDoesNotBecomeAOneTimeNumberingRetry() + { + using var h = new Harness(); await h.Seed(); await h.Run(ImportMode.Observe); + var state = await h.State(); state.SnapshotStatus = "identity_conflict"; + state.ProviderStatus = "success"; state.MetadataRetryAt = h.Now.AddHours(6); + state.InventoryPolicyVersion = 0; await h.Db.SaveImportCoverageAsync(state); + await h.Run(); Assert.Equal("identity_conflict", (await h.State()).SnapshotStatus); + Assert.Equal(0, h.Inventory.Resolutions); + } + [Fact] public async Task IndexedOldFilePreservesFailedRefreshReasonAndBackoffUntilSuccess() + { + using var h = new Harness(); h.Inventory.Count = 1; h.Inventory.Files.Add(1); + h.Inventory.FailEpisode = 1; await h.Seed(); h.Engage(); await h.Run(); + var before = (await h.State()).Items.Single(); Assert.Equal("source_unavailable", before.Failure); + h.Db = new DatabaseManager(h.Directory.Path, NullLogger.Instance); h.Db.Initialise(); + h.Now = h.Now.AddHours(1); await h.Run(); var after = (await h.State()).Items.Single(); + Assert.Equal("indexed", after.State); Assert.Equal(before.Failure, after.Failure); + Assert.Equal(before.NextAttempt, after.NextAttempt); Assert.Equal(1, h.Inventory.Resolutions); + Assert.Contains(1, h.Inventory.Files); Assert.Null(after.LastVersionRefresh); + h.Now = h.Now.AddDays(1); h.Inventory.FailEpisode = 0; await h.Run(); + after = (await h.State()).Items.Single(); Assert.Equal("", after.Failure); + Assert.Null(after.NextAttempt); Assert.Equal(h.Now, after.LastVersionRefresh); + } + [Fact] public async Task ScarceSliceCreditsRotateAcrossMissingAndRefreshWorkAfterRestart() + { + using var h = new Harness(); h.Inventory.Count = 10; h.Item.StrmPath = "/fake/series"; + h.Inventory.Files.UnionWith(Enumerable.Range(6, 5)); await h.Seed(); h.Engage(); + var limit = ImportWorkBudget.For(h.Config, h.Now) with { AttemptsPerSlice = 2 }; + Task Run() => new ImportReconciliationService(h.Db, h.Inventory, () => ImportMode.Repair, + TimeZoneInfo.Utc, () => h.Now, workBudget: () => limit).RunAsync(default, new[] { h.Item }); + await Run(); Assert.All(h.Inventory.ResolvedEpisodes, e => Assert.True(e >= 6)); + Assert.Equal(2, h.Inventory.Published); + h.Db = new DatabaseManager(h.Directory.Path, NullLogger.Instance); h.Db.Initialise(); + h.Inventory.ResolvedEpisodes.Clear(); await Run(); + Assert.Equal(2, h.Inventory.ResolvedEpisodes.Count); + Assert.All(h.Inventory.ResolvedEpisodes, e => Assert.True(e < 6)); + Assert.Equal(4, await h.Db.GetRecentImportAttemptsAsync(h.Now)); + } + [Fact] public async Task NonPriorityLaneBorrowsUnusedCreditsAndRechecksALateBlock() + { + using var h = new Harness(); h.Item.StrmPath = "/fake/series"; + await h.Seed(); h.Engage(); await h.Run(); Assert.Equal(2, h.Inventory.Published); + h.Inventory.Files.Clear(); h.Now = h.Now.AddMinutes(1); + await h.Db.PersistMetadataAsync("import_catch_up_last_priority", "missing"); + h.Inventory.OnObserve = async ep => { if (ep.Episode == 2) + await h.Db.UpsertBlockedItemAsync(h.Item.AioId, null, null, "fixture", "series", "test"); }; + var previous = h.Inventory.Resolutions; await h.Run(); + Assert.Equal(previous, h.Inventory.Resolutions); + } + public class NullLogProxy : System.Reflection.DispatchProxy { protected override object? Invoke(System.Reflection.MethodInfo? method, object?[]? args) @@ -516,9 +626,15 @@ private sealed class Harness : IDisposable } private sealed class FakeInventory : IImportInventory { - public int Resolutions, Published, Notifications, FailEpisode, Count = 2; + public int Resolutions, Published, Notifications, FailEpisode, DisputedEpisode, Count = 2; + public List ResolvedEpisodes = new(); public bool Owned, BadSnapshot, Paused, Indexed = true; public bool ProviderPaused => Paused; public HashSet Files = new(); public Func? OnResolve; public Func? OnResolveWithToken; public Func? OnObserve; - public Task FetchAsync(CatalogItem item, CancellationToken ct) => Task.FromResult(new ImportSnapshot(BadSnapshot ? new() : Enumerable.Range(1, Count).Select(Ep).ToList(), "success")); + public Task FetchAsync(CatalogItem item, CancellationToken ct) + { + var episodes = BadSnapshot ? new List() : Enumerable.Range(1, Count).Select(Ep).ToList(); + if (DisputedEpisode > 0) foreach (var ep in episodes.Where(x => x.Episode == DisputedEpisode)) ep.Numbering = "unconfirmed"; + return Task.FromResult(new ImportSnapshot(episodes, DisputedEpisode > 0 ? "partial_numbering" : "success")); + } public async Task ObserveAsync(CatalogItem item, ImportEpisode ep, CancellationToken ct) { if (OnObserve != null) await OnObserve(ep); @@ -526,7 +642,7 @@ public async Task ObserveAsync(CatalogItem item, ImportEpisod } public bool IsOwned(CatalogItem item) => Owned; public async Task> ResolveAsync(CatalogItem item, ImportEpisode ep, CancellationToken ct) - { Resolutions++; if (OnResolve != null) await OnResolve(); if (OnResolveWithToken != null) await OnResolveWithToken(ct); return ep.Episode == FailEpisode ? new() : new() { new() { Stream = new() { Url = "https://example.invalid/test" } } }; } + { Resolutions++; ResolvedEpisodes.Add(ep.Episode!.Value); if (OnResolve != null) await OnResolve(); if (OnResolveWithToken != null) await OnResolveWithToken(ct); return ep.Episode == FailEpisode ? new() : new() { new() { Stream = new() { Url = "https://example.invalid/test" } } }; } public Task> PublishAsync(CatalogItem item, ImportEpisode ep, List versions, CancellationToken ct) { Published++; Files.Add(ep.Episode!.Value); return Task.FromResult(new List { $"/fake/series/Season 01/e{ep.Episode}.strm" }); } public void Notify(CatalogItem item) => Notifications++; diff --git a/docs/README.md b/docs/README.md index cbb062d..5ac8096 100644 --- a/docs/README.md +++ b/docs/README.md @@ -26,4 +26,4 @@ old test plans and dated reports. These are not current configuration instructions. The implementation and current guides take precedence over historical claims. -Updated September 30, 2026 for the 0.42.10 source tree. +Updated September 30, 2026 for the 0.42.11 source tree. diff --git a/docs/import-reconciliation-qa.md b/docs/import-reconciliation-qa.md index 1355aee..f7ccb27 100644 --- a/docs/import-reconciliation-qa.md +++ b/docs/import-reconciliation-qa.md @@ -1,6 +1,27 @@ # Import reconciliation QA — September 29, 2026 +## Episode numbering and fair reconciliation — September 30, 2026 + +Version 0.42.11: **213/213 tests passed**, zero skipped, against pinned Emby +4.10.0.40 and real SQLite. New regressions check per-key source/provider +agreement, duplicate and unverified numbering, legacy whole-series gate +reevaluation, identity safeguards, failed-refresh diagnostics/backoff across a +database reopen, alternating lane priority and authorization of deferred calls. + +An isolated frozen beta used a new fixed fixture root and separate fixture DB. +With only episode 1 confirmed by the provider inventory, one episode group/two +STRMs published and indexed; episode 2 remained excluded with numbering conflict. +A native restart preserved both outcomes. Advancing the fixture clock and +restoring complete numbering published episode 2 and refreshed episode 1; four +STRMs then indexed as two matching native episodes. There was no fixture DB reset. +These synthetic targets verify publication and native identity/numbering/indexing, +not playback. Production was unchanged throughout this QA. The lab retains its +freeze, read-only physical-media mounts and network isolation; its prior artifact, +configuration and unrelated state are preserved. Temporary QA code never enters +the release DLL. Only the named fixture/library and temporary hidden administrator +are removed during cleanup; the private backup is retained. + ## Scheduled cursor correction — September 29, 2026 All **126 tests passed** against the pinned Emby runtime and real SQLite, with diff --git a/docs/import-reconciliation.md b/docs/import-reconciliation.md index a40f67d..0d83410 100644 --- a/docs/import-reconciliation.md +++ b/docs/import-reconciliation.md @@ -74,8 +74,16 @@ another check. The administrative API retains its technical field/state names. `ImportInventory` wraps source metadata, Emby's in-process `IProviderManager.GetAllEpisodes`, native library reads, the existing resolver and version selector, and the STRM writer. The source inventory remains primary. -Provider-only keys require a compatible inventory with all source numbering keys -represented; ambiguous inventories stop new writes. Provider unavailability is +Numbering is checked per episode key when Emby's provider has an inventory. +A unique matching aired season/episode remains eligible even if another source +key is absent (including future episodes and specials). Unmatched keys are +`unconfirmed`; duplicate source/provider keys are `conflict`. Those keys remain +excluded, without inventing season mappings. Unverified anime stays excluded. +Provider-only additions still require every source key to agree. Empty provider +results/unavailable checks retain the primary-source behavior and are reported; +they do not certify completeness. The old whole-series numbering gate gets one +fresh metadata check under the new policy; genuine identity exclusions, stream +backoff, attempts and user state remain untouched. Provider unavailability is reported explicitly. The API choice follows the [Emby developer discussion](https://emby.media/community/topic/149945-getting-missing-episodes-from-a-series-and-movies-from-collections/). No `GetAllItems` franchise expansion is enabled. Three additive SQLite tables implement the journal without changing existing @@ -127,7 +135,13 @@ These are ceilings, not promised throughput. Upstream latency, unavailable sourc provider backoff, native indexing and the rest of Marvin's work determine actual progress. The existing task schedule stays in place. Catch-up has a separate 64-slot AIO maintenance lane, paced to two starts a second. Missing-file repairs and existing -version refreshes share its larger attempt allowance. Ordinary lookups retain their +version refreshes share its larger attempt allowance. Catch-up alternates which +lane gets first access to credits on each native slice, with a durable priority +checkpoint across restarts. The other lane borrows unused credits after inventory +discovery. A page without any managed paths does not hold missing work behind an +absent refresh lane. Observation/page deadlines can postpone borrowed work; this +is priority rotation, not an exact 50/50 allocation or a throughput guarantee. +Deferred calls recheck authorization, generation, leases and backoff. Ordinary lookups retain their shared two-slot lane. File publication and state changes remain serial. No extra scheduler is started. Completed lookups publish between inventory checks, rather than waiting for a full request batch or the end of the catalog page. Queued lookups are joined before a cancelled run releases @@ -144,7 +158,10 @@ larger than a 200-key page. An unavailable episode does not prevent refreshing its eligible siblings. Valid existing managed files can be refreshed while native indexing is pending; publication still reports awaiting indexing, and an identity mismatch prevents refresh. Failed attempts -retain their backoff; owned media, explicit blocks, pruning and publication checks +retain their backoff and diagnostic even when an old file is indexed; native +indexing does not clear a failed source refresh. Successful replacement clears +that failure. LastRun distinguishes attempted upgrades from successful refreshed +groups, empty-source failures and transport failures; owned media, explicit blocks, pruning and publication checks still apply. This switch changes recovery speed, not quality filters or retention. Disengagement/expiry is checked before another stream attempt and again before @@ -218,6 +235,15 @@ status never creates tables, contacts providers, or triggers work. ## Verification and rollout +September 30, 2026, 0.42.11 regression build: 213/213 tests passed against +pinned Emby 4.10.0.40, with no skipped tests. New cases cover partial/disjoint +provider inventories, duplicate source/provider numbering, unverified anime, +provider-only additions, legacy gate reevaluation, genuine identity conflict, +failed refresh diagnostics across restart, priority rotation and late blocks. +This is implementation evidence; production throughput and playback require +separate live checks. + + September 30, 2026 pruning safety verification: the pinned Emby 4.10.0.40 build passed all 199 tests and published a release DLL. New cases exercise real SQLite absence counting across more than two batches, the one-time policy reset, guarded diff --git a/plugin.json b/plugin.json index 181ccbd..2295c7e 100755 --- a/plugin.json +++ b/plugin.json @@ -5,7 +5,7 @@ "overview": "InfiniteDrive discovers your AIOStreams catalog, writes .strm files, and resolves debrid URLs on demand. Like the Infinite Improbability Drive: a stream will appear. Probably. No external processes needed.", "owner": "InfiniteDrive", "category": "General", - "version": "0.42.10.0", + "version": "0.42.11.0", "targetAbi": "4.10.0.40", "framework": "net8.0", "imageUrl": "" diff --git a/tools/ImportQa/ImportLabService.cs b/tools/ImportQa/ImportLabService.cs index 68c5bd6..84c4fd3 100644 --- a/tools/ImportQa/ImportLabService.cs +++ b/tools/ImportQa/ImportLabService.cs @@ -54,6 +54,10 @@ public async Task Post(ImportQaRequest request) } mode = ImportMode.Observe; break; + case "partial-numbering": + fake.PartialNumbering = true; now = now.AddHours(8); + break; + case "recheck-numbering": now = now.AddHours(16); break; case "fail-second": fake.FailSecond = true; break; case "retry": now = now.AddHours(8); break; case "observe": mode = ImportMode.Observe; break; @@ -118,11 +122,16 @@ private static async Task Status(DatabaseManager db, LabInventory fake) private sealed class LabInventory : IImportInventory { private readonly ImportInventory _real; - public bool FailSecond; public int Resolutions, Published; + public bool FailSecond, PartialNumbering; public int Resolutions, Published; public List Episodes => Enumerable.Range(1, 2).Select(n => new ImportEpisode { Key = $"aired:1:{n}", Season = 1, Episode = n, Released = DateTimeOffset.Parse("2020-01-01T00:00:00Z") }).ToList(); public LabInventory(ImportInventory real) { _real = real; } - public Task FetchAsync(CatalogItem item, CancellationToken ct) => Task.FromResult(new ImportSnapshot(Episodes, "fixture")); + public Task FetchAsync(CatalogItem item, CancellationToken ct) + { + var episodes = Episodes; + var status = PartialNumbering ? ImportInventory.ReconcileNumbering(episodes, episodes.Take(1).ToList()) : "fixture"; + return Task.FromResult(new ImportSnapshot(episodes, status)); + } public Task ObserveAsync(CatalogItem item, ImportEpisode ep, CancellationToken ct) => _real.ObserveAsync(item, ep, ct); public bool IsOwned(CatalogItem item) => false; public Task> ResolveAsync(CatalogItem item, ImportEpisode ep, CancellationToken ct) diff --git a/tools/ImportQa/README.md b/tools/ImportQa/README.md index d8a015e..8de8733 100644 --- a/tools/ImportQa/README.md +++ b/tools/ImportQa/README.md @@ -38,3 +38,10 @@ synthetic `example.invalid` URLs, not playback tests. Clean up only the named QA collection/library, fixture media/database, and temporary QA account/session. Keep pre-test snapshots and the prior plugin artifact for rollback. Do not reset whole beta datasets or restart production. + +Numbering regression: on a freshly prepared fixture, `partial-numbering` passes +only episode 1 through the real per-key numbering policy. Its STRMs publish; +episode 2 stays excluded. `discover` observes native indexing. +`recheck-numbering` advances the fixture clock eight hours and restores the +complete source inventory; episode 2 becomes eligible without a database reset. +These are synthetic targets and do not establish playback. From c32e43ca8974e89082c4883f96cc9974ac339b55 Mon Sep 17 00:00:00 2001 From: OneHotTake Date: Wed, 30 Sep 2026 17:19:21 -0500 Subject: [PATCH 2/3] Add read-only Marvin dashboard and bounded run timing history --- Data/DatabaseManager.ImportCoverage.cs | 26 ++++++ Models/ImportRunTelemetry.cs | 101 ++++++++++++++++++++++++ README.md | 17 ++++ Services/Api/ImportHealthService.cs | 5 +- Services/ImportInventory.cs | 3 + Services/ImportReconciliationService.cs | 54 ++++++++++--- Tests/ImportAnalyticsTests.cs | 75 ++++++++++++++++++ Tests/ImportReconciliationTests.cs | 23 +++++- UI/Settings/SyncAndMarvinTabView.cs | 63 +++++++++++++-- UI/Settings/SyncAndMarvinUI.cs | 6 ++ docs/import-reconciliation-qa.md | 2 +- docs/import-reconciliation.md | 40 +++++++++- tools/ImportQa/ImportLabService.cs | 4 + 13 files changed, 393 insertions(+), 26 deletions(-) create mode 100644 Models/ImportRunTelemetry.cs create mode 100644 Tests/ImportAnalyticsTests.cs diff --git a/Data/DatabaseManager.ImportCoverage.cs b/Data/DatabaseManager.ImportCoverage.cs index e369872..e3919d3 100644 --- a/Data/DatabaseManager.ImportCoverage.cs +++ b/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 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 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> 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);", diff --git a/Models/ImportRunTelemetry.cs b/Models/ImportRunTelemetry.cs new file mode 100644 index 0000000..f7dfb60 --- /dev/null +++ b/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 Timings); + +/// One bounded, thread-safe run observation. Contains no target URLs or exception messages. +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 _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 Samples = new(); + } +} diff --git a/README.md b/README.md index e2a926c..04bf5d2 100755 --- a/README.md +++ b/README.md @@ -212,6 +212,23 @@ days. Disengage it any time: **Normality has been restored.** Actual speed depen 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 diff --git a/Services/Api/ImportHealthService.cs b/Services/Api/ImportHealthService.cs index 404b871..f1fddb8 100644 --- a/Services/Api/ImportHealthService.cs +++ b/Services/Api/ImportHealthService.cs @@ -36,13 +36,14 @@ public async Task 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(), 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, + 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), diff --git a/Services/ImportInventory.cs b/Services/ImportInventory.cs index 5c0dd37..b4280e9 100644 --- a/Services/ImportInventory.cs +++ b/Services/ImportInventory.cs @@ -298,6 +298,7 @@ private async Task> 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); @@ -338,3 +339,5 @@ public void Notify(CatalogItem item) } public sealed class ImportProviderConfigurationException : Exception { } + +public sealed class ImportHttpRateLimitException : IOException { } diff --git a/Services/ImportReconciliationService.cs b/Services/ImportReconciliationService.cs index 1cbcbe5..0ad143e 100644 --- a/Services/ImportReconciliationService.cs +++ b/Services/ImportReconciliationService.cs @@ -22,6 +22,7 @@ public sealed class ImportReconciliationService private readonly Func _cooldown; private readonly TimeZoneInfo _timezone; private readonly Func _workBudget; + private ImportRunTelemetry? _activeTelemetry; public ImportReconciliationService(DatabaseManager db, IImportInventory inventory, Func mode, TimeZoneInfo timezone, Func? clock = null, @@ -56,6 +57,8 @@ public async Task RunAsync(CancellationToken ct, IReadOnlyList? sel var notified = false; var runId = Guid.NewGuid().ToString("N"); var status = "success"; + var telemetry = ImportRunTelemetry.Start(runId); + _activeTelemetry = telemetry; var pending = new List(); var deferred = new List(); // Rotate the first claim on scarce credits across native slices. The other @@ -68,18 +71,29 @@ public async Task RunAsync(CancellationToken ct, IReadOnlyList? sel async Task ResolveAsync(CatalogItem item, ImportEpisode episode) { + telemetry.LookupStarted(); + var timer = Stopwatch.StartNew(); + var failure = ""; + List? versions = null; + using var deadline = CancellationTokenSource.CreateLinkedTokenSource(token); + deadline.CancelAfter(TimeSpan.FromSeconds(allowance.IsCatchUp ? 120 : 60)); try { - using var deadline = CancellationTokenSource.CreateLinkedTokenSource(token); - deadline.CancelAfter(TimeSpan.FromSeconds(allowance.IsCatchUp ? 120 : 60)); - var versions = allowance.IsCatchUp + versions = allowance.IsCatchUp ? await _inventory.ResolveCatchUpAsync(item, episode, deadline.Token) : await _inventory.ResolveAsync(item, episode, deadline.Token); - return new(versions, versions.Count == 0 ? "source_unavailable" : ""); + failure = versions.Count == 0 ? "source_unavailable" : ""; } - catch (ImportProviderConfigurationException) { return new(null, "provider_configuration"); } - catch (OperationCanceledException) when (token.IsCancellationRequested) { return new(null, "cancelled"); } - catch { return new(null, "transport_failure"); } + catch (ImportProviderConfigurationException) { failure = "provider_configuration"; } + catch (ImportHttpRateLimitException) { failure = "http_429"; } + catch (OperationCanceledException) { failure = token.IsCancellationRequested ? "cancelled" : deadline.IsCancellationRequested ? "lookup_deadline" : "transport_failure"; } + catch { failure = "transport_failure"; } + finally + { + telemetry.RecordTiming("resolution", timer.Elapsed.TotalSeconds); + telemetry.LookupFinished(failure); + } + return new(versions, failure); } async Task CompleteAsync(ResolutionWork work) @@ -101,6 +115,8 @@ async Task CompleteAsync(ResolutionWork work) var failure = result.Failure; if (result.Versions is { Count: > 0 }) { + telemetry.Work("publication", work.Item.Title, episode.Key); + var publishTimer = Stopwatch.StartNew(); try { episode.Paths = await _inventory.PublishAsync(work.Item, episode, result.Versions, token); @@ -114,9 +130,11 @@ async Task CompleteAsync(ResolutionWork work) await RegisterPathsAsync(live, work.Item, episode, result.Versions, token); published++; if (work.Upgrade) refreshed++; + telemetry.Published(work.Upgrade); } catch (OperationCanceledException) when (token.IsCancellationRequested) { throw; } - catch { failure = "publication_failed"; } + catch { failure = "publication_failed"; telemetry.PublicationFailed(); } + finally { telemetry.RecordTiming("publication", publishTimer.Elapsed.TotalSeconds); } } if (failure.Length > 0) { @@ -185,6 +203,7 @@ async Task DispatchAsync(ResolutionCandidate candidate) if (!candidate.Upgrade) episode.InitialFailure = true; episode.Attempts++; await _db.RecordImportAttemptAsync(lease, now, token); + telemetry.Attempt(candidate.Upgrade); await _db.SaveImportCoverageAsync(live, token); var index = candidate.Coverage.Items.FindIndex(x => x.Key == episode.Key); if (index >= 0) candidate.Coverage.Items[index] = episode; @@ -237,6 +256,7 @@ async Task DispatchAsync(ResolutionCandidate candidate) if (_mode() == ImportMode.Off) break; try { + telemetry.Work("inventory", item.Title); var aliases = ImportInventory.Aliases(item); if (aliases.Count == 0) continue; var identity = await _db.FindImportIdentityAsync(aliases, token) ?? aliases[0]; @@ -276,6 +296,8 @@ async Task DispatchAsync(ResolutionCandidate candidate) { metadata++; coverage.InventoryPolicyVersion = 1; + telemetry.Work("metadata", item.Title); + var metadataTimer = Stopwatch.StartNew(); try { using var deadline = CancellationTokenSource.CreateLinkedTokenSource(token); @@ -295,6 +317,7 @@ async Task DispatchAsync(ResolutionCandidate candidate) coverage.SnapshotStatus = "stale_or_unavailable"; coverage.MetadataRetryAt = now.AddMinutes(15); } + finally { telemetry.RecordTiming("metadata", metadataTimer.Elapsed.TotalSeconds); } } // Existing source inventories are an explicitly stale migration baseline, // useful for observing files, never sufficient for new automatic imports. @@ -312,11 +335,14 @@ async Task DispatchAsync(ResolutionCandidate candidate) await DrainCompletedAsync(); token.ThrowIfCancellationRequested(); now = _clock(); + telemetry.Work("observation", item.Title, episode.Key); ImportObservation observation; + var observationTimer = Stopwatch.StartNew(); try { observation = await _inventory.ObserveAsync(item, episode, token); } catch (OperationCanceledException) { throw; } catch { episode.State = "observation_unavailable"; episode.ObservedAt = now; coverage.Cursor = episode.Key; await SaveObservedAsync(coverage, token); continue; } + finally { telemetry.Checked(); telemetry.RecordTiming("observation", observationTimer.Elapsed.TotalSeconds); } var file = observation.Paths.Count > 0; if (file) episode.EverPublished = true; episode.Paths = observation.Paths; @@ -382,11 +408,13 @@ async Task DispatchAsync(ResolutionCandidate candidate) await _db.PersistMetadataAsync(cursorKey, sourceItem.Id, token); } } + telemetry.Work("borrowed_capacity"); foreach (var candidate in deferred) { await DrainCompletedAsync(); await DispatchAsync(candidate); } + telemetry.Work("waiting_for_sources"); while (pending.Count > 0) await DrainOneAsync(); } catch (OperationCanceledException) { status = ct.IsCancellationRequested ? "cancelled" : "budget_deferred"; } @@ -396,12 +424,15 @@ async Task DispatchAsync(ResolutionCandidate candidate) // Cancellation ends all lookups before another run can acquire RunGate. budget.Cancel(); await Task.WhenAll(pending.Select(x => x.Resolution)); - try { await _db.PersistMetadataAsync("import_last_run", System.Text.Json.JsonSerializer.Serialize(new + telemetry.Finish(status); + try { var report = System.Text.Json.JsonSerializer.Serialize(new { Id = runId, Status = status, FinishedAt = _clock(), Scanned = scanned, Attempts = attempts, Metadata = metadata, Published = published, Upgrades = upgrades, Refreshed = refreshed, SourceFailures = sourceFailures, TransportFailures = transportFailures, Priority = allowance.IsCatchUp ? priority : "normal", ElapsedSeconds = elapsed.Elapsed.TotalSeconds, Speed = allowance.IsCatchUp ? "catch_up" : "normal", allowance.AttemptsPerDay, allowance.Parallelism, - allowance.CatchUpStartedAt, allowance.CatchUpUntil }), CancellationToken.None); } + allowance.CatchUpStartedAt, allowance.CatchUpUntil, Analytics = telemetry.Snapshot() }); + await _db.PersistMetadataAsync("import_last_run", report, CancellationToken.None); + await _db.SaveImportRunReportAsync(runId, DateTimeOffset.UtcNow, report, CancellationToken.None); } finally { RunGate.Release(); } } } @@ -435,6 +466,7 @@ private async Task IsAuthorizedAsync(ImportCoverage coverage, CatalogItem private async Task SaveObservedAsync(ImportCoverage state, CancellationToken ct) { + var timer = Stopwatch.StartNew(); await MutationGate.WaitAsync(ct); try { @@ -446,7 +478,7 @@ private async Task SaveObservedAsync(ImportCoverage state, CancellationToken ct) } await _db.SaveImportCoverageAsync(state, ct); } - finally { MutationGate.Release(); } + finally { MutationGate.Release(); _activeTelemetry?.RecordTiming("checkpoint", timer.Elapsed.TotalSeconds); } } private async Task RegisterPathsAsync(ImportCoverage state, CatalogItem item, ImportEpisode episode, List versions, CancellationToken ct) diff --git a/Tests/ImportAnalyticsTests.cs b/Tests/ImportAnalyticsTests.cs new file mode 100644 index 0000000..5344442 --- /dev/null +++ b/Tests/ImportAnalyticsTests.cs @@ -0,0 +1,75 @@ +using System; +using System.IO; +using System.Linq; +using System.Text.Json; +using System.Threading.Tasks; +using InfiniteDrive.Data; +using InfiniteDrive.Models; +using Microsoft.Extensions.Logging.Abstractions; +using Xunit; + +namespace InfiniteDrive.Tests; +public sealed class ImportAnalyticsTests +{ + static ImportAnalyticsTests() => SqliteTestRuntime.EnsureInitialized(); + [Fact] public async Task ConcurrentLookupsKeepAccurateCountersAndBoundedTimingSamples() + { + var telemetry = new ImportRunTelemetry("test"); + await Task.WhenAll(Enumerable.Range(0, 1000).Select(i => Task.Run(() => + { + telemetry.Attempt(i % 2 == 0); telemetry.LookupStarted(); + telemetry.RecordTiming("resolution", i); telemetry.LookupFinished(i % 2 == 0 ? "" : "source_unavailable"); + }))); + var run = telemetry.Snapshot(); var timing = run.Timings["resolution"]; + Assert.Equal(0, run.ActiveLookups); Assert.Equal(500, run.MissingAttempts); + Assert.Equal(500, run.RefreshAttempts); Assert.Equal(500, run.Matched); Assert.Equal(500, run.EmptyResults); + Assert.Equal(1000, timing.Count); Assert.Equal(512, timing.SampleCount); + Assert.Equal(499500, timing.TotalSeconds); Assert.Equal(499.5, timing.MeanSeconds); Assert.Equal(999, timing.MaxSeconds); + } + [Fact] public void LookupMatchesAndPublishedGroupsAreDifferentUnits() + { + var t = new ImportRunTelemetry("test"); t.Work("metadata", "A title", "aired:1:1"); + for (var i = 1; i <= 20; i++) t.RecordTiming("metadata", i); + t.LookupStarted(); t.LookupFinished(""); t.PublicationFailed(); + var before = t.Snapshot(); Assert.Equal(1, before.Matched); Assert.Equal(0, before.Published); + Assert.Equal(1, before.PublicationFailures); Assert.Equal(19, before.Timings["metadata"].P95Seconds); + t.Published(true); t.Finish("budget_deferred"); + var done = t.Snapshot(); Assert.NotNull(done.FinishedAt); Assert.Equal("budget_deferred", done.Status); + Assert.Equal(1, done.Published); Assert.Equal(1, done.Refreshed); Assert.Equal("finished", done.Phase); + Assert.DoesNotContain("https://", JsonSerializer.Serialize(done)); + Assert.Throws(() => t.RecordTiming("https://private.invalid/token", 1)); + } + [Fact] public async Task ReadingOldHistoryCreatesNoSchemaAndNewReportsSurviveReopenWithoutResettingLedger() + { + var root = Path.Combine(Path.GetTempPath(), "analytics-"+Guid.NewGuid().ToString("N")); Directory.CreateDirectory(root); + try + { + var db = new DatabaseManager(root, NullLogger.Instance); db.Initialise(); + Assert.Empty(await db.GetImportRunHistoryAsync()); + await db.EnsureImportCoverageAsync(); var now = DateTimeOffset.UtcNow; + await db.RecordImportAttemptAsync("keep", now, default); + await db.SaveImportRunReportAsync("old", now.AddDays(-8), "old", default); + await db.SaveImportRunReportAsync("current", now, "current", default); + db = new DatabaseManager(root, NullLogger.Instance); db.Initialise(); + Assert.Equal(new[] { "current" }, await db.GetImportRunHistoryAsync()); + Assert.Equal(1, await db.GetRecentImportAttemptsAsync(now)); + } + finally { Directory.Delete(root, true); } + } + [Fact] public async Task NextCreditUsesTheCorrectExpiringAttemptWhenReturningToALowerCeiling() + { + var root = Path.Combine(Path.GetTempPath(), "credit-"+Guid.NewGuid().ToString("N")); Directory.CreateDirectory(root); + try + { + var db = new DatabaseManager(root, NullLogger.Instance); db.Initialise(); await db.EnsureImportCoverageAsync(); + var now = DateTimeOffset.Parse("2026-09-30T20:00:00Z"); + await db.RecordImportAttemptAsync("first", now.AddHours(-2), default); + await db.RecordImportAttemptAsync("second", now.AddHours(-1), default); + await db.RecordImportAttemptAsync("third", now, default); + Assert.Equal(now.AddHours(23), await db.GetNextImportCreditAsync(now, 2)); + Assert.Equal(now.AddHours(22), await db.GetNextImportCreditAsync(now, 3)); + Assert.Null(await db.GetNextImportCreditAsync(now, 4)); + } + finally { Directory.Delete(root, true); } + } +} diff --git a/Tests/ImportReconciliationTests.cs b/Tests/ImportReconciliationTests.cs index 070fc79..723ee99 100644 --- a/Tests/ImportReconciliationTests.cs +++ b/Tests/ImportReconciliationTests.cs @@ -601,6 +601,25 @@ [Fact] public async Task NonPriorityLaneBorrowsUnusedCreditsAndRechecksALateBloc Assert.Equal(previous, h.Inventory.Resolutions); } + [Fact] public async Task AnalyticsSeparatesExplicitHttp429FromEmptySourcesAndKeepsBackoff() + { + using var h = new Harness(); h.Inventory.Count = 1; h.Inventory.RateLimit = true; + await h.Seed(); await h.Run(); var run = ImportRunTelemetry.Current!; + Assert.Equal(1, run.Http429); Assert.Equal(0, run.EmptyResults); + Assert.Equal(0, run.TransportFailures); Assert.Equal(0, run.Published); + Assert.Equal(0, run.ActiveLookups); Assert.Equal(1, run.MissingAttempts); + var episode = (await h.State()).Items.Single(); Assert.Equal("http_429", episode.Failure); + Assert.True(episode.NextAttempt > h.Now); Assert.Single(await h.Db.GetImportRunHistoryAsync()); + } + [Fact] public async Task AnalyticsDoesNotCallAnUncancelledHttpOperationTimeoutALookupDeadline() + { + using var h = new Harness(); h.Inventory.Count = 1; + h.Inventory.OnResolve = () => throw new OperationCanceledException(); + await h.Seed(); await h.Run(); var run = ImportRunTelemetry.Current!; + Assert.Equal(1, run.TransportFailures); Assert.Equal(0, run.LookupDeadlines); + Assert.Equal(0, run.ActiveLookups); Assert.Equal("transport_failure", (await h.State()).Items.Single().Failure); + } + public class NullLogProxy : System.Reflection.DispatchProxy { protected override object? Invoke(System.Reflection.MethodInfo? method, object?[]? args) @@ -628,7 +647,7 @@ private sealed class FakeInventory : IImportInventory { public int Resolutions, Published, Notifications, FailEpisode, DisputedEpisode, Count = 2; public List ResolvedEpisodes = new(); - public bool Owned, BadSnapshot, Paused, Indexed = true; public bool ProviderPaused => Paused; public HashSet Files = new(); public Func? OnResolve; public Func? OnResolveWithToken; public Func? OnObserve; + public bool Owned, BadSnapshot, Paused, RateLimit, Indexed = true; public bool ProviderPaused => Paused; public HashSet Files = new(); public Func? OnResolve; public Func? OnResolveWithToken; public Func? OnObserve; public Task FetchAsync(CatalogItem item, CancellationToken ct) { var episodes = BadSnapshot ? new List() : Enumerable.Range(1, Count).Select(Ep).ToList(); @@ -642,7 +661,7 @@ public async Task ObserveAsync(CatalogItem item, ImportEpisod } public bool IsOwned(CatalogItem item) => Owned; public async Task> ResolveAsync(CatalogItem item, ImportEpisode ep, CancellationToken ct) - { Resolutions++; ResolvedEpisodes.Add(ep.Episode!.Value); if (OnResolve != null) await OnResolve(); if (OnResolveWithToken != null) await OnResolveWithToken(ct); return ep.Episode == FailEpisode ? new() : new() { new() { Stream = new() { Url = "https://example.invalid/test" } } }; } + { Resolutions++; ResolvedEpisodes.Add(ep.Episode!.Value); if (RateLimit) throw new ImportHttpRateLimitException(); if (OnResolve != null) await OnResolve(); if (OnResolveWithToken != null) await OnResolveWithToken(ct); return ep.Episode == FailEpisode ? new() : new() { new() { Stream = new() { Url = "https://example.invalid/test" } } }; } public Task> PublishAsync(CatalogItem item, ImportEpisode ep, List versions, CancellationToken ct) { Published++; Files.Add(ep.Episode!.Value); return Task.FromResult(new List { $"/fake/series/Season 01/e{ep.Episode}.strm" }); } public void Notify(CatalogItem item) => Notifications++; diff --git a/UI/Settings/SyncAndMarvinTabView.cs b/UI/Settings/SyncAndMarvinTabView.cs index 6aad61c..ca0616a 100644 --- a/UI/Settings/SyncAndMarvinTabView.cs +++ b/UI/Settings/SyncAndMarvinTabView.cs @@ -6,6 +6,7 @@ using ItemStatus = Emby.Web.GenericEdit.Elements.ItemStatus; using Emby.Web.GenericEdit.Elements.List; using System.Threading.Tasks; +using System.Threading; using Emby.Web.GenericEdit.Elements; using InfiniteDrive.UI; using MediaBrowser.Model.Plugins.UI.Views; @@ -23,6 +24,7 @@ public SyncAndMarvinTabView(string pluginId, SyncAndMarvinUI ui) } private int _importOffset; + private readonly SemaphoreSlim _loadGate = new(1, 1); private SyncAndMarvinUI UI => (SyncAndMarvinUI)ContentData; public override bool IsCommandAllowed(string commandKey) => true; @@ -84,6 +86,7 @@ public override async Task RunCommand(string itemId, string comma private async Task LoadImportsAsync() { + await _loadGate.WaitAsync(); try { var db = Plugin.Instance.DatabaseManager; @@ -113,20 +116,46 @@ private async Task LoadImportsAsync() CommandId = speed.IsCatchUp ? "ImportNormalSpeed" : "ImportCatchUp", }, }); + var hasLedger = db.GetMetadata("import_schema") == "1"; + var used = hasLedger ? await db.GetRecentImportAttemptsAsync(now) : 0; + var nextCredit = hasLedger ? await db.GetNextImportCreditAsync(now, speed.AttemptsPerDay) : null; + UI.DashboardItems.Clear(); UI.RecentRuns.Clear(); + AddDashboard("Read at", $"{now:yyyy-MM-dd HH:mm:ss} UTC · Refresh dashboard to update"); + AddDashboard("Rolling 24-hour allowance", $"{used:N0} / {speed.AttemptsPerDay:N0} attempts · {Math.Max(0, speed.AttemptsPerDay-used):N0} available" + + (nextCredit.HasValue ? $" · earliest credit {When(nextCredit, now)}" : "")); + var current = ImportRunTelemetry.Current; var lastRun = db.GetMetadata("import_last_run"); + if (current != null) DrawRun(current, current.Status == "running" ? "Current run" : "Latest run in this server session"); + else if (lastRun != null) + { + using var saved = JsonDocument.Parse(lastRun); + if (saved.RootElement.TryGetProperty("Analytics", out var analytics)) + { + var restored = analytics.Deserialize(); + if (restored != null) DrawRun(restored, "Last saved run (before this server restart)"); + } + else AddDashboard("Current run", "No timing sample yet. Older reports have counts only."); + } + foreach (var payload in await db.GetImportRunHistoryAsync()) + { + using var document = JsonDocument.Parse(payload); var row = document.RootElement; + UI.RecentRuns.Add(new GenericListItem { + PrimaryText = $"{row.GetProperty("FinishedAt").GetDateTimeOffset():HH:mm:ss} UTC · {row.GetProperty("Status").GetString()}", + SecondaryText = $"{row.GetProperty("ElapsedSeconds").GetDouble():N0}s · {row.GetProperty("Attempts").GetInt32():N0} attempts · {row.GetProperty("Published").GetInt32():N0} groups written" }); + } if (lastRun == null) { UI.MarvinStatus.StatusText = "No report yet. Run Marvin to check the library."; UI.MarvinStatus.Status = ItemStatus.None; return; } - UI.ImportStatus.StatusText += $" · {await db.GetRecentImportAttemptsAsync(now):N0} of {speed.AttemptsPerDay:N0} stream checks used in the last 24 hours"; + UI.ImportStatus.StatusText += $" · {used:N0} of {speed.AttemptsPerDay:N0} stream checks used in the last 24 hours"; using (var report = JsonDocument.Parse(lastRun)) { var run = report.RootElement; var finished = run.GetProperty("FinishedAt").GetDateTimeOffset(); var summary = $"{When(finished, now)} · {run.GetProperty("Attempts").GetInt32()} stream checks"; - if (run.TryGetProperty("Published", out var published)) summary += $" · {published.GetInt32()} items refreshed"; + if (run.TryGetProperty("Published", out var published)) summary += $" · {published.GetInt32()} groups written"; var status = run.GetProperty("Status").GetString(); UI.MarvinStatus.StatusText = summary + (status switch { @@ -148,16 +177,16 @@ private async Task LoadImportsAsync() var detail = (expected > 0 ? $"{indexed}/{expected} released items in Emby" : "No released items confirmed") + $" · checked {When(title.CheckedAt, now)}"; if (title.Exclusion.Length > 0) detail += " · " + Explain(title.Exclusion); else if (title.SnapshotStatus != "success") detail += " · " + Explain(title.SnapshotStatus); - if (title.ProviderStatus is "provider_check_unavailable" or "numbering_conflict") + if (title.ProviderStatus is "provider_check_unavailable" or "numbering_conflict" or "partial_numbering") detail += " · " + Explain(title.ProviderStatus); UI.ImportItems.Add(new GenericListItem { PrimaryText = title.Title, SecondaryText = detail, Icon = IconNames.info, IconMode = ItemListIconMode.SmallRegular, Button1 = new ButtonItem(title.IncludeSpecials ? "Skip specials" : "Include released specials") { Data1 = title.Identity, CommandId = title.IncludeSpecials ? "ImportExcludeSpecials" : "ImportIncludeSpecials" } }); - foreach (var episode in title.Items.Where(x => x.State != "indexed").Take(50)) + foreach (var episode in title.Items.Where(x => x.State != "indexed" || x.Failure.Length > 0).Take(50)) { - var reason = episode.State == "excluded" ? Explain(episode.Eligibility) : Explain(episode.State); + var reason = episode.State == "indexed" && episode.Failure.Length > 0 ? "In Emby; source refresh failed" : episode.State == "excluded" ? Explain(episode.Eligibility) : Explain(episode.State); var more = episode.Failure.Length > 0 ? Explain(episode.Failure) : ""; if (episode.NextAttempt.HasValue) more += (more.Length > 0 ? " · " : "") + (episode.NextAttempt > now ? "Retry " + When(episode.NextAttempt, now) : "Retry due"); UI.ImportItems.Add(new GenericListItem @@ -173,7 +202,26 @@ private async Task LoadImportsAsync() UI.ImportStatus.StatusText = "Couldn't load the library checks. Try Refresh status."; UI.ImportStatus.Status = ItemStatus.Warning; } - finally { RaiseUIViewInfoChanged(); } + finally { _loadGate.Release(); RaiseUIViewInfoChanged(); } + } + + private void AddDashboard(string title, string detail) => + UI.DashboardItems.Add(new GenericListItem { PrimaryText = title, SecondaryText = detail }); + + private void DrawRun(ImportRunSnapshot run, string label) + { + AddDashboard(label, $"{run.Status} · {run.ElapsedSeconds:N0}s elapsed · started {run.StartedAt:HH:mm:ss} UTC"); + if (run.Status == "running") AddDashboard("Doing", $"{run.Phase.Replace('_',' ')} · {run.PhaseSeconds:N0}s in this step" + + (run.Title.Length > 0 ? $" · {run.Title} {run.EpisodeKey}" : "")); + AddDashboard("Lookups", $"{run.ActiveLookups:N0} in flight (includes paced waiting) · {run.MissingAttempts:N0} missing-file attempts · {run.RefreshAttempts:N0} refresh attempts"); + AddDashboard("Results", $"{run.Matched:N0} matched · {run.EmptyResults:N0} empty · {run.TransportFailures:N0} transport failures · {run.LookupDeadlines:N0} deadlines · {run.Http429:N0} HTTP 429 · {run.ProviderConfigurationFailures:N0} provider settings failures · {run.CancelledLookups:N0} cancelled"); + AddDashboard("Files published", $"{run.Published:N0} movie/episode groups written · {run.Refreshed:N0} replaced old choices · {run.PublicationFailures:N0} publication failures · {run.EpisodesChecked:N0} episode/movie checks"); + foreach (var entry in run.Timings) + { + var timing = entry.Value; + AddDashboard(entry.Key + " timing", $"{timing.Count:N0} calls · mean {timing.MeanSeconds:N2}s · p95 {timing.P95Seconds:N2}s · max {timing.MaxSeconds:N2}s · total {timing.TotalSeconds:N1}s"); + } + AddDashboard("Timing scope", "Resolution includes dispatch pacing and provider wait. Parallel call totals overlap; they are not wall-clock time. p95 uses the latest 512 samples per stage; other statistics cover this run. Checkpoint timing covers observation saves only."); } private static string When(DateTimeOffset? value, DateTimeOffset now) @@ -206,6 +254,9 @@ private static string When(DateTimeOffset? value, DateTimeOffset now) "stale_or_unavailable" => "Couldn't refresh the episode list", "provider_check_unavailable" => "Emby's episode metadata is unavailable", "source_unavailable" => "No streams matched your settings", + "partial_numbering" => "Some episode numbers need a check; confirmed siblings can proceed", + "lookup_deadline" => "Source lookup reached its deadline", + "http_429" => "Source returned HTTP 429; waiting to retry", "transport_failure" => "Couldn't reach the source", "provider_configuration" => "Check your provider settings, then retry", _ => "Needs another check", diff --git a/UI/Settings/SyncAndMarvinUI.cs b/UI/Settings/SyncAndMarvinUI.cs index 97a7cc2..458014f 100644 --- a/UI/Settings/SyncAndMarvinUI.cs +++ b/UI/Settings/SyncAndMarvinUI.cs @@ -11,6 +11,12 @@ public class SyncAndMarvinUI : EditableOptionsBase public override string EditorTitle => "Marvin"; public override string EditorDescription => "Brain the size of a planet. Still on library duty."; + public CaptionItem DashboardCaption { get; set; } = new CaptionItem("Marvin dashboard"); + public LabelItem DashboardHelp { get; set; } = new LabelItem("A dated snapshot of the worker. Refresh to update. Writes are movie/episode groups, not alternate files or proof of playback."); + public ButtonItem RefreshDashboard { get; set; } = new ButtonItem("Refresh dashboard") { Data1 = "ImportRefresh" }; + public GenericItemList DashboardItems { get; set; } = new GenericItemList(); + public CaptionItem HistoryCaption { get; set; } = new CaptionItem("Recent passes"); + public GenericItemList RecentRuns { get; set; } = new GenericItemList(); public StatusItem MarvinStatus { get; set; } = new StatusItem("Last library pass", "No report yet", ItemStatus.None); public ButtonItem RunMarvinNowButton { get; set; } = new ButtonItem("Run Marvin now") { diff --git a/docs/import-reconciliation-qa.md b/docs/import-reconciliation-qa.md index f7ccb27..cd77a9e 100644 --- a/docs/import-reconciliation-qa.md +++ b/docs/import-reconciliation-qa.md @@ -3,7 +3,7 @@ ## Episode numbering and fair reconciliation — September 30, 2026 -Version 0.42.11: **213/213 tests passed**, zero skipped, against pinned Emby +Version 0.42.11: **219/219 tests passed**, zero skipped, against pinned Emby 4.10.0.40 and real SQLite. New regressions check per-key source/provider agreement, duplicate and unverified numbering, legacy whole-series gate reevaluation, identity safeguards, failed-refresh diagnostics/backoff across a diff --git a/docs/import-reconciliation.md b/docs/import-reconciliation.md index 0d83410..f44bc6a 100644 --- a/docs/import-reconciliation.md +++ b/docs/import-reconciliation.md @@ -214,6 +214,35 @@ thirty minutes apart. After those attempts and two hours, unresolved native indexing becomes `indexing_attention`. Large or blocked Emby scans can therefore leave publication pending; the worker does not repeatedly download/rewrite it. +## Dashboard and timing history + +The native Marvin page has a **Refresh dashboard** button. It reads a timestamped +snapshot, including the current phase/title/episode, elapsed time, lookups in +flight (including paced waiting), episode/movie checks, missing and refresh +attempts, source matches, empty responses, transport errors, lookup deadlines, +explicit HTTP 429, provider configuration failures, cancellations and successful +publication groups. Matched lookups can still fail or be discarded before +publication. Groups are movies/episodes, not individual alternate STRMs, grouped +Emby titles or proof of playback. Old indexed files with failed refreshes retain +their reason and next retry in the coverage list. + +Five stage timers measure metadata, observation, resolution, publication and +observation checkpoint saves. Resolution includes pacing/provider wait; publication +includes file writes and publication state/registration. Checkpoint timings exclude +other database operations. Parallel call totals overlap and are not wall-clock time. +Each stage retains at most 512 recent samples for p95; count/total/mean/max cover the +whole run. The live snapshot is process memory; no per-event database polling or +new worker is added. Refreshing the page is read-only and does not trigger work. + +The additive `import_run_history` table retains seven days of completed reports +(one write per run plus pruning old analytics). `import_last_run` stays compatible +and gains an `Analytics` object. No history is invented for old versions. History +pruning never resets the attempt ledger, coverage, retry/backoff or catalog/user +state. After restart the dashboard labels the prior saved run explicitly. The +next-credit estimate considers the correct expiring attempt when a lower normal +ceiling follows catch-up; it is a quota estimate, not a source-availability or +completion estimate. + ## Administrative API Both routes require a native administrator session through the existing @@ -221,8 +250,9 @@ Both routes require a native administrator session through the existing - `GET /InfiniteDrive/Imports?Offset=0&Limit=50`: read-only, bounded page; dated coverage, provider status, episode reasons, next retries, collection summary, - speed, expiry and rolling attempt usage. LastRun includes elapsed seconds and - publication/upgrade counts; they are not playback or completion claims. + speed, expiry and rolling attempt usage. CurrentRun adds the in-memory snapshot, NextCreditAt the earliest budget credit, + and RecentRuns up to 12 saved reports. LastRun includes elapsed seconds and + publication/upgrade counts and an Analytics snapshot; they are not playback or completion claims. - `POST /InfiniteDrive/Imports/Action`: `Action` is `mode`, `check`, `retry`, `include_specials`, `exclude_specials`, `resume_provider`, `start_catch_up`, or `stop_catch_up`. Catch-up actions need no extra fields; start requires Repair. @@ -235,11 +265,13 @@ status never creates tables, contacts providers, or triggers work. ## Verification and rollout -September 30, 2026, 0.42.11 regression build: 213/213 tests passed against +September 30, 2026, 0.42.11 regression build: 219/219 tests passed against pinned Emby 4.10.0.40, with no skipped tests. New cases cover partial/disjoint provider inventories, duplicate source/provider numbering, unverified anime, provider-only additions, legacy gate reevaluation, genuine identity conflict, -failed refresh diagnostics across restart, priority rotation and late blocks. +failed refresh diagnostics across restart, priority rotation and late blocks. Analytics regressions cover concurrent counters, +bounded timing samples, HTTP 429 versus operation timeouts, report retention across +restart, and credit expiry under a lower ceiling. This is implementation evidence; production throughput and playback require separate live checks. diff --git a/tools/ImportQa/ImportLabService.cs b/tools/ImportQa/ImportLabService.cs index 84c4fd3..2a5c51f 100644 --- a/tools/ImportQa/ImportLabService.cs +++ b/tools/ImportQa/ImportLabService.cs @@ -106,6 +106,10 @@ public async Task Post(ImportQaRequest request) await db.PersistMetadataAsync("qa_collection", boxes.Contains(box, seriesItem.Id) ? "confirmed" : "failed"); mode = ImportMode.Observe; break; + case "dashboard": + var view = new InfiniteDrive.UI.Settings.SyncAndMarvinTabView(p.Id.ToString(), new()); + await view.RunCommand("", "ImportRefresh", ""); + return new { Dashboard = view.ContentData, CurrentRun = ImportRunTelemetry.Current }; case "status": return await Status(db, fake); default: throw new ArgumentException("Unknown QA step"); } From a05d32bab9759dd3806e1b9cfda04e32c4788e62 Mon Sep 17 00:00:00 2001 From: OneHotTake Date: Wed, 30 Sep 2026 17:35:22 -0500 Subject: [PATCH 3/3] Polish native dashboard layout and distinguish worker phases from repair reports --- Services/Api/ImportHealthService.cs | 4 ++-- Tasks/MarvinTask.cs | 6 +++++ UI/Settings/SyncAndMarvinTabView.cs | 35 ++++++++++++++++++++--------- UI/Settings/SyncAndMarvinUI.cs | 7 ++++-- docs/import-reconciliation.md | 5 ++++- 5 files changed, 42 insertions(+), 15 deletions(-) diff --git a/Services/Api/ImportHealthService.cs b/Services/Api/ImportHealthService.cs index f1fddb8..466ef64 100644 --- a/Services/Api/ImportHealthService.cs +++ b/Services/Api/ImportHealthService.cs @@ -36,13 +36,13 @@ public async Task 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(), CurrentRun = ImportRunTelemetry.Current }; + 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, - CurrentRun = ImportRunTelemetry.Current, NextCreditAt = await db.GetNextImportCreditAsync(DateTimeOffset.UtcNow, speed.AttemptsPerDay), + 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, diff --git a/Tasks/MarvinTask.cs b/Tasks/MarvinTask.cs index e4b6488..a50fb84 100644 --- a/Tasks/MarvinTask.cs +++ b/Tasks/MarvinTask.cs @@ -99,10 +99,12 @@ public async Task Execute(CancellationToken cancellationToken, IProgress await Plugin.SyncLock.WaitAsync(cancellationToken); try { + Plugin.Pipeline.SetPhase("Marvin", "ImportReconciliation"); await ExecuteInternalAsync(cancellationToken, progress); } finally { + Plugin.Pipeline.Clear(); Plugin.SyncLock.Release(); } } @@ -220,19 +222,23 @@ private async Task ExecuteInternalAsync(CancellationToken cancellationToken, IPr } phaseSw.Restart(); + Plugin.Pipeline.SetPhase("Marvin", "Validation"); await ValidationPassAsync(cancellationToken); _logger.LogDebug("[Marvin] Phase 4a (Validation) completed in {Ms}ms", phaseSw.ElapsedMilliseconds); phaseSw.Restart(); + Plugin.Pipeline.SetPhase("Marvin", "PhysicalMediaReconciliation"); var readoption = new LibraryPostScanReadoptionService(_libraryManager, _logManager); await readoption.Run(new Progress(), cancellationToken); _logger.LogDebug("[Marvin] Phase 4b (PhysicalMediaReconciliation) completed in {Ms}ms", phaseSw.ElapsedMilliseconds); phaseSw.Restart(); + Plugin.Pipeline.SetPhase("Marvin", "Enrichment"); await EnrichmentTrickleAsync(cancellationToken); _logger.LogDebug("[Marvin] Phase 4c (Enrichment) completed in {Ms}ms", phaseSw.ElapsedMilliseconds); phaseSw.Restart(); + Plugin.Pipeline.SetPhase("Marvin", "TokenRenewal"); await TokenRenewalAsync(cancellationToken); _logger.LogDebug("[Marvin] Phase 4d (TokenRenewal) completed in {Ms}ms", phaseSw.ElapsedMilliseconds); progress?.Report(0.90); diff --git a/UI/Settings/SyncAndMarvinTabView.cs b/UI/Settings/SyncAndMarvinTabView.cs index ca0616a..fe1046c 100644 --- a/UI/Settings/SyncAndMarvinTabView.cs +++ b/UI/Settings/SyncAndMarvinTabView.cs @@ -119,28 +119,31 @@ private async Task LoadImportsAsync() var hasLedger = db.GetMetadata("import_schema") == "1"; var used = hasLedger ? await db.GetRecentImportAttemptsAsync(now) : 0; var nextCredit = hasLedger ? await db.GetNextImportCreditAsync(now, speed.AttemptsPerDay) : null; - UI.DashboardItems.Clear(); UI.RecentRuns.Clear(); + UI.DashboardItems.Clear(); UI.TimingItems.Clear(); UI.RecentRuns.Clear(); + var phase = Plugin.Pipeline.Current; + AddDashboard("Worker", phase == null ? "Between scheduled passes" : $"{phase.TaskName} · {PhaseLabel(phase.PhaseName)} · {Math.Max(0,(now-phase.StartedAt).TotalSeconds):N0}s in this step"); AddDashboard("Read at", $"{now:yyyy-MM-dd HH:mm:ss} UTC · Refresh dashboard to update"); AddDashboard("Rolling 24-hour allowance", $"{used:N0} / {speed.AttemptsPerDay:N0} attempts · {Math.Max(0, speed.AttemptsPerDay-used):N0} available" + (nextCredit.HasValue ? $" · earliest credit {When(nextCredit, now)}" : "")); var current = ImportRunTelemetry.Current; var lastRun = db.GetMetadata("import_last_run"); - if (current != null) DrawRun(current, current.Status == "running" ? "Current run" : "Latest run in this server session"); + if (current != null) DrawRun(current, current.Status == "running" ? "Current repair pass" : "Latest repair pass"); else if (lastRun != null) { using var saved = JsonDocument.Parse(lastRun); if (saved.RootElement.TryGetProperty("Analytics", out var analytics)) { var restored = analytics.Deserialize(); - if (restored != null) DrawRun(restored, "Last saved run (before this server restart)"); + if (restored != null) DrawRun(restored, "Last saved repair pass"); } - else AddDashboard("Current run", "No timing sample yet. Older reports have counts only."); + else AddDashboard("Repair pass", "No timing sample yet. Older reports have counts only."); } foreach (var payload in await db.GetImportRunHistoryAsync()) { using var document = JsonDocument.Parse(payload); var row = document.RootElement; UI.RecentRuns.Add(new GenericListItem { - PrimaryText = $"{row.GetProperty("FinishedAt").GetDateTimeOffset():HH:mm:ss} UTC · {row.GetProperty("Status").GetString()}", + Icon = IconNames.history, IconMode = ItemListIconMode.SmallRegular, Status = ItemStatus.None, + PrimaryText = $"{row.GetProperty("FinishedAt").GetDateTimeOffset():MMM dd HH:mm:ss} UTC · {row.GetProperty("Status").GetString()}", SecondaryText = $"{row.GetProperty("ElapsedSeconds").GetDouble():N0}s · {row.GetProperty("Attempts").GetInt32():N0} attempts · {row.GetProperty("Published").GetInt32():N0} groups written" }); } if (lastRun == null) @@ -192,7 +195,7 @@ private async Task LoadImportsAsync() UI.ImportItems.Add(new GenericListItem { PrimaryText = (episode.Season.HasValue ? $"S{episode.Season:D2}E{episode.Episode:D2}" : "Movie") + " · " + reason, - SecondaryText = more, + SecondaryText = more, Icon = IconNames.info, IconMode = ItemListIconMode.SmallRegular, Status = ItemStatus.None, }); } } @@ -206,7 +209,7 @@ private async Task LoadImportsAsync() } private void AddDashboard(string title, string detail) => - UI.DashboardItems.Add(new GenericListItem { PrimaryText = title, SecondaryText = detail }); + UI.DashboardItems.Add(new GenericListItem { PrimaryText = title, SecondaryText = detail, Icon = IconNames.info, IconMode = ItemListIconMode.SmallRegular, Status = ItemStatus.None }); private void DrawRun(ImportRunSnapshot run, string label) { @@ -214,16 +217,28 @@ private void DrawRun(ImportRunSnapshot run, string label) if (run.Status == "running") AddDashboard("Doing", $"{run.Phase.Replace('_',' ')} · {run.PhaseSeconds:N0}s in this step" + (run.Title.Length > 0 ? $" · {run.Title} {run.EpisodeKey}" : "")); AddDashboard("Lookups", $"{run.ActiveLookups:N0} in flight (includes paced waiting) · {run.MissingAttempts:N0} missing-file attempts · {run.RefreshAttempts:N0} refresh attempts"); - AddDashboard("Results", $"{run.Matched:N0} matched · {run.EmptyResults:N0} empty · {run.TransportFailures:N0} transport failures · {run.LookupDeadlines:N0} deadlines · {run.Http429:N0} HTTP 429 · {run.ProviderConfigurationFailures:N0} provider settings failures · {run.CancelledLookups:N0} cancelled"); + AddDashboard("Source results", $"{run.Matched:N0} matched · {run.EmptyResults:N0} empty · {run.CancelledLookups:N0} cancelled"); + AddDashboard("Source failures", $"{run.TransportFailures:N0} transport · {run.LookupDeadlines:N0} lookup deadlines · {run.Http429:N0} HTTP 429 · {run.ProviderConfigurationFailures:N0} provider settings"); AddDashboard("Files published", $"{run.Published:N0} movie/episode groups written · {run.Refreshed:N0} replaced old choices · {run.PublicationFailures:N0} publication failures · {run.EpisodesChecked:N0} episode/movie checks"); foreach (var entry in run.Timings) { var timing = entry.Value; - AddDashboard(entry.Key + " timing", $"{timing.Count:N0} calls · mean {timing.MeanSeconds:N2}s · p95 {timing.P95Seconds:N2}s · max {timing.MaxSeconds:N2}s · total {timing.TotalSeconds:N1}s"); + UI.TimingItems.Add(new GenericListItem { PrimaryText = PhaseLabel(entry.Key), SecondaryText = $"{timing.Count:N0} calls · average {timing.MeanSeconds:N2}s · p95 {timing.P95Seconds:N2}s · slowest {timing.MaxSeconds:N2}s · total {timing.TotalSeconds:N1}s", Icon = IconNames.timer, IconMode = ItemListIconMode.SmallRegular, Status = ItemStatus.None }); } - AddDashboard("Timing scope", "Resolution includes dispatch pacing and provider wait. Parallel call totals overlap; they are not wall-clock time. p95 uses the latest 512 samples per stage; other statistics cover this run. Checkpoint timing covers observation saves only."); + } + private static string PhaseLabel(string phase) => phase switch + { + "ImportReconciliation" => "Repairing stream choices", "Sync+Populate" => "Syncing catalogs", + "Resolve" => "Resolving metadata", "VersionRefresh" => "Refreshing versions", + "CollectionPopulation" => "Updating collections", "Repair" => "Checking system state", + "Validation" => "Validating the library", "PhysicalMediaReconciliation" => "Checking owned media", + "Enrichment" => "Enriching metadata", "TokenRenewal" => "Renewing stream tokens", + "metadata" => "Metadata", "observation" => "Library observation", "resolution" => "Source lookup", + "publication" => "File publication", "checkpoint" => "Saving observations", _ => phase.Replace('_',' '), + }; + private static string When(DateTimeOffset? value, DateTimeOffset now) { if (!value.HasValue) return "not yet"; diff --git a/UI/Settings/SyncAndMarvinUI.cs b/UI/Settings/SyncAndMarvinUI.cs index 458014f..d1c1b0a 100644 --- a/UI/Settings/SyncAndMarvinUI.cs +++ b/UI/Settings/SyncAndMarvinUI.cs @@ -12,10 +12,13 @@ public class SyncAndMarvinUI : EditableOptionsBase public override string EditorDescription => "Brain the size of a planet. Still on library duty."; public CaptionItem DashboardCaption { get; set; } = new CaptionItem("Marvin dashboard"); - public LabelItem DashboardHelp { get; set; } = new LabelItem("A dated snapshot of the worker. Refresh to update. Writes are movie/episode groups, not alternate files or proof of playback."); + public LabelItem DashboardHelp { get; set; } = new LabelItem("Current activity and results from the repair pass. Refresh to update. A saved stream still needs a playback test."); public ButtonItem RefreshDashboard { get; set; } = new ButtonItem("Refresh dashboard") { Data1 = "ImportRefresh" }; public GenericItemList DashboardItems { get; set; } = new GenericItemList(); - public CaptionItem HistoryCaption { get; set; } = new CaptionItem("Recent passes"); + public CaptionItem TimingCaption { get; set; } = new CaptionItem("Repair timings"); + public GenericItemList TimingItems { get; set; } = new GenericItemList(); + public LabelItem TimingHelp { get; set; } = new LabelItem("Resolution includes pacing and provider wait. Parallel totals overlap. p95 uses the latest 512 calls; averages and totals cover the pass. Checkpoint covers observation saves."); + public CaptionItem HistoryCaption { get; set; } = new CaptionItem("Recent repair passes"); public GenericItemList RecentRuns { get; set; } = new GenericItemList(); public StatusItem MarvinStatus { get; set; } = new StatusItem("Last library pass", "No report yet", ItemStatus.None); public ButtonItem RunMarvinNowButton { get; set; } = new ButtonItem("Run Marvin now") diff --git a/docs/import-reconciliation.md b/docs/import-reconciliation.md index f44bc6a..615c1cc 100644 --- a/docs/import-reconciliation.md +++ b/docs/import-reconciliation.md @@ -216,7 +216,10 @@ leave publication pending; the worker does not repeatedly download/rewrite it. ## Dashboard and timing history -The native Marvin page has a **Refresh dashboard** button. It reads a timestamped +The native Marvin page has a **Refresh dashboard** button. Current worker phase is +shown separately from the repair-pass report: Marvin can continue catalog, collection +and maintenance work after that report finishes. Repair timings and recent passes +have their own sections, with compact native Emby rows. It reads a timestamped snapshot, including the current phase/title/episode, elapsed time, lookups in flight (including paced waiting), episode/movie checks, missing and refresh attempts, source matches, empty responses, transport errors, lookup deadlines,