Fix diversity deadline drift and rejected lookup leases

InfiniteDrive.csproj

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

Models/ImportDiversityEntry.cs

@@ -49,8 +49,21 @@ public static void Reconcile(ImportCoverage coverage, ImportEpisode ep, string p
         if (!Eligible(coverage, ep))
         {
             if (ep.Diversity != null)
-            { ep.Diversity.Status = providers?.Count > 1 ? "resolved" : "retired";
-              ep.Diversity.Reason = providers?.Count > 1 ? "multiple_sources" : "ineligible"; }
+            {
+                // A successful ordinary refresh waits for Emby indexing. It is not a
+                // new diversity generation and must not keep restarting the six-hour wait.
+                var awaitingIndex = coverage.Exclusion.Length == 0 && coverage.SnapshotStatus == "success" &&
+                    ep.Expected && ep.Eligible && ep.Failure.Length == 0 && ep.State == "awaiting_indexing" &&
+                    providers?.Count == 1 && ep.Diversity.Profile == profile &&
+                    ep.Diversity.Generation == coverage.Generation;
+                // Resolved/retired entries are entering a new single-source wait. A
+                // resolved entry may have no deadline; do not strand it by preserving null.
+                if (awaitingIndex && (ep.Diversity.Status is "resolved" or "retired" ||
+                    ep.Diversity.NextAttempt == null))
+                    ep.Diversity.NextAttempt = now.AddHours(6);
+                ep.Diversity.Status = awaitingIndex ? "waiting" : providers?.Count > 1 ? "resolved" : "retired";
+                ep.Diversity.Reason = awaitingIndex ? "awaiting_indexing" : providers?.Count > 1 ? "multiple_sources" : "ineligible";
+            }
             return;
         }
         if (ep.Diversity == null || ep.Diversity.Profile != profile || ep.Diversity.Generation != coverage.Generation)
@@ -59,6 +72,7 @@ public static void Reconcile(ImportCoverage coverage, ImportEpisode ep, string p
         else if (ep.Diversity.Status is "retired" or "resolved")
         { ep.Diversity.Status = "waiting"; ep.Diversity.Reason = "single_source";
           ep.Diversity.NextAttempt = now.AddHours(6); }
+        else if (ep.Diversity.Reason == "awaiting_indexing") ep.Diversity.Reason = "single_source";
         if (ep.Diversity.Status == "in_flight" && (ep.LeaseUntil == null || ep.LeaseUntil <= now))
         { ep.Diversity.Status = "waiting"; ep.Diversity.Reason = "interrupted";
           ep.Diversity.NextAttempt = now.AddMinutes(5); }

README.md

@@ -9,9 +9,9 @@ imports titles and repairs missing files.
 
 A stream will appear. Probably.
 
-**Current version: 0.42.14**
+**Current version: 0.42.15**
 
-Release 0.42.14 adds eventual provider diversity while retaining the administrator whole-title blocks and recoverable managed-stream cleanup from 0.42.13. It targets **Emby 4.10.0.40**.
+Release 0.42.15 fixes diversity deadlines across routine refresh indexing and immediately releases rejected diversity leases. It retains the eventual provider diversity queue from 0.42.14 and administrator whole-title blocks from 0.42.13. It targets **Emby 4.10.0.40**.
 
 ## How we run it
 
@@ -240,7 +240,7 @@ With **Repair & refresh** enabled, Marvin gradually revisits indexed movies and
 
 The queue defaults on. **Pause source diversity** stops its lookups/additions while essential repair continues; **Enable eventual source diversity** resumes normal scheduled processing. Entries, backoff and attempt history persist. Neither button triggers a pass or resets retries. There is no force-drain action. Unknown labels, blocked/removed/owned titles, unconfirmed identities and essential failures are excluded.
 
-New entries wait six hours. Missing items and ordinary refreshes have priority; diversity borrows at most two checks per normal pass or eight per catch-up pass from the existing shared allowance, serially. With stale health it can make one shared probe only if essential work has not already reserved a check. Same-source/empty diversity results escalate through 6h, 1d, 2d, 3d and 7d, plus up to 20% jitter and longer cooldowns. The queue persists beyond catch-up expiry; gradual processing does not guarantee a second source exists.
+New entries wait six hours. Routine refresh and re-indexing preserve an existing diversity deadline and backoff; they do not restart that wait. Final guard rejection clears only the completed lookup’s own lease. Missing items and ordinary refreshes have priority; diversity borrows at most two checks per normal pass or eight per catch-up pass from the existing shared allowance, serially. With stale health it can make one shared probe only if essential work has not already reserved a check. Same-source/empty diversity results escalate through 6h, 1d, 2d, 3d and 7d, plus up to 20% jitter and longer cooldowns. The queue persists beyond catch-up expiry; gradual processing does not guarantee a second source exists.
 
 A successful diversity check **adds** one validated variant and preserves every working file. New variants require a known different label and positive exact upstream size no greater than 40GB. An item with all eight version slots occupied waits for capacity; Marvin does not delete a working choice to force diversity. A saved addition does not prove playback or independent provider infrastructure. Ordinary refresh retains its existing replacement behavior and can change coverage later.
 

Services/ImportReconciliationService.cs

@@ -117,19 +117,27 @@ async Task CompleteAsync(ResolutionWork work)
             {
                 var live = await _db.GetImportCoverageAsync(work.Coverage.Identity);
                 var episode = live?.Items.FirstOrDefault(x => x.Key == work.Episode.Key);
-                if (_mode() != ImportMode.Repair || live == null || episode == null ||
-                    live.Generation != work.Generation || episode.Lease != work.Lease ||
-                    !episode.Eligible || !episode.Expected ||
-                    !await IsAuthorizedAsync(live, work.Item, token) || _inventory.IsOwned(work.Item)) return;
+                // Only the owner of this lease may settle it, even when other final
+                // guards changed. A stale result must never clear a replacement lease.
+                if (live == null || episode == null || episode.Lease != work.Lease) return;
+                var authorized = episode.Eligible && episode.Expected &&
+                    await IsAuthorizedAsync(live, work.Item, token) && !_inventory.IsOwned(work.Item);
+                var finalGuards = authorized && _mode() == ImportMode.Repair && live.Generation == work.Generation;
                 var now = _clock();
                 if (work.Diversity)
                 {
                     var entry = episode.Diversity;
-                    if (!_diversityEnabled() || entry == null || entry.Profile != _inventory.ProfileFingerprint ||
+                    if (!finalGuards || !_diversityEnabled() || entry == null || entry.Profile != _inventory.ProfileFingerprint ||
                         entry.Generation != live.Generation || !ImportDiversityPolicy.Eligible(live, episode) ||
                         DiversitySignature(episode) != work.Signature)
                     {
-                        if (entry != null) { entry.Status = "waiting"; entry.Reason = "guard_changed"; entry.NextAttempt = now.AddMinutes(5); }
+                        if (entry != null)
+                        {
+                            var retired = !authorized || live.Exclusion.Length > 0 || episode.Failure.Length > 0;
+                            entry.Status = retired ? "retired" : "waiting";
+                            entry.Reason = retired ? "ineligible" : "guard_changed";
+                            if (!retired) entry.NextAttempt = now.AddMinutes(5);
+                        }
                         episode.Lease = null; episode.LeaseUntil = null;
                         await _db.SaveImportCoverageAsync(live, token); return;
                     }
@@ -168,6 +176,7 @@ async Task CompleteAsync(ResolutionWork work)
                     if (ix >= 0) work.Coverage.Items[ix] = episode;
                     return;
                 }
+                if (!finalGuards) return;
                 var failure = result.Failure;
                 sourceHealth.Record(failure, now);
                 await _db.SaveImportSourceHealthAsync(sourceHealth, token);
@@ -204,8 +213,6 @@ async Task CompleteAsync(ResolutionWork work)
                 }
                 episode.Lease = null; episode.LeaseUntil = null;
                 ImportDiversityPolicy.Reconcile(live, episode, _inventory.ProfileFingerprint, now);
-                if (failure.Length == 0 && episode.Diversity?.Status == "waiting")
-                    episode.Diversity.NextAttempt = now.AddHours(6);
                 await _db.SaveImportCoverageAsync(live, token);
                 if (work.Reprocess) await _db.SetImportReprocessStatusAsync(live.Identity, episode.Key,
                     failure.Length == 0 ? "published" : "failed", now, token);

Tests/ImportDiversityTests.cs

@@ -44,13 +44,55 @@ [Fact] public void QueueIsIdempotentAndProfileGenerationChangesRestartItsOwnWait
         var second = ep.Diversity; ImportDiversityPolicy.Reconcile(c, ep, "profile2", Now.AddHours(2));
         Assert.NotSame(second, ep.Diversity); Assert.Equal("profile2", ep.Diversity!.Profile);
     }
-    [Theory] [InlineData("excluded")] [InlineData("index_mismatch")] [InlineData("retrying")] [InlineData("awaiting_indexing")]
+    [Theory] [InlineData("excluded")] [InlineData("index_mismatch")] [InlineData("retrying")]
     public void NonIndexedStatesRetireInsteadOfConsumingDiversityCredits(string state)
     {
         var ep = Single(); var c = Coverage(ep); ImportDiversityPolicy.Reconcile(c, ep, "p", Now);
         ep.State = state; ImportDiversityPolicy.Reconcile(c, ep, "p", Now);
         Assert.Equal("retired", ep.Diversity!.Status);
     }
+    [Fact] public void TransientRefreshIndexingPreservesPersistedDueTimeAndBackoff()
+    {
+        var ep = Single(); var c = Coverage(ep); ImportDiversityPolicy.Reconcile(c, ep, "p", Now);
+        ImportDiversityPolicy.Defer(ep.Diversity!, "same_source", Now, 0, DateTimeOffset.MinValue);
+        var due = ep.Diversity!.NextAttempt;
+        for (var hour = 1; hour <= 8; hour++)
+        {
+            ep.State = "awaiting_indexing";
+            ImportDiversityPolicy.Reconcile(c, ep, "p", Now.AddHours(hour));
+            Assert.Equal("waiting", ep.Diversity.Status);
+            Assert.Equal("awaiting_indexing", ep.Diversity.Reason);
+            Assert.False(ImportDiversityPolicy.Eligible(c, ep));
+            Assert.Equal(due, ep.Diversity.NextAttempt); Assert.Equal(1, ep.Diversity.Streak);
+            ep.State = "indexed";
+            ImportDiversityPolicy.Reconcile(c, ep, "p", Now.AddHours(hour));
+            Assert.Equal("single_source", ep.Diversity.Reason);
+            Assert.Equal(due, ep.Diversity.NextAttempt); Assert.Equal(1, ep.Diversity.Streak);
+        }
+        ep.State = "awaiting_indexing"; c.Exclusion = "not_authorized";
+        ImportDiversityPolicy.Reconcile(c, ep, "p", Now.AddHours(9));
+        Assert.Equal("retired", ep.Diversity.Status);
+    }
+    [Theory] [InlineData("resolved", true)] [InlineData("resolved", false)]
+    [InlineData("retired", true)] [InlineData("retired", false)]
+    public void ReenteringSingleSourceAfterRefreshGetsANewDeadline(string status, bool noDeadline)
+    {
+        var ep = Single(); var c = Coverage(ep); ImportDiversityPolicy.Reconcile(c, ep, "p", Now);
+        ep.Diversity!.Status = status;
+        ep.Diversity.NextAttempt = noDeadline ? null : Now.AddHours(-1);
+        ep.Diversity.Attempts = 3; ep.Diversity.Streak = 2;
+        ep.State = "awaiting_indexing";
+        ImportDiversityPolicy.Reconcile(c, ep, "p", Now.AddHours(1));
+        Assert.Equal("waiting", ep.Diversity.Status);
+        Assert.Equal("awaiting_indexing", ep.Diversity.Reason);
+        Assert.Equal(Now.AddHours(7), ep.Diversity.NextAttempt);
+        Assert.False(ImportDiversityPolicy.Eligible(c, ep));
+        ep.State = "indexed";
+        ImportDiversityPolicy.Reconcile(c, ep, "p", Now.AddHours(2));
+        Assert.Equal("single_source", ep.Diversity.Reason);
+        Assert.Equal(Now.AddHours(7), ep.Diversity.NextAttempt);
+        Assert.Equal(3, ep.Diversity.Attempts); Assert.Equal(2, ep.Diversity.Streak);
+    }
     [Fact] public void ActualFailuresAndBlocksWinOverDiversity()
     {
         var ep = Single(); var c = Coverage(ep); ImportDiversityPolicy.Reconcile(c, ep, "p", Now);

Tests/ImportReconciliationTests.cs

@@ -794,6 +794,89 @@ [Fact] public async Task DiversityEssentialAttemptWinsAndStaleHealthAllowsOnlyOn
         Assert.Equal(new[]{2},h.Inventory.ResolvedEpisodes);
     }
 
+    [Fact] public async Task HourlyRefreshAndReindexingPreserveDiversityDueTimeUntilItCanDispatch()
+    {
+        using var h = new Harness(); await SeedDiversity(h);
+        h.Config.ImportCatchUpStartedAt = ""; h.Config.ImportCatchUpUntil = "";
+        var c = await h.State(); var due = h.Now.AddHours(6);
+        c.Items[0].Diversity!.NextAttempt = due;
+        c.Items[0].Diversity!.Streak = 2;
+        await h.Db.SaveImportCoverageAsync(c);
+        h.Inventory.Service = "TorBox";
+        for (var hour = 1; hour <= 6; hour++)
+        {
+            h.Now = Now.AddHours(hour).AddMinutes(hour);
+            await h.Run(); // Successful ordinary refresh is temporarily awaiting indexing.
+            Assert.Equal("awaiting_indexing", (await h.State()).Items[0].State);
+            await h.Run(ImportMode.Observe); // Emby indexes the unchanged single-provider choices.
+            var episode = (await h.State()).Items[0];
+            Assert.Equal(due, episode.Diversity!.NextAttempt);
+            Assert.Equal(2, episode.Diversity.Streak);
+        }
+        h.Db = new DatabaseManager(h.Directory.Path, NullLogger.Instance); h.Db.Initialise();
+        h.Inventory.Service = "Usenet";
+        await h.Run(); // The latest refresh is fresh; the original diversity due time has elapsed.
+        var final = (await h.State()).Items[0];
+        Assert.Equal("resolved", final.Diversity!.Status);
+        Assert.Equal(1, h.Inventory.Added);
+        Assert.Equal(6, h.Inventory.Published);
+        Assert.Equal(7, await h.Db.GetRecentImportAttemptsAsync(h.Now));
+    }
+
+    [Theory]
+    [InlineData("observe", "waiting")]
+    [InlineData("generation", "waiting")]
+    [InlineData("blocked", "retired")]
+    [InlineData("owned", "retired")]
+    [InlineData("ineligible", "retired")]
+    [InlineData("unexpected", "retired")]
+    public async Task RejectedDiversityCompletionImmediatelyClearsItsLease(string change, string status)
+    {
+        using var h = new Harness(); await SeedDiversity(h); h.Inventory.Service = "Usenet";
+        var mode = ImportMode.Repair;
+        h.Inventory.OnResolve = async () =>
+        {
+            if (change == "observe") mode = ImportMode.Observe;
+            else if (change == "owned") h.Inventory.Owned = true;
+            else if (change == "blocked")
+                await h.Db.UpsertBlockedItemAsync(h.Item.AioId, null, null, "fixture", "series", "test");
+            else
+            {
+                var live = await h.State();
+                if (change == "generation") live.Generation++;
+                if (change == "ineligible") live.Items[0].Eligible = false;
+                if (change == "unexpected") live.Items[0].Expected = false;
+                await h.Db.SaveImportCoverageAsync(live);
+            }
+        };
+        await new ImportReconciliationService(h.Db, h.Inventory, () => mode, TimeZoneInfo.Utc,
+            () => h.Now, () => h.Cooldown, () => ImportWorkBudget.For(h.Config, h.Now),
+            () => h.Config.ImportProviderDiversityEnabled).RunAsync(default, new[] { h.Item });
+        var e = (await h.State()).Items[0];
+        Assert.Null(e.Lease); Assert.Null(e.LeaseUntil);
+        Assert.Equal(status, e.Diversity!.Status);
+        Assert.Equal(0, e.Diversity.Streak);
+        Assert.Equal(1, e.Diversity.Attempts);
+        Assert.Equal(0, h.Inventory.Added); Assert.Single(e.Paths);
+        Assert.Equal("", e.Failure);
+        Assert.Equal(1, await h.Db.GetRecentImportAttemptsAsync(h.Now));
+    }
+
+    [Fact] public async Task StaleDiversityCompletionCannotClearAReplacementLease()
+    {
+        using var h = new Harness(); await SeedDiversity(h); h.Inventory.Service = "Usenet";
+        var expiry = h.Now.AddMinutes(20);
+        h.Inventory.OnResolve = async () =>
+        {
+            var live = await h.State();
+            live.Items[0].Lease = "replacement"; live.Items[0].LeaseUntil = expiry;
+            await h.Db.SaveImportCoverageAsync(live);
+        };
+        await h.Run(); var e = (await h.State()).Items[0];
+        Assert.Equal("replacement", e.Lease); Assert.Equal(expiry, e.LeaseUntil);
+        Assert.Equal("in_flight", e.Diversity!.Status); Assert.Equal(0, h.Inventory.Added);
+    }
+
     private sealed class Harness : IDisposable
     {
         public TempDir Directory = new(); public DatabaseManager Db; public FakeInventory Inventory = new();

docs/README.md

@@ -28,4 +28,4 @@
 old test plans and dated reports. These are not current configuration instructions.
 The implementation and current guides take precedence over historical claims.
 
-Updated October 3, 2026 for the 0.42.14 source tree.
+Updated October 4, 2026 for the 0.42.15 source tree.