From 41d33a5ce012ea02e7d0c6d9493b6650895a626a Mon Sep 17 00:00:00 2001 From: Ochenstarik Date: Sun, 19 Jul 2026 22:00:38 +0700 Subject: [PATCH] Reconcile expired provisioning jobs --- docs/roadmap.md | 2 +- .../ControlMaintenance.cs | 15 +++- .../ControlStore.cs | 39 ++++++++- src/ServerMonitorManager.Control/Program.cs | 31 +++++++ .../ProvisioningStore.cs | 80 +++++++++++++++++++ .../ControlMaintenanceTests.cs | 49 ++++++++++++ 6 files changed, 207 insertions(+), 9 deletions(-) diff --git a/docs/roadmap.md b/docs/roadmap.md index 7750855..bbd4e30 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -118,7 +118,7 @@ - [ ] versioned JSON schemas для остальных action type; - [ ] restricted root helper через Unix socket; - [ ] structured redacted events и progress; -- [ ] `NeedsReconciliation` после неопределённого результата; +- [x] `NeedsReconciliation` после истечения execution TTL и неопределённого результата; - [ ] desired/factual configuration и drift; - [x] запрет параллельных активных заданий на одном Node (безопасный первый вариант). diff --git a/src/ServerMonitorManager.Control/ControlMaintenance.cs b/src/ServerMonitorManager.Control/ControlMaintenance.cs index 10285c1..de3db45 100644 --- a/src/ServerMonitorManager.Control/ControlMaintenance.cs +++ b/src/ServerMonitorManager.Control/ControlMaintenance.cs @@ -10,7 +10,9 @@ public sealed record ControlMaintenanceResult( int MetricsDeleted, int IdempotencyDeleted, int AuditDeleted, - int TokensDeleted); + int TokensDeleted, + int ProvisioningJobsCancelled, + int ProvisioningJobsNeedingReconciliation); public sealed class LinkExpirationBackgroundService( LinkService links, @@ -65,15 +67,20 @@ public sealed class ControlMaintenanceBackgroundService( { var result = await store.MaintainAsync(timeProvider.GetUtcNow(), stoppingToken); await backups.CreateIfDueAsync(timeProvider.GetUtcNow(), stoppingToken); - if (result.MetricsDeleted + result.IdempotencyDeleted + result.AuditDeleted + result.TokensDeleted > 0) + if (result.MetricsDeleted + result.IdempotencyDeleted + result.AuditDeleted + + result.TokensDeleted + result.ProvisioningJobsCancelled + + result.ProvisioningJobsNeedingReconciliation > 0) { logger.LogInformation( "Control maintenance removed {Metrics} metrics, {Idempotency} replay records, " - + "{Audit} audit records, and {Tokens} enrollment tokens.", + + "{Audit} audit records, and {Tokens} enrollment tokens; cancelled {Cancelled} " + + "expired jobs and marked {Reconciliation} jobs for reconciliation.", result.MetricsDeleted, result.IdempotencyDeleted, result.AuditDeleted, - result.TokensDeleted); + result.TokensDeleted, + result.ProvisioningJobsCancelled, + result.ProvisioningJobsNeedingReconciliation); } } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) diff --git a/src/ServerMonitorManager.Control/ControlStore.cs b/src/ServerMonitorManager.Control/ControlStore.cs index bcb5370..478f698 100644 --- a/src/ServerMonitorManager.Control/ControlStore.cs +++ b/src/ServerMonitorManager.Control/ControlStore.cs @@ -203,8 +203,37 @@ public sealed partial class ControlStore(IOptions options) CancellationToken cancellationToken = default) { await using var connection = await OpenAsync(cancellationToken); + await using var transaction = + (SqliteTransaction)await connection.BeginTransactionAsync(cancellationToken); var command = connection.CreateCommand(); + command.Transaction = transaction; command.CommandText = """ + INSERT INTO provisioning_events(job_id, recorded_at, event_type, state, message) + SELECT id, $now, 'job.expired', 'Cancelled', + 'Provisioning job expired before execution.' + FROM provisioning_jobs + WHERE expires_at <= $now + AND state IN ('Queued', 'AwaitingConfirmation'); + UPDATE provisioning_jobs SET + state = 'Cancelled', cancelled_at = $now, updated_at = $now, + current_step = 'expired', version = version + 1, + last_error = 'job.ttl_expired' + WHERE expires_at <= $now + AND state IN ('Queued', 'AwaitingConfirmation'); + SELECT changes(); + INSERT INTO provisioning_events(job_id, recorded_at, event_type, state, message) + SELECT id, $now, 'job.reconciliation.required', 'NeedsReconciliation', + 'Execution lease expired; factual state must be inspected.' + FROM provisioning_jobs + WHERE expires_at <= $now + AND state IN ('Preflight', 'Running', 'Verifying'); + UPDATE provisioning_jobs SET + state = 'NeedsReconciliation', updated_at = $now, + current_step = 'reconcile', version = version + 1, + last_error = 'job.ttl_expired' + WHERE expires_at <= $now + AND state IN ('Preflight', 'Running', 'Verifying'); + SELECT changes(); DELETE FROM metric_samples WHERE recorded_at < $metric_cutoff; SELECT changes(); DELETE FROM idempotency WHERE created_at < $idempotency_cutoff; @@ -225,24 +254,26 @@ public sealed partial class ControlStore(IOptions options) command.Parameters.AddWithValue( "$audit_cutoff", now.AddDays(-_options.AuditRetentionDays).ToString("O")); command.Parameters.AddWithValue("$now", now.ToString("O")); - var deleted = new int[6]; + var changes = new int[8]; await using (var reader = await command.ExecuteReaderAsync(cancellationToken)) { - for (var index = 0; index < deleted.Length; index++) + for (var index = 0; index < changes.Length; index++) { if (await reader.ReadAsync(cancellationToken)) { - deleted[index] = reader.GetInt32(0); + changes[index] = reader.GetInt32(0); } await reader.NextResultAsync(cancellationToken); } } + await transaction.CommitAsync(cancellationToken); var optimize = connection.CreateCommand(); optimize.CommandText = "PRAGMA optimize; PRAGMA wal_checkpoint(PASSIVE);"; await optimize.ExecuteNonQueryAsync(cancellationToken); return new ControlMaintenanceResult( - deleted[0], deleted[1], deleted[2], deleted[3] + deleted[4] + deleted[5]); + changes[2], changes[3], changes[4], changes[5] + changes[6] + changes[7], + changes[0], changes[1]); } public async Task BackupDatabaseAsync(string destinationPath, CancellationToken cancellationToken = default) diff --git a/src/ServerMonitorManager.Control/Program.cs b/src/ServerMonitorManager.Control/Program.cs index 31a6ed5..ce82667 100644 --- a/src/ServerMonitorManager.Control/Program.cs +++ b/src/ServerMonitorManager.Control/Program.cs @@ -500,6 +500,37 @@ control.MapPost("/provisioning/jobs/{id}/cancel", async ( CancellationToken cancellationToken) => await ChangeProvisioningJobAsync( id, request, context, controlStore, confirm: false, cancellationToken)); +control.MapPost("/provisioning/jobs/{id}/retry", async ( + string id, + ProvisioningJobCommandRequest request, + HttpContext context, + ControlStore controlStore, + CancellationToken cancellationToken) => +{ + if (!ProvisioningJobValidator.IsValidId(id) + || !ProvisioningJobValidator.IsValid(request)) + { + return Results.ValidationProblem(new Dictionary + { + ["provisioningJob"] = ["Invalid job id, reason, or idempotency key."] + }); + } + try + { + var actor = context.User.FindFirstValue(ClaimTypes.NameIdentifier)!; + var job = await controlStore.RetryProvisioningJobAsync( + id, request, actor, cancellationToken); + return job is null ? Results.NotFound() : Results.Ok(job); + } + catch (IdempotencyConflictException) + { + return Results.Conflict(new ProblemDetails { Title = "Idempotency key conflict" }); + } + catch (ProvisioningTransitionException exception) + { + return Results.Conflict(new ProblemDetails { Title = exception.Message }); + } +}); control.MapPost("/automations/token", async ( AutomationTokenCreateRequest request, HttpContext context, diff --git a/src/ServerMonitorManager.Control/ProvisioningStore.cs b/src/ServerMonitorManager.Control/ProvisioningStore.cs index 35c7484..c5676bf 100644 --- a/src/ServerMonitorManager.Control/ProvisioningStore.cs +++ b/src/ServerMonitorManager.Control/ProvisioningStore.cs @@ -258,6 +258,86 @@ public sealed partial class ControlStore return updated; } + public async Task RetryProvisioningJobAsync( + string id, + ProvisioningJobCommandRequest request, + string actor, + CancellationToken cancellationToken = default) + { + await using var connection = await OpenAsync(cancellationToken); + await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync(cancellationToken); + var operationKey = $"provisioning-retry:{actor}:{id}:{request.IdempotencyKey}"; + var requestHash = Fingerprint(request, SmmJsonContext.Default.ProvisioningJobCommandRequest); + var cached = await ReadIdempotentAsync( + connection, transaction, operationKey, requestHash, + SmmJsonContext.Default.ProvisioningJob, cancellationToken); + if (cached is not null) + { + await transaction.CommitAsync(cancellationToken); + return cached; + } + + var current = await ReadProvisioningJobAsync(connection, transaction, id, cancellationToken); + if (current is null) + { + await transaction.RollbackAsync(cancellationToken); + return null; + } + if (current.State is not (ProvisioningJobStates.Failed + or ProvisioningJobStates.NeedsReconciliation)) + { + throw new ProvisioningTransitionException(current.State, ProvisioningJobStates.Queued); + } + + var now = DateTimeOffset.UtcNow; + var originalLifetime = current.ExpiresAt - current.CreatedAt; + var retryLifetime = originalLifetime < TimeSpan.FromMinutes(5) + ? TimeSpan.FromMinutes(5) + : originalLifetime; + var updated = current with + { + State = ProvisioningJobStates.Queued, + ProgressPercent = 0, + CurrentStep = "retry-queued", + UpdatedAt = now, + ExpiresAt = now.Add(retryLifetime), + Version = current.Version + 1, + LastError = null + }; + var update = connection.CreateCommand(); + update.Transaction = transaction; + update.CommandText = """ + UPDATE provisioning_jobs SET + state = $state, progress_percent = 0, current_step = $step, + updated_at = $updated, expires_at = $expires, + version = $version, last_error = NULL + WHERE id = $id AND version = $previous_version; + """; + update.Parameters.AddWithValue("$state", updated.State); + update.Parameters.AddWithValue("$step", updated.CurrentStep); + update.Parameters.AddWithValue("$updated", updated.UpdatedAt.ToString("O")); + update.Parameters.AddWithValue("$expires", updated.ExpiresAt.ToString("O")); + update.Parameters.AddWithValue("$version", updated.Version); + update.Parameters.AddWithValue("$id", id); + update.Parameters.AddWithValue("$previous_version", current.Version); + if (await update.ExecuteNonQueryAsync(cancellationToken) != 1) + { + throw new ProvisioningTransitionException(current.State, ProvisioningJobStates.Queued); + } + + await WriteProvisioningEventAsync( + connection, transaction, id, "job.retry.queued", updated.State, + request.Reason, now, cancellationToken); + await WriteIdempotentAsync( + connection, transaction, operationKey, requestHash, updated, + SmmJsonContext.Default.ProvisioningJob, cancellationToken); + await WriteAuditAsync( + connection, transaction, actor, "provisioning.job.retry", id, + JsonSerializer.Serialize(new { request.Reason }), cancellationToken); + await transaction.CommitAsync(cancellationToken); + return updated; + } + private async Task TransitionProvisioningJobAsync( string id, ProvisioningJobCommandRequest request, diff --git a/tests/ServerMonitorManager.Control.Tests/ControlMaintenanceTests.cs b/tests/ServerMonitorManager.Control.Tests/ControlMaintenanceTests.cs index f70a384..7d26549 100644 --- a/tests/ServerMonitorManager.Control.Tests/ControlMaintenanceTests.cs +++ b/tests/ServerMonitorManager.Control.Tests/ControlMaintenanceTests.cs @@ -1,6 +1,7 @@ using Microsoft.Data.Sqlite; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; +using System.Text.Json; using ServerMonitorManager.Control; using ServerMonitorManager.Core; using Xunit; @@ -108,6 +109,54 @@ public sealed class ControlMaintenanceTests : IAsyncDisposable Assert.Equal(3L, (long)(await version.ExecuteScalarAsync(cancellationToken))!); } + [Fact] + public async Task ExpiredProvisioningJobsAreCancelledOrRequireReconciliationAndCanRetry() + { + var cancellationToken = TestContext.Current.CancellationToken; + var (store, _) = CreateServices(); + await store.InitializeAsync(cancellationToken); + await EnrollAgentAsync(store, "queued-node", "AABB", cancellationToken); + await EnrollAgentAsync(store, "running-node", "CCDD", cancellationToken); + using var parameters = JsonDocument.Parse("{}"); + var queued = await store.CreateProvisioningJobAsync( + "queued-node", + new ProvisioningJobCreateRequest( + "system.base-install", 1, parameters.RootElement.Clone(), 5, + "Await approval", Guid.NewGuid().ToString()), + "operator", + cancellationToken); + var running = await store.CreateProvisioningJobAsync( + "running-node", + new ProvisioningJobCreateRequest( + "preflight", 1, parameters.RootElement.Clone(), 5, + "Inspect node", Guid.NewGuid().ToString()), + "operator", + cancellationToken); + Assert.NotNull(await store.ClaimNextProvisioningJobAsync("running-node", cancellationToken)); + + var result = await store.MaintainAsync( + running.ExpiresAt.AddSeconds(1), cancellationToken); + + Assert.Equal(1, result.ProvisioningJobsCancelled); + Assert.Equal(1, result.ProvisioningJobsNeedingReconciliation); + Assert.Equal(ProvisioningJobStates.Cancelled, + (await store.GetProvisioningJobAsync(queued.Id, cancellationToken))!.State); + var reconciliation = await store.GetProvisioningJobAsync(running.Id, cancellationToken); + Assert.Equal(ProvisioningJobStates.NeedsReconciliation, reconciliation!.State); + Assert.Equal("job.ttl_expired", reconciliation.LastError); + + var retried = await store.RetryProvisioningJobAsync( + running.Id, + new ProvisioningJobCommandRequest("Factual state checked", Guid.NewGuid().ToString()), + "operator", + cancellationToken); + Assert.Equal(ProvisioningJobStates.Queued, retried!.State); + Assert.Equal(0, retried.ProgressPercent); + Assert.Null(retried.LastError); + Assert.True(retried.ExpiresAt > running.ExpiresAt); + Assert.NotNull(await store.ClaimNextProvisioningJobAsync("running-node", cancellationToken)); + } + [Fact] public async Task BackupCanRestoreDatabaseAndCertificateAuthority() {