feat: Better handle long-running jobs and the possibility of failure

PatchPanda.Units/Helpers/JobRegistryTests.cs

@@ -0,0 +1,72 @@
+namespace PatchPanda.Units.Helpers;
+
+public class JobRegistryTests
+{
+    [Fact]
+    public async Task TryMarkForResetAll_DedupesQueuedAndProcessingJobs()
+    {
+        var registry = new JobRegistry(new JobQueue());
+
+        var firstQueued = await registry.TryMarkForResetAll();
+        var secondQueued = await registry.TryMarkForResetAll();
+
+        Assert.True(firstQueued);
+        Assert.False(secondQueued);
+
+        var resetJob = Assert.Single(registry.GetSnapshot().OfType<PendingResetAll>());
+
+        Assert.True(registry.TryStartProcessing(resetJob.Sequence));
+        Assert.False(await registry.TryMarkForResetAll());
+
+        registry.FinishProcessing(resetJob.Sequence);
+
+        Assert.True(await registry.TryMarkForResetAll());
+    }
+
+    [Fact]
+    public async Task TryMarkForCheckUpdatesAll_DedupesQueuedAndProcessingJobs()
+    {
+        var registry = new JobRegistry(new JobQueue());
+
+        var firstQueued = await registry.TryMarkForCheckUpdatesAll();
+        var secondQueued = await registry.TryMarkForCheckUpdatesAll();
+
+        Assert.True(firstQueued);
+        Assert.False(secondQueued);
+
+        var checkJob = Assert.Single(registry.GetSnapshot().OfType<PendingCheckForUpdatesAll>());
+
+        Assert.True(registry.TryStartProcessing(checkJob.Sequence));
+        Assert.False(await registry.TryMarkForCheckUpdatesAll());
+
+        registry.FinishProcessing(checkJob.Sequence);
+
+        Assert.True(await registry.TryMarkForCheckUpdatesAll());
+    }
+
+    [Fact]
+    public async Task TryMarkForRestartStack_DedupesPerStack()
+    {
+        var registry = new JobRegistry(new JobQueue());
+
+        var firstQueued = await registry.TryMarkForRestartStack(12);
+        var duplicateQueued = await registry.TryMarkForRestartStack(12);
+        var differentStackQueued = await registry.TryMarkForRestartStack(24);
+
+        Assert.True(firstQueued);
+        Assert.False(duplicateQueued);
+        Assert.True(differentStackQueued);
+
+        var restartJob = Assert.Single(
+            registry.GetSnapshot().OfType<PendingRestartStack>(),
+            x => x.StackId == 12
+        );
+
+        Assert.True(registry.TryStartProcessing(restartJob.Sequence));
+        Assert.False(await registry.TryMarkForRestartStack(12));
+
+        registry.FinishProcessing(restartJob.Sequence);
+
+        Assert.True(await registry.TryMarkForRestartStack(12));
+    }
+}

PatchPanda.Units/Services/UpdateServiceTests.cs

@@ -123,7 +123,14 @@ string parameterName
         );
 
         dockerMock
-            .Setup(x => x.RunDockerComposeOnPath(It.IsAny<ComposeStack>(), It.IsAny<string>()))
+            .Setup(x =>
+                x.RunDockerComposeOnPath(
+                    It.IsAny<ComposeStack>(),
+                    It.IsAny<string>(),
+                    It.IsAny<Action<string>?>(),
+                    It.IsAny<CancellationToken>()
+                )
+            )
             .ReturnsAsync((string.Empty, string.Empty, 0));
 
         var tasks = await new UpdateService(
@@ -209,10 +216,14 @@ public async Task PortainerComposeUpdateTest()
 
         _portainerService.Setup(p => p.IsConfigured).Returns(true);
         _portainerService
-            .Setup(p => p.GetStackFileContentAsync(It.IsAny<string>()))
+            .Setup(p => p.GetStackFileContentAsync(It.IsAny<string>(), It.IsAny<CancellationToken>()))
             .ReturnsAsync(composeContent);
         _portainerService.Setup(p =>
-            p.UpdateStackFileContentAsync(It.IsAny<string>(), It.IsAny<string>())
+            p.UpdateStackFileContentAsync(
+                It.IsAny<string>(),
+                It.IsAny<string>(),
+                It.IsAny<CancellationToken>()
+            )
         );
 
         var dbContextFactory = Helper.CreateInMemoryFactory();
@@ -236,7 +247,14 @@ public async Task PortainerComposeUpdateTest()
         );
 
         dockerMock
