Execute confirmed timezone provisioning safely #6
6 changed files with 207 additions and 9 deletions
|
|
@ -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 (безопасный первый вариант).
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -203,8 +203,37 @@ public sealed partial class ControlStore(IOptions<ControlOptions> 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<ControlOptions> 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)
|
||||
|
|
|
|||
|
|
@ -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<string, string[]>
|
||||
{
|
||||
["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,
|
||||
|
|
|
|||
|
|
@ -258,6 +258,86 @@ public sealed partial class ControlStore
|
|||
return updated;
|
||||
}
|
||||
|
||||
public async Task<ProvisioningJob?> 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<ProvisioningJob?> TransitionProvisioningJobAsync(
|
||||
string id,
|
||||
ProvisioningJobCommandRequest request,
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in a new issue