diff --git a/PatchPanda.Units/Helpers/JobRegistryTests.cs b/PatchPanda.Units/Helpers/JobRegistryTests.cs new file mode 100644 index 0000000..9700fd8 --- /dev/null +++ b/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()); + + 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()); + + 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(), + 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)); + } +} diff --git a/PatchPanda.Units/Services/UpdateServiceTests.cs b/PatchPanda.Units/Services/UpdateServiceTests.cs index 2dc6e27..d94d389 100644 --- a/PatchPanda.Units/Services/UpdateServiceTests.cs +++ b/PatchPanda.Units/Services/UpdateServiceTests.cs @@ -123,7 +123,14 @@ string parameterName ); dockerMock - .Setup(x => x.RunDockerComposeOnPath(It.IsAny(), It.IsAny())) + .Setup(x => + x.RunDockerComposeOnPath( + It.IsAny(), + It.IsAny(), + It.IsAny?>(), + It.IsAny() + ) + ) .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())) + .Setup(p => p.GetStackFileContentAsync(It.IsAny(), It.IsAny())) .ReturnsAsync(composeContent); _portainerService.Setup(p => - p.UpdateStackFileContentAsync(It.IsAny(), It.IsAny()) + p.UpdateStackFileContentAsync( + It.IsAny(), + It.IsAny(), + It.IsAny() + ) ); var dbContextFactory = Helper.CreateInMemoryFactory(); @@ -236,7 +247,14 @@ public async Task PortainerComposeUpdateTest() ); dockerMock - .Setup(x => x.RunDockerComposeOnPath(It.IsAny(), It.IsAny())) + .Setup(x => + x.RunDockerComposeOnPath( + It.IsAny(), + It.IsAny(), + It.IsAny?>(), + It.IsAny() + ) + ) .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()), + Times.Once + ); _portainerService.Verify( p => p.UpdateStackFileContentAsync( stack.StackName, - It.Is(s => s.Contains(TestData.IMAGE_NEW_VERSION)) + It.Is(s => s.Contains(TestData.IMAGE_NEW_VERSION)), + It.IsAny() ), Times.Once ); @@ -331,7 +353,14 @@ public async Task UpdatePropagatesToMatchingFullImage() ); dockerMock - .Setup(x => x.RunDockerComposeOnPath(It.IsAny(), It.IsAny())) + .Setup(x => + x.RunDockerComposeOnPath( + It.IsAny(), + It.IsAny(), + It.IsAny?>(), + It.IsAny() + ) + ) .ReturnsAsync((string.Empty, string.Empty, 0)); var updateService = new UpdateService( diff --git a/PatchPanda.Web/Constants.cs b/PatchPanda.Web/Constants.cs index 8c26abe..12fbd2f 100644 --- a/PatchPanda.Web/Constants.cs +++ b/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; } } diff --git a/PatchPanda.Web/Helpers/JobRegistry.cs b/PatchPanda.Web/Helpers/JobRegistry.cs index 840782c..5ba8f4b 100644 --- a/PatchPanda.Web/Helpers/JobRegistry.cs +++ b/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 _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 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 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 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 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 GetSnapshot() { StackId = pendingRestartStack.StackId, IsProcessing = pendingRestartStack.IsProcessing, - Sequence = pendingRestartStack.Sequence + Sequence = pendingRestartStack.Sequence, }; lock (pendingRestartStack.Output) @@ -231,7 +285,7 @@ private static void CreateJobCopy(List list, TPending p var copy = new TPending { IsProcessing = pending.IsProcessing, - Sequence = pending.Sequence + Sequence = pending.Sequence, }; lock (pending.Output) { diff --git a/PatchPanda.Web/Services/Background/UpdateBackgroundService.cs b/PatchPanda.Web/Services/Background/UpdateBackgroundService.cs index 3710904..9c226c4 100644 --- a/PatchPanda.Web/Services/Background/UpdateBackgroundService.cs +++ b/PatchPanda.Web/Services/Background/UpdateBackgroundService.cs @@ -2,25 +2,16 @@ namespace PatchPanda.Web.Services.Background; -public class UpdateBackgroundService : IHostedService, IDisposable +public class UpdateBackgroundService( + IServiceScopeFactory serviceProvider, + JobRegistry jobRegistry, + JobQueue queue +) : IHostedService, IDisposable { - private readonly IServiceScopeFactory _serviceProvider; - private readonly JobRegistry _jobRegistry; - private readonly JobQueue _queue; + private const int JobTimeoutSeconds = Constants.Limits.UPDATE_JOB_TIMEOUT_SECONDS; private CancellationTokenSource? _cts; private Task? _processingTask; - public UpdateBackgroundService( - IServiceScopeFactory serviceProvider, - JobRegistry jobRegistry, - JobQueue queue - ) - { - _serviceProvider = serviceProvider; - _jobRegistry = jobRegistry; - _queue = queue; - } - public void Dispose() { _cts?.Cancel(); @@ -30,7 +21,7 @@ public void Dispose() public Task StartAsync(CancellationToken cancellationToken) { - using var scope = _serviceProvider.CreateScope(); + using var scope = serviceProvider.CreateScope(); var logger = scope.ServiceProvider.GetRequiredService>(); logger.LogInformation("Update background service starting (queue consumer)"); @@ -43,7 +34,7 @@ public Task StartAsync(CancellationToken cancellationToken) public Task StopAsync(CancellationToken cancellationToken) { - using var scope = _serviceProvider.CreateScope(); + using var scope = serviceProvider.CreateScope(); var logger = scope.ServiceProvider.GetService>(); logger?.LogInformation("Update background service stopping"); @@ -62,23 +53,54 @@ public Task StopAsync(CancellationToken cancellationToken) private async Task ProcessJob( TJob job, ILogger logger, - IServiceScope scope, - Func function + CancellationToken cancellationToken, + Func function ) where TJob : AbstractJob { - if (!_jobRegistry.TryStartProcessing(job.Sequence)) + if (!jobRegistry.TryStartProcessing(job.Sequence)) return; var jobName = job.GetType().Name; - _jobRegistry.AppendOutput(job.Sequence, $"Starting job type {jobName} (queued)..."); + jobRegistry.AppendOutput(job.Sequence, $"Starting job type {jobName} (queued)..."); + + using var timeoutCts = new CancellationTokenSource( + TimeSpan.FromSeconds(Math.Max(1, JobTimeoutSeconds)) + ); + using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + timeoutCts.Token + ); try { - await function.Invoke(jobName); + await function.Invoke(jobName, linkedCts.Token); - _jobRegistry.AppendOutput(job.Sequence, $"{jobName} finished."); + jobRegistry.AppendOutput(job.Sequence, $"{jobName} finished."); + } + catch (OperationCanceledException ex) when (timeoutCts.IsCancellationRequested) + { + logger.LogError( + ex, + "Job {JobName} timed out after {TimeoutSeconds} seconds", + jobName, + JobTimeoutSeconds + ); + jobRegistry.AppendOutput( + job.Sequence, + $"{jobName} timed out after {JobTimeoutSeconds} seconds." + ); + } + catch (OperationCanceledException ex) when (cancellationToken.IsCancellationRequested) + { + logger.LogInformation(ex, "Job {JobName} cancelled due to host shutdown", jobName); + jobRegistry.AppendOutput(job.Sequence, $"{jobName} cancelled due to host shutdown."); + } + catch (TimeoutException ex) + { + logger.LogError(ex, "Job {JobName} timed out", jobName); + jobRegistry.AppendOutput(job.Sequence, $"{jobName} timed out: " + ex.Message); } catch (Exception ex) { @@ -88,159 +110,169 @@ Func function jobName, JsonSerializer.Serialize(job) ); - _jobRegistry.AppendOutput(job.Sequence, $"{jobName} failed: " + ex.Message); + jobRegistry.AppendOutput(job.Sequence, $"{jobName} failed: " + ex.Message); } finally { - _jobRegistry.FinishProcessing(job.Sequence); + jobRegistry.FinishProcessing(job.Sequence); } } private async Task ProcessQueueAsync(CancellationToken cancellationToken) { - await foreach (var job in _queue.Reader.ReadAllAsync(cancellationToken)) + try { - if (cancellationToken.IsCancellationRequested) - break; - - using var scope = _serviceProvider.CreateScope(); - var logger = scope.ServiceProvider.GetRequiredService< - ILogger - >(); - - switch (job) + await foreach (var job in queue.Reader.ReadAllAsync(cancellationToken)) { - case UpdateJob updateJob: - await ProcessJob( - updateJob, - logger, - scope, - async (string jobName) => - { - var updateService = - scope.ServiceProvider.GetRequiredService(); - var dbFactory = scope.ServiceProvider.GetRequiredService< - IDbContextFactory - >(); - - await using var db = await dbFactory.CreateDbContextAsync( - cancellationToken - ); - var app = await db - .Containers.Include(x => x.NewerVersions) - .FirstOrDefaultAsync( - x => x.Id == updateJob.ContainerId, - cancellationToken - ); + if (cancellationToken.IsCancellationRequested) + break; - if (app is null) + using var scope = serviceProvider.CreateScope(); + var logger = scope.ServiceProvider.GetRequiredService< + ILogger + >(); + + switch (job) + { + case UpdateJob updateJob: + await ProcessJob( + updateJob, + logger, + cancellationToken, + async (string jobName, CancellationToken jobCancellationToken) => { - logger.LogInformation( - "Disregarding job for missing container {ContainerId}", - updateJob.ContainerId + var updateService = + scope.ServiceProvider.GetRequiredService(); + var dbFactory = scope.ServiceProvider.GetRequiredService< + IDbContextFactory + >(); + + await using var db = await dbFactory.CreateDbContextAsync( + jobCancellationToken ); - _jobRegistry.AppendOutput( - updateJob.Sequence, - "Container not found." + var app = await db + .Containers.Include(x => x.NewerVersions) + .FirstOrDefaultAsync( + x => x.Id == updateJob.ContainerId, + jobCancellationToken + ); + + if (app is null) + { + logger.LogInformation( + "Disregarding job for missing container {ContainerId}", + updateJob.ContainerId + ); + jobRegistry.AppendOutput( + updateJob.Sequence, + "Container not found." + ); + return; + } + + var targetVersion = app.NewerVersions.FirstOrDefault(v => + v.Id == updateJob.TargetVersionId ); - _jobRegistry.FinishProcessing(updateJob.Sequence); - return; - } - var targetVersion = app.NewerVersions.FirstOrDefault(v => - v.Id == updateJob.TargetVersionId - ); - - if (targetVersion is null) - return; - - await updateService.Update( - app, - false, - targetVersion, - (line) => _jobRegistry.AppendOutput(updateJob.Sequence, line), - updateJob.IsAutomatic - ); - } - ); - break; + if (targetVersion is null) + return; - case ResetAllJob resetAllJob: - await ProcessJob( - resetAllJob, - logger, - scope, - async (string jobName) => - { - var dockerService = - scope.ServiceProvider.GetRequiredService(); - await dockerService.ResetComposeStacks(); - } - ); - break; - - case CheckForUpdatesAllJob checkForUpdatesAllJob: - await ProcessJob( - checkForUpdatesAllJob, - logger, - scope, - async (string jobName) => - { - var updateService = - scope.ServiceProvider.GetRequiredService(); - await updateService.CheckAllForUpdates(); - } - ); - break; - - case RestartStackJob restartStackJob: - await ProcessJob( - restartStackJob, - logger, - scope, - async (string jobName) => - { - var dockerService = - scope.ServiceProvider.GetRequiredService(); - var dbFactory = scope.ServiceProvider.GetRequiredService< - IDbContextFactory - >(); - - using var db = dbFactory.CreateDbContext(); - var stack = await db - .Stacks.Include(x => x.Apps) - .FirstOrDefaultAsync( - x => x.Id == restartStackJob.StackId, - cancellationToken + await updateService.Update( + app, + false, + targetVersion, + (line) => jobRegistry.AppendOutput(updateJob.Sequence, line), + updateJob.IsAutomatic, + jobCancellationToken ); - - if (stack is null) + } + ); + break; + + case ResetAllJob resetAllJob: + await ProcessJob( + resetAllJob, + logger, + cancellationToken, + async (string jobName, CancellationToken jobCancellationToken) => + { + var dockerService = + scope.ServiceProvider.GetRequiredService(); + await dockerService.ResetComposeStacks(jobCancellationToken); + } + ); + break; + + case CheckForUpdatesAllJob checkForUpdatesAllJob: + await ProcessJob( + checkForUpdatesAllJob, + logger, + cancellationToken, + async (string jobName, CancellationToken jobCancellationToken) => + { + var updateService = + scope.ServiceProvider.GetRequiredService(); + await updateService.CheckAllForUpdates(jobCancellationToken); + } + ); + break; + + case RestartStackJob restartStackJob: + await ProcessJob( + restartStackJob, + logger, + cancellationToken, + async (string jobName, CancellationToken jobCancellationToken) => { - logger.LogInformation( - "Disregarding job for missing stack {StackId}", - restartStackJob.StackId + var dockerService = + scope.ServiceProvider.GetRequiredService(); + var dbFactory = scope.ServiceProvider.GetRequiredService< + IDbContextFactory + >(); + + await using var db = await dbFactory.CreateDbContextAsync( + jobCancellationToken ); - _jobRegistry.AppendOutput( - restartStackJob.Sequence, - "Stack not found." + var stack = await db + .Stacks.Include(x => x.Apps) + .FirstOrDefaultAsync( + x => x.Id == restartStackJob.StackId, + jobCancellationToken + ); + + if (stack is null) + { + logger.LogInformation( + "Disregarding job for missing stack {StackId}", + restartStackJob.StackId + ); + jobRegistry.AppendOutput( + restartStackJob.Sequence, + "Stack not found." + ); + return; + } + + await dockerService.RunDockerComposeOnPath( + stack, + "restart", + (line) => + jobRegistry.AppendOutput(restartStackJob.Sequence, line), + jobCancellationToken ); - _jobRegistry.FinishProcessing(restartStackJob.Sequence); - return; } - - await dockerService.RunDockerComposeOnPath( - stack, - "restart", - (line) => _jobRegistry.AppendOutput(restartStackJob.Sequence, line) - ); - } - ); - break; - - default: - logger.LogError("Disregarding unknown job type {JobType}", job.GetType().Name); - break; + ); + break; + + default: + logger.LogError( + "Disregarding unknown job type {JobType}", + job.GetType().Name + ); + break; + } } } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { } } } diff --git a/PatchPanda.Web/Services/DockerService.cs b/PatchPanda.Web/Services/DockerService.cs index 027d245..5c075fb 100644 --- a/PatchPanda.Web/Services/DockerService.cs +++ b/PatchPanda.Web/Services/DockerService.cs @@ -62,14 +62,17 @@ public async Task IsAliveAsync() private DockerClient GetClient() => new DockerClientConfiguration(new Uri(DockerSocket)).CreateClient(); - private async Task?> GetAllContainers() + private async Task?> GetAllContainers( + CancellationToken cancellationToken = default + ) { try { using var dockerClient = GetClient(); var containers = await dockerClient.Containers.ListContainersAsync( - new ContainersListParameters { All = true, Limit = 999 } + new ContainersListParameters { All = true, Limit = 999 }, + cancellationToken ); return containers; @@ -81,9 +84,11 @@ private DockerClient GetClient() => } } - private async Task?> GetRunningStacks() + private async Task?> GetRunningStacks( + CancellationToken cancellationToken = default + ) { - var containers = await GetAllContainers(); + var containers = await GetAllContainers(cancellationToken); if (containers is null) { @@ -125,7 +130,8 @@ existingStack.ConfigFile is null ) { var stackFileContent = await _portainerService.GetStackFileContentAsync( - stackName + stackName, + cancellationToken ); _logger.LogInformation( "Retrieved stack file content for {StackName}, length: {Length}", @@ -230,10 +236,12 @@ out var appVersion /// Resets current list of containers and fills it with existing containers. /// /// if successfully reset, otherwise . - public async Task ResetComposeStacks() + public async Task ResetComposeStacks(CancellationToken cancellationToken = default) { using var db = _dbContextFactory.CreateDbContext(); + cancellationToken.ThrowIfCancellationRequested(); + if (!await IsAliveAsync()) { return false; @@ -242,9 +250,9 @@ public async Task ResetComposeStacks() var existingStacks = await db .Stacks.Include(x => x.Apps) .ThenInclude(x => x.NewerVersions) - .ToListAsync(); + .ToListAsync(cancellationToken); - var runningStacks = await GetRunningStacks(); + var runningStacks = await GetRunningStacks(cancellationToken); if (runningStacks is null) { @@ -273,9 +281,9 @@ public async Task ResetComposeStacks() foreach (var runningContainer in runningStack.Apps) { - var existingContainer = existingStack - .Apps.Where(x => x.Name == runningContainer.Name) - .FirstOrDefault(); + var existingContainer = existingStack.Apps.FirstOrDefault(x => + x.Name == runningContainer.Name + ); var foundApps = new List(); if (existingContainer is not null) @@ -324,13 +332,13 @@ existingContainer.OverrideGitHubRepo is null db.MultiContainerApps.RemoveRange(db.MultiContainerApps); - await db.SaveChangesAsync(); + await db.SaveChangesAsync(cancellationToken); - var stacks = await db.Stacks.Include(x => x.Apps).ToListAsync(); + var stacks = await db.Stacks.Include(x => x.Apps).ToListAsync(cancellationToken); stacks.ForEach(x => MultiContainerAppDetector.FillMultiContainerApps(x, db)); - await db.SaveChangesAsync(); + await db.SaveChangesAsync(cancellationToken); return true; } @@ -338,7 +346,8 @@ existingContainer.OverrideGitHubRepo is null public virtual async Task<(string stdOut, string stdErr, int exitCode)> RunDockerComposeOnPath( ComposeStack stack, string command, - Action? outputCallback = null + Action? outputCallback = null, + CancellationToken cancellationToken = default ) { var fileName = "docker"; @@ -381,7 +390,33 @@ existingContainer.OverrideGitHubRepo is null process.BeginOutputReadLine(); process.BeginErrorReadLine(); - await process.WaitForExitAsync(); + try + { + await process.WaitForExitAsync(cancellationToken); + } + catch (OperationCanceledException ex) + { + if (cancellationToken.IsCancellationRequested) + throw; + + try + { + if (!process.HasExited) + process.Kill(true); + } + catch (Exception killException) + { + _logger.LogWarning( + killException, + "Failed killing timed out docker compose process" + ); + } + + throw new TimeoutException( + $"Docker compose command '{fileName} {arguments}' timed out.", + ex + ); + } var stdOut = standardOutput.ToString(); var stdErr = standardError.ToString(); diff --git a/PatchPanda.Web/Services/Hosted/VersionCheckHostedService.cs b/PatchPanda.Web/Services/Hosted/VersionCheckHostedService.cs index 3bd2a45..8b1a4cd 100644 --- a/PatchPanda.Web/Services/Hosted/VersionCheckHostedService.cs +++ b/PatchPanda.Web/Services/Hosted/VersionCheckHostedService.cs @@ -1,11 +1,12 @@ namespace PatchPanda.Web.Services.Hosted; -public class VersionCheckHostedService : IHostedService +public class VersionCheckHostedService : IHostedService, IDisposable { private readonly ILogger _logger; private readonly JobRegistry _jobRegistry; private Timer? _timer; + private int _isRunning; public VersionCheckHostedService( ILogger logger, @@ -50,9 +51,37 @@ public Task StopAsync(CancellationToken cancellationToken) return Task.CompletedTask; } - public async void DoWork(object? state) + private async void DoWork(object? state) { - await _jobRegistry.MarkForResetAll(); - await _jobRegistry.MarkForCheckUpdatesAll(); + if (Interlocked.Exchange(ref _isRunning, 1) == 1) + { + _logger.LogInformation( + "Skipping scheduled version check because a previous run is still enqueueing jobs." + ); + return; + } + + try + { + var queuedReset = await _jobRegistry.TryMarkForResetAll(); + if (!queuedReset) + _logger.LogInformation( + "Skipping ResetAll enqueue because one is already queued or processing." + ); + + var queuedCheck = await _jobRegistry.TryMarkForCheckUpdatesAll(); + if (!queuedCheck) + _logger.LogInformation( + "Skipping CheckForUpdatesAll enqueue because one is already queued or processing." + ); + } + catch (Exception ex) + { + _logger.LogError(ex, "Error while enqueueing scheduled version check jobs"); + } + finally + { + Interlocked.Exchange(ref _isRunning, 0); + } } } diff --git a/PatchPanda.Web/Services/Interfaces/IPortainerService.cs b/PatchPanda.Web/Services/Interfaces/IPortainerService.cs index c33bb55..d2d55af 100644 --- a/PatchPanda.Web/Services/Interfaces/IPortainerService.cs +++ b/PatchPanda.Web/Services/Interfaces/IPortainerService.cs @@ -6,9 +6,16 @@ public interface IPortainerService bool IsAccessTokenConfigured { get; } - Task ValidateAccessTokenAsync(); + Task ValidateAccessTokenAsync(CancellationToken cancellationToken = default); - Task GetStackFileContentAsync(string stackName); + Task GetStackFileContentAsync( + string stackName, + CancellationToken cancellationToken = default + ); - Task UpdateStackFileContentAsync(string stackName, string newFileContent); + Task UpdateStackFileContentAsync( + string stackName, + string newFileContent, + CancellationToken cancellationToken = default + ); } diff --git a/PatchPanda.Web/Services/PortainerService.cs b/PatchPanda.Web/Services/PortainerService.cs index c211536..75b73ee 100644 --- a/PatchPanda.Web/Services/PortainerService.cs +++ b/PatchPanda.Web/Services/PortainerService.cs @@ -41,6 +41,7 @@ ILogger logger var ignoreSsl = configuration.GetValue(Constants.VariableKeys.PORTAINER_IGNORE_SSL); _username = configuration.GetValue(Constants.VariableKeys.PORTAINER_USERNAME); _password = configuration.GetValue(Constants.VariableKeys.PORTAINER_PASSWORD); + var timeoutSeconds = Constants.Limits.PORTAINER_HTTP_TIMEOUT_SECONDS; if (!IsUrlConfigured) { @@ -75,7 +76,8 @@ ILogger logger _httpClient = new HttpClient(handler) { - BaseAddress = new Uri(_url!) + BaseAddress = new Uri(_url!), + Timeout = TimeSpan.FromSeconds(Math.Max(1, timeoutSeconds)) }; if (IsAccessTokenConfigured) @@ -86,14 +88,14 @@ ILogger logger logger.LogInformation("PortainerService initialized with URL: {Url}", _url); } - public async Task ValidateAccessTokenAsync() + public async Task ValidateAccessTokenAsync(CancellationToken cancellationToken = default) { if (!IsConfigured || !IsAccessTokenConfigured || _httpClient is null) return false; try { - var response = await _httpClient.GetAsync("/api/motd"); + var response = await _httpClient.GetAsync("/api/motd", cancellationToken); if (response.IsSuccessStatusCode) { @@ -119,7 +121,7 @@ public async Task ValidateAccessTokenAsync() } } - private async Task EnsureAuthenticatedAsync() + private async Task EnsureAuthenticatedAsync(CancellationToken cancellationToken = default) { if (!IsUrlConfigured || IsAccessTokenConfigured || !IsUsernamePasswordConfigured || _httpClient is null) return; @@ -140,7 +142,8 @@ private async Task EnsureAuthenticatedAsync() var payload = JsonSerializer.Serialize(new { username = _username, password = _password }); var resp = await _httpClient.PostAsync( "/api/auth", - new StringContent(payload, Encoding.UTF8, "application/json") + new StringContent(payload, Encoding.UTF8, "application/json"), + cancellationToken ); if (!resp.IsSuccessStatusCode) @@ -155,7 +158,9 @@ private async Task EnsureAuthenticatedAsync() return; } - var json = await resp.Content.ReadFromJsonAsync(); + var json = await resp.Content.ReadFromJsonAsync( + cancellationToken: cancellationToken + ); if (json?.Jwt is not null) { @@ -170,15 +175,21 @@ private async Task EnsureAuthenticatedAsync() _logger.LogWarning("Failed parsing Portainer auth response"); } - private async Task GetStack(string stackName) + private async Task GetStack( + string stackName, + CancellationToken cancellationToken = default + ) { if (!IsConfigured || _httpClient is null) return null; - await EnsureAuthenticatedAsync(); + await EnsureAuthenticatedAsync(cancellationToken); var filters = JsonSerializer.Serialize(new { Name = stackName }); - var resp = await _httpClient.GetAsync($"/api/stacks?filters={filters}"); + var resp = await _httpClient.GetAsync( + $"/api/stacks?filters={filters}", + cancellationToken + ); if (!resp.IsSuccessStatusCode) { @@ -190,7 +201,9 @@ private async Task EnsureAuthenticatedAsync() return null; } - var stacks = await resp.Content.ReadFromJsonAsync(); + var stacks = await resp.Content.ReadFromJsonAsync( + cancellationToken: cancellationToken + ); if (stacks is null || stacks.Length == 0) { @@ -204,15 +217,19 @@ private async Task EnsureAuthenticatedAsync() return stacks[0]; } - public async Task GetStackFileContentAsync(string stackName) + public async Task GetStackFileContentAsync( + string stackName, + CancellationToken cancellationToken = default + ) { - var first = await GetStack(stackName); + var first = await GetStack(stackName, cancellationToken); if (first is null) return null; var fileResp = await _httpClient!.GetAsync( - $"/api/stacks/{first.Id}/file?endpointId={first.EndpointId}" + $"/api/stacks/{first.Id}/file?endpointId={first.EndpointId}", + cancellationToken ); if (!fileResp.IsSuccessStatusCode) @@ -225,13 +242,19 @@ private async Task EnsureAuthenticatedAsync() return null; } - var fileDto = await fileResp.Content.ReadFromJsonAsync(); + var fileDto = await fileResp.Content.ReadFromJsonAsync( + cancellationToken: cancellationToken + ); return fileDto?.StackFileContent; } - public async Task UpdateStackFileContentAsync(string stackName, string newFileContent) + public async Task UpdateStackFileContentAsync( + string stackName, + string newFileContent, + CancellationToken cancellationToken = default + ) { - var first = await GetStack(stackName); + var first = await GetStack(stackName, cancellationToken); if (first is null) throw new("No such stack found."); @@ -241,7 +264,8 @@ public async Task UpdateStackFileContentAsync(string stackName, string newFileCo ); var putResp = await _httpClient!.PutAsync( $"/api/stacks/{first.Id}?endpointId={first.EndpointId}", - new StringContent(payload, Encoding.UTF8, "application/json") + new StringContent(payload, Encoding.UTF8, "application/json"), + cancellationToken ); if (!putResp.IsSuccessStatusCode) diff --git a/PatchPanda.Web/Services/UpdateService.cs b/PatchPanda.Web/Services/UpdateService.cs index 08f66f8..c5c607e 100644 --- a/PatchPanda.Web/Services/UpdateService.cs +++ b/PatchPanda.Web/Services/UpdateService.cs @@ -78,9 +78,9 @@ public bool IsUpdateAvailable(Container app) => && app.GitHubVersionRegex is not null && app.Version is not null; - public async Task CheckAllForUpdates() + public async Task CheckAllForUpdates(CancellationToken cancellationToken = default) { - await using var db = await _dbContextFactory.CreateDbContextAsync(); + await using var db = await _dbContextFactory.CreateDbContextAsync(cancellationToken); var containers = await db .Containers.Include(x => x.NewerVersions) @@ -89,7 +89,7 @@ public async Task CheckAllForUpdates() && !x.IgnoreContainer && (x.GitHubRepo != null || x.OverrideGitHubRepo != null) ) - .ToListAsync(); + .ToListAsync(cancellationToken); var uniqueRepoGroups = containers .GroupBy(x => x.GetGitHubRepo()!) @@ -97,6 +97,8 @@ public async Task CheckAllForUpdates() foreach (var uniqueRepoGroup in uniqueRepoGroups) { + cancellationToken.ThrowIfCancellationRequested(); + _logger.LogDebug("Checking unique repo group: {Repo}", uniqueRepoGroup.Key); var currentVersionGroups = uniqueRepoGroup @@ -131,7 +133,7 @@ [.. otherApps] { var container = await db .Containers.Include(x => x.NewerVersions) - .FirstAsync(x => x.Id == mainApp.Id); + .FirstAsync(x => x.Id == mainApp.Id, cancellationToken); var toNotify = container.NewerVersions.Where(x => !x.Notified).ToList(); var result = await _notificationService.SendNewVersion( @@ -147,7 +149,7 @@ [.. otherApps] foreach (var v in toNotify) v.Notified = true; - await db.SaveChangesAsync(); + await db.SaveChangesAsync(cancellationToken); } catch (Exception ex) { @@ -171,12 +173,19 @@ [.. otherApps] } } - await ProcessAutoUpdates(db); + await ProcessAutoUpdates(db, cancellationToken); } - private async Task ProcessAutoUpdates(DataContext db) + private async Task ProcessAutoUpdates( + DataContext db, + CancellationToken cancellationToken = default + ) { - var settings = await db.AppSettings.ToDictionaryAsync(x => x.Key, x => x.Value); + var settings = await db.AppSettings.ToDictionaryAsync( + x => x.Key, + x => x.Value, + cancellationToken + ); if ( !settings.TryGetValue(Constants.SettingsKeys.AUTO_UPDATE_ENABLED, out var enabledStr) @@ -204,10 +213,12 @@ private async Task ProcessAutoUpdates(DataContext db) && v.DateDiscovered < threshold ) ) - .ToListAsync(); + .ToListAsync(cancellationToken); foreach (var container in candidates) { + cancellationToken.ThrowIfCancellationRequested(); + if ( _jobRegistry.GetQueuedUpdateForContainer(container.Id) is not null || _jobRegistry.GetProcessingUpdateForContainer(container.Id) is not null @@ -218,7 +229,7 @@ private async Task ProcessAutoUpdates(DataContext db) .UpdateAttempts.Where(x => x.ContainerId == container.Id) .OrderByDescending(x => x.EndedAt) .Take(10) - .ToListAsync(); + .ToListAsync(cancellationToken); var consecutiveFailures = 0; var lastFailureTime = DateTime.MinValue; @@ -278,7 +289,7 @@ private async Task ProcessAutoUpdates(DataContext db) if (targetVersion == null) continue; - var plan = await Update(container, true, targetVersion); + var plan = await Update(container, true, targetVersion, cancellationToken: cancellationToken); if (plan.Steps is null) continue; @@ -319,7 +330,8 @@ public async Task Update( bool planOnly, AppVersion targetVersion, Action? outputCallback = null, - bool isAutomatic = false + bool isAutomatic = false, + CancellationToken cancellationToken = default ) { if (!IsUpdateAvailable(app)) @@ -341,8 +353,10 @@ public async Task Update( ArgumentNullException.ThrowIfNull(app.GitHubVersionRegex); ArgumentNullException.ThrowIfNull(app.Version); - await using var db = await _dbContextFactory.CreateDbContextAsync(); - var stack = await db.Stacks.FirstAsync(x => x.Id == app.StackId); + await using var db = await _dbContextFactory.CreateDbContextAsync( + CancellationToken.None + ); + var stack = await db.Stacks.FirstAsync(x => x.Id == app.StackId, CancellationToken.None); var configPath = stack.ConfigFile; if (configPath is null && (!stack.PortainerManaged || !_portainerService.IsConfigured)) @@ -361,7 +375,8 @@ public async Task Update( else { configFileContent = await _portainerService.GetStackFileContentAsync( - stack.StackName + stack.StackName, + cancellationToken ); if (configFileContent is null) @@ -461,7 +476,7 @@ public async Task Update( { var targetMultiContainer = await db .MultiContainerApps.Include(x => x.Containers) - .FirstAsync(x => x.Id == app.MultiContainerAppId); + .FirstAsync(x => x.Id == app.MultiContainerAppId, cancellationToken); appsWithSharedEnvVersion = [ .. targetMultiContainer.Containers.Where(x => @@ -530,7 +545,8 @@ .. targetMultiContainer.Containers.Where(x => { await _portainerService.UpdateStackFileContentAsync( stack.StackName, - configFileContent.Replace(app.TargetImage, resultingImage) + configFileContent.Replace(app.TargetImage, resultingImage), + cancellationToken ); } catch (Exception ex) @@ -547,14 +563,15 @@ await _portainerService.UpdateStackFileContentAsync( if (attemptCount > 0) { var delayMs = (int)Math.Pow(2, attemptCount) * 1000; - await Task.Delay(delayMs); + await Task.Delay(delayMs, CancellationToken.None); } try { await _portainerService.UpdateStackFileContentAsync( stack.StackName, - configFileContent + configFileContent, + CancellationToken.None ); rollbackStdErr += $"\n[ROLLBACK SUCCESS]\n"; @@ -604,7 +621,12 @@ await _portainerService.UpdateStackFileContentAsync( try { (string pullStdOut, string pullStdErr, int pullExitCode) = - await _dockerService.RunDockerComposeOnPath(stack, "pull", outputCallback); + await _dockerService.RunDockerComposeOnPath( + stack, + "pull", + outputCallback, + cancellationToken + ); combinedStdOuts += "[PULL STDOUT]\n" + pullStdOut + "\nExit code: " + pullExitCode; @@ -612,7 +634,12 @@ await _portainerService.UpdateStackFileContentAsync( "[PULL STDERR]\n" + pullStdErr + "\nExit code: " + pullExitCode; (string upStdOut, string upStdErr, int upExitCode) = - await _dockerService.RunDockerComposeOnPath(stack, "up -d", outputCallback); + await _dockerService.RunDockerComposeOnPath( + stack, + "up -d", + outputCallback, + cancellationToken + ); combinedStdOuts += "\n[UP STDOUT]\n" + upStdOut + "\nExit code: " + upExitCode; combinedStdErrs += "\n[UP STDERR]\n" + upStdErr + "\nExit code: " + upExitCode; @@ -638,14 +665,15 @@ envFile is not null if (attemptCount > 0) { var delayMs = (int)Math.Pow(2, attemptCount) * 1000; // 2s, 4s, 8s - await Task.Delay(delayMs); + await Task.Delay(delayMs, CancellationToken.None); } (string reupStdOut, string reupStdErr, int reupExitCode) = await _dockerService.RunDockerComposeOnPath( stack, "up -d", - outputCallback + outputCallback, + CancellationToken.None ); if (reupExitCode != 0) @@ -679,7 +707,7 @@ await _dockerService.RunDockerComposeOnPath( var targetApp = await db .Containers.Include(x => x.NewerVersions) - .FirstAsync(x => x.Id == app.Id); + .FirstAsync(x => x.Id == app.Id, cancellationToken); var removedVersions = targetApp.NewerVersions.RemoveAll(x => targetVersion.Id == x.Id || targetVersion.VersionNumber.IsNewerThan(x.VersionNumber) @@ -692,7 +720,7 @@ await _dockerService.RunDockerComposeOnPath( && c.Id != targetApp.Id && c.TargetImage == targetApp.TargetImage ) - .ToListAsync(); + .ToListAsync(cancellationToken); if (resultingImage is not null) { @@ -725,7 +753,7 @@ await _dockerService.RunDockerComposeOnPath( sideEffectUpdated = await db .Containers.Include(x => x.NewerVersions) .Where(c => appsWithSharedEnvVersion.Select(x => x.Id).Contains(c.Id)) - .ToListAsync(); + .ToListAsync(cancellationToken); foreach (var sideApp in sideEffectUpdated) { @@ -752,7 +780,7 @@ await _dockerService.RunDockerComposeOnPath( StdOut = combinedStdOuts, } ); - await db.SaveChangesAsync(); + await db.SaveChangesAsync(CancellationToken.None); await _notificationService.SendAutoUpdateResult( app, @@ -775,8 +803,10 @@ await _notificationService.SendAutoUpdateResult( { // DockerCommandException bubbles up from inner try-catch after rollback is performed // This catch handles post-failure cleanup: UpdateAttempt logging, job cleanup, notifications - await using var db = await _dbContextFactory.CreateDbContextAsync(); - var stack = await db.Stacks.FirstAsync(x => x.Id == app.StackId); + await using var db = await _dbContextFactory.CreateDbContextAsync( + CancellationToken.None + ); + var stack = await db.Stacks.FirstAsync(x => x.Id == app.StackId, CancellationToken.None); if (rollbackFailed) rollbackStdErr += $"\n{LogAndGetRollbackFailedMessage()}"; @@ -795,7 +825,7 @@ await _notificationService.SendAutoUpdateResult( UsedPlan = string.Join(", ", updateSteps), } ); - await db.SaveChangesAsync(); + await db.SaveChangesAsync(CancellationToken.None); CleanUpAllContainerJobs(app.Id, relatedApps, sideEffectUpdated); @@ -814,8 +844,13 @@ await _notificationService.SendAutoUpdateResult( { if (!planOnly) { - await using var db = await _dbContextFactory.CreateDbContextAsync(); - var stack = await db.Stacks.FirstAsync(x => x.Id == app.StackId); + await using var db = await _dbContextFactory.CreateDbContextAsync( + CancellationToken.None + ); + var stack = await db.Stacks.FirstAsync( + x => x.Id == app.StackId, + CancellationToken.None + ); db.UpdateAttempts.Add( new() @@ -837,7 +872,7 @@ await _notificationService.SendAutoUpdateResult( UsedPlan = string.Join(", ", updateSteps), } ); - await db.SaveChangesAsync(); + await db.SaveChangesAsync(CancellationToken.None); CleanUpAllContainerJobs(app.Id, relatedApps, sideEffectUpdated);