-            .Setup(x => x.RunDockerComposeOnPath(It.IsAny<ComposeStack>(), It.IsAny<string>()))
+            .Setup(x =>
+                x.RunDockerComposeOnPath(
+                    It.IsAny<ComposeStack>(),
+                    It.IsAny<string>(),
+                    It.IsAny<Action<string>?>(),
+                    It.IsAny<CancellationToken>()
+                )
+            )
             .ReturnsAsync((string.Empty, string.Empty, 0));
 
         var updateService = new UpdateService(
@@ -258,12 +276,16 @@ public async Task PortainerComposeUpdateTest()
 
         Assert.NotNull(tasks);
 
-        _portainerService.Verify(p => p.GetStackFileContentAsync(stack.StackName), Times.Once);
+        _portainerService.Verify(
+            p => p.GetStackFileContentAsync(stack.StackName, It.IsAny<CancellationToken>()),
+            Times.Once
+        );
         _portainerService.Verify(
             p =>
                 p.UpdateStackFileContentAsync(
                     stack.StackName,
-                    It.Is<string>(s => s.Contains(TestData.IMAGE_NEW_VERSION))
+                    It.Is<string>(s => s.Contains(TestData.IMAGE_NEW_VERSION)),
+                    It.IsAny<CancellationToken>()
                 ),
             Times.Once
         );
@@ -331,7 +353,14 @@ public async Task UpdatePropagatesToMatchingFullImage()
         );
 
         dockerMock
-            .Setup(x => x.RunDockerComposeOnPath(It.IsAny<ComposeStack>(), It.IsAny<string>()))
+            .Setup(x =>
+                x.RunDockerComposeOnPath(
+                    It.IsAny<ComposeStack>(),
+                    It.IsAny<string>(),
+                    It.IsAny<Action<string>?>(),
+                    It.IsAny<CancellationToken>()
+                )
+            )
             .ReturnsAsync((string.Empty, string.Empty, 0));
 
         var updateService = new UpdateService(

PatchPanda.Web/Constants.cs

@@ -43,5 +43,7 @@ public static class Limits
     {
         public const int MAX_OLLAMA_ATTEMPTS = 3;
         public const int MINIMUM_UPDATE_STEPS = 3;
+        public const int PORTAINER_HTTP_TIMEOUT_SECONDS = 60;
+        public const int UPDATE_JOB_TIMEOUT_SECONDS = 300;
     }
 }

PatchPanda.Web/Helpers/JobRegistry.cs

@@ -35,17 +35,11 @@ public class PendingCheckForUpdatesAll : PendingUpdate
     public override string Kind => "CheckForUpdatesAll";
 }
 
-public class JobRegistry
+public class JobRegistry(JobQueue updateQueue)
 {
     private readonly ConcurrentDictionary<long, PendingUpdate> _pending = new();
-    private readonly JobQueue _updateQueue;
     private long _sequenceCounter;
-    private readonly object _processingLock = new();
-
-    public JobRegistry(JobQueue updateQueue)
-    {
-        _updateQueue = updateQueue;
-    }
+    private readonly Lock _processingLock = new();
 
     private long GetNextSequence() => Interlocked.Increment(ref _sequenceCounter);
 
@@ -64,47 +58,107 @@ public async Task MarkForUpdate(
             TargetVersionId = targetVersionId,
             TargetVersionNumber = targetVersionNumber,
             IsAutomatic = isAutomatic,
-            Sequence = seq
+            Sequence = seq,
         };
 
         _pending.TryAdd(seq, pending);
 
-        await _updateQueue.EnqueueAsync(
+        await updateQueue.EnqueueAsync(
             new UpdateJob(seq, containerId, targetVersionId, targetVersionNumber, isAutomatic)
         );
     }
 
     public async Task MarkForResetAll()
     {
-        var seq = GetNextSequence();
+        await TryMarkForResetAll();
+    }
 
-        var pending = new PendingResetAll { Sequence = seq };
+    public async Task MarkForCheckUpdatesAll()
+    {
+        await TryMarkForCheckUpdatesAll();
+    }
 
-        _pending.TryAdd(seq, pending);
+    public async Task MarkForRestartStack(int stackId)
+    {
+        await TryMarkForRestartStack(stackId);
+    }
+
+    public async Task<bool> TryMarkForRestartStack(int stackId)
+    {
+        PendingRestartStack? pending;
+
+        lock (_processingLock)
+        {
+            if (_pending.Values.Any(p => p is PendingRestartStack rs && rs.StackId == stackId))
+                return false;
 
-        await _updateQueue.EnqueueAsync(new ResetAllJob(seq));
+            var seq = GetNextSequence();
+            pending = new() { StackId = stackId, Sequence = seq };
+            _pending.TryAdd(seq, pending);
+        }
+
+        try
+        {
+            await updateQueue.EnqueueAsync(new RestartStackJob(pending.Sequence, stackId));
+            return true;
+        }
+        catch
+        {
+            _pending.TryRemove(pending.Sequence, out _);
+            throw;
+        }
     }
 
-    public async Task MarkForCheckUpdatesAll()
+    public async Task<bool> TryMarkForResetAll()
     {
-        var seq = GetNextSequence();
+        PendingResetAll? pending;
 
-        var pending = new PendingCheckForUpdatesAll { Sequence = seq };
+        lock (_processingLock)
+        {
+            if (_pending.Values.Any(p => p is PendingResetAll))
+                return false;
 
-        _pending.TryAdd(seq, pending);
+            var seq = GetNextSequence();
+            pending = new() { Sequence = seq };
+            _pending.TryAdd(seq, pending);
+        }
 
-        await _updateQueue.EnqueueAsync(new CheckForUpdatesAllJob(seq));
+        try
+        {
+            await updateQueue.EnqueueAsync(new ResetAllJob(pending.Sequence));
+            return true;
+        }
+        catch
+        {
+            _pending.TryRemove(pending.Sequence, out _);
+            throw;
+        }
     }
 
-    public async Task MarkForRestartStack(int stackId)
+    public async Task<bool> TryMarkForCheckUpdatesAll()
     {
-        var seq = GetNextSequence();
+        PendingCheckForUpdatesAll? pending;
 
-        var pending = new PendingRestartStack { StackId = stackId, Sequence = seq };
+        lock (_processingLock)
+        {
+            if (_pending.Values.Any(p => p is PendingCheckForUpdatesAll))
+                return false;
 
-        _pending.TryAdd(seq, pending);
+            var seq = GetNextSequence();
+            pending = new() { Sequence = seq };
+            _pending.TryAdd(seq, pending);
+        }
 
-        await _updateQueue.EnqueueAsync(new RestartStackJob(seq, stackId));
+        try
+        {
+            await updateQueue.EnqueueAsync(new CheckForUpdatesAllJob(pending.Sequence));
+            return true;
+        }
+        catch
+        {
+            _pending.TryRemove(pending.Sequence, out _);
+            throw;
+        }
     }
 
     public bool TryStartProcessing(long sequence)
@@ -188,7 +242,7 @@ public List<PendingUpdate> GetSnapshot()
                     TargetVersionNumber = pendingUpdateJob.TargetVersionNumber,
                     IsAutomatic = pendingUpdateJob.IsAutomatic,
                     IsProcessing = pendingUpdateJob.IsProcessing,
-                    Sequence = pendingUpdateJob.Sequence
+                    Sequence = pendingUpdateJob.Sequence,
                 };
 
                 lock (pendingUpdateJob.Output)
@@ -204,7 +258,7 @@ public List<PendingUpdate> GetSnapshot()
                 {
                     StackId = pendingRestartStack.StackId,
                     IsProcessing = pendingRestartStack.IsProcessing,
-                    Sequence = pendingRestartStack.Sequence
+                    Sequence = pendingRestartStack.Sequence,
                 };
 
                 lock (pendingRestartStack.Output)
@@ -231,7 +285,7 @@ private static void CreateJobCopy<TPending>(List<PendingUpdate> list, TPending p
         var copy = new TPending
         {
             IsProcessing = pending.IsProcessing,
-            Sequence = pending.Sequence
+            Sequence = pending.Sequence,
         };
         lock (pending.Output)
         {