mirror of
https://github.com/okieselbach/AutopilotMonitor.git
synced 2026-10-24 04:32:32 +00:00
feat(backend/backup): PR1 — critical-table backup feature (daily timer + manual GA trigger)
Adds the full backup pipeline that snapshots 15 critical tables (Admin/Tenant config, rules, identities, audit, preview) into a logically-isolated blob container, daily at 04:00 UTC and on operator demand. Restore endpoints follow in PR2/PR3. Architecture: * Daily TimerTrigger + manual POST /api/global/backups/trigger (GA-only, fail-hard producer). HTTP returns 202 + jobId on enqueue, 500 if the queue write fails (and the BackupJobs row is rolled to Failed). * BackgroundService queue worker (not QueueTrigger — matches the Analyze/ IndexReconcile pattern that avoids the Functions-host connection-binding app-setting). Worker holds the maintenance lease for the whole run; service is intentionally lease-unaware so it stays unit-testable. * BlobBackupStore wraps the new 'critical-table-backups' container with a shared maintenance-lease sentinel at '_lock/maintenance.lock' that serialises every backup or restore op across worker/timer/single-row paths. * CriticalTableBackupService streams each table through CryptoStream(SHA256) into per-table NDJSON blobs, then writes manifest.json LAST as the durability anchor. Per-table failures degrade Outcome to Partial without aborting; per-run budget (50 min) → Skipped status. * BackupJobsRepository persists the six-state machine (Queued/Running/Completed/Failed/Skipped/BlockedTerminal) with ETag-CAS updates. Enum→string mapping rejects unknown values on read so refactors surface loudly instead of silently shifting state. * BackupJobWatchdog (30 min timer) heals stuck rows via a probe lease on the shared sentinel: Queued > 60 min → Failed, Running with stale heartbeat AND free lease → Failed (lease-busy means a live worker — leave alone). * OpsEvents: distinct CriticalTableBackupPartial (Warning) vs Failed (Error) so operators can wire softer Telegram rules for partial captures. * MaintenanceService.MonitoredPoisonQueues + OPS_EVENT_TYPES catalog registered (dual-register memory). Test coverage (15 new tests, 1952/1952 backend xUnit green): * BackupManifestJsonTests: enum-as-string (allowIntegerValues=false rejects numeric values), DictionaryKeyPolicy unset (Azure column names preserved case-faithfully), manifest field round-trip. * BackupJobsRepositoryMappingTests: full entity↔domain round-trip, unknown Kind/State/BackupOutcome rejected, legacy BackupOutcome="Failed" rejected. * CriticalTableBackupServiceTests: happy/partial/empty/budget-exceeded paths exercise the per-table loop with in-memory table + blob fakes. Setup: scripts/Backup/setup-backup-container.ps1 — idempotent az-CLI bootstrap for container + lifecycle rule (90 d delete on prefix). Soft-Delete and Versioning are NOT applied automatically (storage-account-wide blast radius — operator must validate per separate doc note). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
@@ -0,0 +1,92 @@
|
||||
# setup-backup-container.ps1
|
||||
#
|
||||
# Idempotent setup for the critical-table backup feature (plan §PR1).
|
||||
#
|
||||
# Creates the 'critical-table-backups' container in the backend storage account
|
||||
# and applies a lifecycle rule that deletes blobs older than 90 days under the
|
||||
# 'critical-table-backups/' prefix. Run once per environment, then again any
|
||||
# time the storage account is rebuilt.
|
||||
#
|
||||
# WARNING (plan §Wave15 Azure-Hinweis): Blob Versioning and Blob Soft-Delete
|
||||
# are STORAGE-ACCOUNT settings (not container-scoped). Enabling them affects
|
||||
# every container in the account (e.g. 'diagnostics', 'offboarding-state').
|
||||
# This script does NOT enable them automatically — operator must validate the
|
||||
# blast radius separately and run the corresponding `az storage account
|
||||
# blob-service-properties update` command if desired.
|
||||
|
||||
[CmdletBinding()]
|
||||
param(
|
||||
[Parameter(Mandatory = $true)] [string] $StorageAccountName,
|
||||
[Parameter(Mandatory = $true)] [string] $ResourceGroupName,
|
||||
[string] $ContainerName = 'critical-table-backups',
|
||||
[int] $RetentionDays = 90
|
||||
)
|
||||
|
||||
$ErrorActionPreference = 'Stop'
|
||||
|
||||
Write-Host "==> Verifying Azure CLI login"
|
||||
az account show --only-show-errors > $null
|
||||
|
||||
Write-Host "==> Ensuring container '$ContainerName' in account '$StorageAccountName'"
|
||||
az storage container create `
|
||||
--account-name $StorageAccountName `
|
||||
--name $ContainerName `
|
||||
--auth-mode login `
|
||||
--only-show-errors `
|
||||
--fail-on-exist false | Out-Null
|
||||
|
||||
Write-Host "==> Building lifecycle rule (delete > $RetentionDays days, prefix '$ContainerName/')"
|
||||
$policyJson = @"
|
||||
{
|
||||
"rules": [
|
||||
{
|
||||
"enabled": true,
|
||||
"name": "delete-critical-table-backups",
|
||||
"type": "Lifecycle",
|
||||
"definition": {
|
||||
"actions": {
|
||||
"baseBlob": {
|
||||
"delete": { "daysAfterModificationGreaterThan": $RetentionDays }
|
||||
}
|
||||
},
|
||||
"filters": {
|
||||
"blobTypes": [ "blockBlob" ],
|
||||
"prefixMatch": [ "$ContainerName/" ]
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
"@
|
||||
|
||||
$tmpFile = [System.IO.Path]::GetTempFileName()
|
||||
try {
|
||||
Set-Content -Path $tmpFile -Value $policyJson -Encoding UTF8
|
||||
Write-Host "==> Applying lifecycle policy to storage account '$StorageAccountName'"
|
||||
Write-Host " NOTE: this REPLACES the account's lifecycle policy. If other rules"
|
||||
Write-Host " exist (e.g. for 'deletion-manifests'), merge them in manually."
|
||||
az storage account management-policy create `
|
||||
--account-name $StorageAccountName `
|
||||
--resource-group $ResourceGroupName `
|
||||
--policy "@$tmpFile" `
|
||||
--only-show-errors | Out-Null
|
||||
}
|
||||
finally {
|
||||
Remove-Item -Path $tmpFile -ErrorAction SilentlyContinue
|
||||
}
|
||||
|
||||
Write-Host ""
|
||||
Write-Host "==> Done."
|
||||
Write-Host " Container ready: $ContainerName"
|
||||
Write-Host " Lifecycle: delete blobs > $RetentionDays days under '$ContainerName/' prefix"
|
||||
Write-Host ""
|
||||
Write-Host "Operator follow-ups (NOT applied by this script):"
|
||||
Write-Host " * If you also want Blob Versioning + Soft-Delete (defence-in-depth against"
|
||||
Write-Host " operator mistakes), run separately and confirm the blast radius first:"
|
||||
Write-Host " az storage account blob-service-properties update \\"
|
||||
Write-Host " --account-name $StorageAccountName \\"
|
||||
Write-Host " --resource-group $ResourceGroupName \\"
|
||||
Write-Host " --enable-versioning true \\"
|
||||
Write-Host " --enable-delete-retention true \\"
|
||||
Write-Host " --delete-retention-days 14"
|
||||
Write-Host " Both settings affect ALL containers in the account."
|
||||
@@ -0,0 +1,135 @@
|
||||
using System;
|
||||
using AutopilotMonitor.Functions.DataAccess.TableStorage;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Pure Entity ↔ Domain roundtrip + unknown-string-rejection for
|
||||
/// <see cref="BackupJobsRepository"/>. Enum refactorings on
|
||||
/// <see cref="BackupJobState"/> / <see cref="BackupJobKind"/> / <see cref="BackupOutcome"/>
|
||||
/// must NOT silently shift the persisted column — the mapper throws on unknown
|
||||
/// values so the operator sees a clear "schema drift" error instead of an
|
||||
/// unexplained state collapse.
|
||||
/// </summary>
|
||||
public class BackupJobsRepositoryMappingTests
|
||||
{
|
||||
[Fact]
|
||||
public void Roundtrip_Completed_PartialOutcome_AllFieldsPreserved()
|
||||
{
|
||||
var now = new DateTime(2026, 5, 22, 4, 5, 6, DateTimeKind.Utc);
|
||||
var dto = new BackupJobStatus
|
||||
{
|
||||
JobId = "abc123",
|
||||
Kind = BackupJobKind.Backup,
|
||||
State = BackupJobState.Completed,
|
||||
RequestedBy = "alice@contoso.test",
|
||||
QueuedAtUtc = now,
|
||||
StartedAtUtc = now.AddSeconds(1),
|
||||
CompletedAtUtc = now.AddMinutes(3),
|
||||
LastHeartbeatUtc = now.AddMinutes(3),
|
||||
BackupId = "20260522T040000Z_aa",
|
||||
BackupOutcome = BackupOutcome.Partial,
|
||||
};
|
||||
|
||||
var entity = BackupJobsRepository.MapToEntity(dto);
|
||||
var roundtripped = BackupJobsRepository.MapToDomain(entity);
|
||||
|
||||
Assert.Equal(dto.JobId, roundtripped.JobId);
|
||||
Assert.Equal(BackupJobKind.Backup, roundtripped.Kind);
|
||||
Assert.Equal(BackupJobState.Completed, roundtripped.State);
|
||||
Assert.Equal(dto.RequestedBy, roundtripped.RequestedBy);
|
||||
Assert.Equal(dto.QueuedAtUtc, roundtripped.QueuedAtUtc);
|
||||
Assert.Equal(dto.StartedAtUtc, roundtripped.StartedAtUtc);
|
||||
Assert.Equal(dto.CompletedAtUtc, roundtripped.CompletedAtUtc);
|
||||
Assert.Equal(dto.LastHeartbeatUtc, roundtripped.LastHeartbeatUtc);
|
||||
Assert.Equal(dto.BackupId, roundtripped.BackupId);
|
||||
Assert.Equal(BackupOutcome.Partial, roundtripped.BackupOutcome);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Roundtrip_RestoreTable_FieldsPreserved()
|
||||
{
|
||||
var dto = new BackupJobStatus
|
||||
{
|
||||
JobId = "restore01",
|
||||
Kind = BackupJobKind.RestoreTable,
|
||||
State = BackupJobState.Running,
|
||||
RequestedBy = "bob@contoso.test",
|
||||
QueuedAtUtc = DateTime.UtcNow,
|
||||
LastHeartbeatUtc = DateTime.UtcNow,
|
||||
SourceBackupId = "20260522T040000Z_aa",
|
||||
TableName = "AnalyzeRules",
|
||||
Strategy = "upsert-only",
|
||||
Progress = "{\"phase\":\"Upsert\"}",
|
||||
};
|
||||
|
||||
var roundtripped = BackupJobsRepository.MapToDomain(BackupJobsRepository.MapToEntity(dto));
|
||||
|
||||
Assert.Equal(BackupJobKind.RestoreTable, roundtripped.Kind);
|
||||
Assert.Equal(BackupJobState.Running, roundtripped.State);
|
||||
Assert.Equal(dto.SourceBackupId, roundtripped.SourceBackupId);
|
||||
Assert.Equal(dto.TableName, roundtripped.TableName);
|
||||
Assert.Equal(dto.Strategy, roundtripped.Strategy);
|
||||
Assert.Equal(dto.Progress, roundtripped.Progress);
|
||||
Assert.Null(roundtripped.BackupOutcome);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Map_ThrowsOnUnknownStateString()
|
||||
{
|
||||
var entity = new BackupJobStatusEntity
|
||||
{
|
||||
RowKey = "x",
|
||||
Kind = "Backup",
|
||||
State = "FuturisticGhostState", // unknown — refactor / hand-edit
|
||||
RequestedBy = "y",
|
||||
};
|
||||
Assert.Throws<InvalidOperationException>(() => BackupJobsRepository.MapToDomain(entity));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Map_ThrowsOnUnknownKindString()
|
||||
{
|
||||
var entity = new BackupJobStatusEntity
|
||||
{
|
||||
RowKey = "x",
|
||||
Kind = "RestoreShard", // unknown
|
||||
State = "Queued",
|
||||
RequestedBy = "y",
|
||||
};
|
||||
Assert.Throws<InvalidOperationException>(() => BackupJobsRepository.MapToDomain(entity));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Map_ThrowsOnLegacyFailedBackupOutcome()
|
||||
{
|
||||
// Pre-Wave21 plans allowed BackupOutcome=Failed; the current schema restricts
|
||||
// BackupOutcome to {Success, Partial} (real failures live on JobState=Failed
|
||||
// with BackupOutcome=null). A legacy "Failed" row must surface loudly.
|
||||
var entity = new BackupJobStatusEntity
|
||||
{
|
||||
RowKey = "x",
|
||||
Kind = "Backup",
|
||||
State = "Completed",
|
||||
RequestedBy = "y",
|
||||
BackupOutcome = "Failed", // legacy
|
||||
};
|
||||
Assert.Throws<InvalidOperationException>(() => BackupJobsRepository.MapToDomain(entity));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Map_BackupOutcomeNull_RoundtripsAsNull()
|
||||
{
|
||||
var entity = new BackupJobStatusEntity
|
||||
{
|
||||
RowKey = "x",
|
||||
Kind = "Backup",
|
||||
State = "Failed",
|
||||
RequestedBy = "y",
|
||||
BackupOutcome = null,
|
||||
};
|
||||
var dto = BackupJobsRepository.MapToDomain(entity);
|
||||
Assert.Null(dto.BackupOutcome);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Text.Json;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
using AutopilotMonitor.Shared.Models.Deletion;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Locks the JSON shape produced by <see cref="BackupManifestJson.SerializerOptions"/>
|
||||
/// so the manifest blob format stays implementation-stable across enum refactors.
|
||||
/// </summary>
|
||||
public class BackupManifestJsonTests
|
||||
{
|
||||
[Fact]
|
||||
public void Enums_Serialize_AsStrings_NotNumbers()
|
||||
{
|
||||
var manifest = new CriticalTableBackupManifest
|
||||
{
|
||||
BackupId = "test",
|
||||
Outcome = BackupOutcome.Partial,
|
||||
Tables = { new CriticalTableBackupTableEntry { TableName = "X", Status = TableBackupStatus.Skipped } },
|
||||
};
|
||||
|
||||
var json = JsonSerializer.Serialize(manifest, BackupManifestJson.SerializerOptions);
|
||||
|
||||
Assert.Contains("\"outcome\":\"Partial\"", json);
|
||||
Assert.Contains("\"status\":\"Skipped\"", json);
|
||||
Assert.DoesNotContain("\"outcome\":1", json);
|
||||
Assert.DoesNotContain("\"status\":2", json);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Deserialize_RejectsNumericEnumValues()
|
||||
{
|
||||
var jsonWithNumber = "{\"backupId\":\"x\",\"outcome\":1,\"tables\":[]}";
|
||||
Assert.Throws<JsonException>(() =>
|
||||
JsonSerializer.Deserialize<CriticalTableBackupManifest>(jsonWithNumber, BackupManifestJson.SerializerOptions));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DictionaryKeyPolicy_Unset_PreservesAzureTableColumnNames()
|
||||
{
|
||||
// DeletionRowDump.Props uses Azure Table column names as dictionary keys; camelcasing them
|
||||
// would manufacture new columns on restore. The serializer-options block in
|
||||
// BackupManifestJson deliberately leaves DictionaryKeyPolicy unset for this reason.
|
||||
var dump = new DeletionRowDump
|
||||
{
|
||||
Pk = "p", Rk = "r",
|
||||
Props = new Dictionary<string, DeletionPropValue>
|
||||
{
|
||||
["IsEnabled"] = new() { EdmType = DeletionPropEdmType.Boolean, Value = ParseJson("true") },
|
||||
["EventType"] = new() { EdmType = DeletionPropEdmType.String, Value = ParseJson("\"x\"") },
|
||||
},
|
||||
};
|
||||
|
||||
var json = JsonSerializer.Serialize(dump, BackupManifestJson.SerializerOptions);
|
||||
|
||||
Assert.Contains("\"IsEnabled\"", json);
|
||||
Assert.Contains("\"EventType\"", json);
|
||||
// Negative — camelcased variants would silently break restore against the live table.
|
||||
Assert.DoesNotContain("\"isEnabled\":", json);
|
||||
Assert.DoesNotContain("\"eventType\":", json);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Roundtrip_ManifestFieldsPreserved()
|
||||
{
|
||||
var startedAt = new DateTime(2026, 5, 22, 4, 0, 0, DateTimeKind.Utc);
|
||||
var manifest = new CriticalTableBackupManifest
|
||||
{
|
||||
SchemaVersion = 1,
|
||||
BackupId = "20260522T040000Z_abcdef01",
|
||||
StartedAtUtc = startedAt,
|
||||
CompletedAtUtc = startedAt.AddMinutes(5),
|
||||
TriggeredBy = "alice@contoso.test",
|
||||
Outcome = BackupOutcome.Success,
|
||||
Tables =
|
||||
{
|
||||
new CriticalTableBackupTableEntry
|
||||
{
|
||||
TableName = "AdminConfiguration",
|
||||
Status = TableBackupStatus.Ok,
|
||||
RowCount = 12,
|
||||
ByteSize = 4096,
|
||||
Sha256Hex = "ABCDEF",
|
||||
BlobName = "20260522T040000Z_abcdef01/AdminConfiguration.ndjson",
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
var json = JsonSerializer.Serialize(manifest, BackupManifestJson.SerializerOptions);
|
||||
var roundtripped = JsonSerializer.Deserialize<CriticalTableBackupManifest>(json, BackupManifestJson.SerializerOptions);
|
||||
|
||||
Assert.NotNull(roundtripped);
|
||||
Assert.Equal(manifest.BackupId, roundtripped!.BackupId);
|
||||
Assert.Equal(manifest.TriggeredBy, roundtripped.TriggeredBy);
|
||||
Assert.Equal(BackupOutcome.Success, roundtripped.Outcome);
|
||||
Assert.Single(roundtripped.Tables);
|
||||
Assert.Equal("AdminConfiguration", roundtripped.Tables[0].TableName);
|
||||
Assert.Equal(TableBackupStatus.Ok, roundtripped.Tables[0].Status);
|
||||
Assert.Equal(12, roundtripped.Tables[0].RowCount);
|
||||
}
|
||||
|
||||
private static JsonElement ParseJson(string json)
|
||||
{
|
||||
using var doc = JsonDocument.Parse(json);
|
||||
return doc.RootElement.Clone();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,237 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Azure;
|
||||
using Azure.Data.Tables;
|
||||
using AutopilotMonitor.Functions.Services;
|
||||
using AutopilotMonitor.Functions.Services.Backup;
|
||||
using AutopilotMonitor.Shared;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Service-level happy/partial/empty/budget paths for the critical-table backup feature.
|
||||
/// Uses in-memory fakes for table enumeration and blob I/O so the loop's structural
|
||||
/// invariants (manifest written LAST, per-table failure → Partial, budget → Skipped)
|
||||
/// can be locked without touching Azurite.
|
||||
/// </summary>
|
||||
public class CriticalTableBackupServiceTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task Happy_AllTables_OutcomeSuccess_ManifestWritten()
|
||||
{
|
||||
var tables = new FakeTableStorage(new Dictionary<string, List<TableEntity>>
|
||||
{
|
||||
// Use real critical-table names so the service finds them in CriticalBackupTables.All.
|
||||
[Constants.TableNames.AdminConfiguration] = new()
|
||||
{
|
||||
new TableEntity("admin", "row1") { ["Setting"] = "value" },
|
||||
},
|
||||
[Constants.TableNames.AnalyzeRules] = new()
|
||||
{
|
||||
new TableEntity("tenant1", "rule1") { ["Enabled"] = true },
|
||||
},
|
||||
}, defaultEmpty: true);
|
||||
var store = new FakeBlobStore();
|
||||
var clock = new FakeClock(new DateTime(2026, 5, 22, 4, 0, 0, DateTimeKind.Utc));
|
||||
var svc = new CriticalTableBackupService(tables, store, NullLogger<CriticalTableBackupService>.Instance, clock);
|
||||
|
||||
var backupId = svc.GenerateBackupId();
|
||||
var result = await svc.RunBackupUnderLeaseAsync(backupId, "Timer", CancellationToken.None);
|
||||
|
||||
Assert.Equal(BackupOutcome.Success, result.Outcome);
|
||||
Assert.True(store.ManifestWritten, "manifest must be written LAST");
|
||||
Assert.Equal(15, result.Manifest.Tables.Count); // critical-tables catalog count
|
||||
Assert.All(result.Manifest.Tables, t =>
|
||||
Assert.True(t.Status == TableBackupStatus.Ok || t.Status == TableBackupStatus.Empty));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Partial_MiddleTableThrows_OutcomePartial_ManifestStillWritten()
|
||||
{
|
||||
var tables = new FakeTableStorage(new Dictionary<string, List<TableEntity>>
|
||||
{
|
||||
[Constants.TableNames.AdminConfiguration] = new() { new TableEntity("a", "b") { ["X"] = 1 } },
|
||||
}, defaultEmpty: true, throwOnTable: Constants.TableNames.AnalyzeRules);
|
||||
var store = new FakeBlobStore();
|
||||
var clock = new FakeClock(new DateTime(2026, 5, 22, 4, 0, 0, DateTimeKind.Utc));
|
||||
var svc = new CriticalTableBackupService(tables, store, NullLogger<CriticalTableBackupService>.Instance, clock);
|
||||
|
||||
var result = await svc.RunBackupUnderLeaseAsync(svc.GenerateBackupId(), "Timer", CancellationToken.None);
|
||||
|
||||
Assert.Equal(BackupOutcome.Partial, result.Outcome);
|
||||
Assert.True(store.ManifestWritten);
|
||||
var failedEntry = result.Manifest.Tables.First(t => t.TableName == Constants.TableNames.AnalyzeRules);
|
||||
Assert.Equal(TableBackupStatus.Failed, failedEntry.Status);
|
||||
Assert.NotNull(failedEntry.ErrorMessage);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Empty_AllTablesEmpty_OutcomeSuccess_EveryEntryEmptyStatus()
|
||||
{
|
||||
var tables = new FakeTableStorage(new(), defaultEmpty: true);
|
||||
var store = new FakeBlobStore();
|
||||
var clock = new FakeClock(new DateTime(2026, 5, 22, 4, 0, 0, DateTimeKind.Utc));
|
||||
var svc = new CriticalTableBackupService(tables, store, NullLogger<CriticalTableBackupService>.Instance, clock);
|
||||
|
||||
var result = await svc.RunBackupUnderLeaseAsync(svc.GenerateBackupId(), "Timer", CancellationToken.None);
|
||||
|
||||
Assert.Equal(BackupOutcome.Success, result.Outcome);
|
||||
Assert.All(result.Manifest.Tables, t => Assert.Equal(TableBackupStatus.Empty, t.Status));
|
||||
Assert.All(result.Manifest.Tables, t => Assert.Equal(0, t.RowCount));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task BudgetExceeded_PreLoop_RemainingTablesSkipped_OutcomePartial()
|
||||
{
|
||||
// Clock jumps forward by 51 min after the very first GetUtcNow call so the
|
||||
// per-run budget (50 min) is exceeded before any subsequent table iteration.
|
||||
var startUtc = new DateTime(2026, 5, 22, 4, 0, 0, DateTimeKind.Utc);
|
||||
var clock = new FakeClock(startUtc);
|
||||
var tables = new FakeTableStorage(new(), defaultEmpty: true);
|
||||
var store = new FakeBlobStore();
|
||||
|
||||
// First clock read = startedAtUtc inside RunBackupUnderLeaseAsync; after that we
|
||||
// jump 51 min so the very first per-table budget check trips. Use a hand-crafted
|
||||
// backupId so GenerateBackupId() doesn't pre-consume a clock read.
|
||||
clock.AdvanceOnNextRead = TimeSpan.FromMinutes(51);
|
||||
clock.AdvanceAfterReadNumber = 1;
|
||||
|
||||
var svc = new CriticalTableBackupService(tables, store, NullLogger<CriticalTableBackupService>.Instance, clock);
|
||||
var result = await svc.RunBackupUnderLeaseAsync("20260522T040000Z_deadbeef", "Timer", CancellationToken.None);
|
||||
|
||||
Assert.Equal(BackupOutcome.Partial, result.Outcome);
|
||||
Assert.True(store.ManifestWritten);
|
||||
Assert.All(result.Manifest.Tables, t => Assert.Equal(TableBackupStatus.Skipped, t.Status));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void GenerateBackupId_FormatIsStable()
|
||||
{
|
||||
var clock = new FakeClock(new DateTime(2026, 5, 22, 4, 0, 0, DateTimeKind.Utc));
|
||||
var tables = new FakeTableStorage(new(), defaultEmpty: true);
|
||||
var store = new FakeBlobStore();
|
||||
var svc = new CriticalTableBackupService(tables, store, NullLogger<CriticalTableBackupService>.Instance, clock);
|
||||
|
||||
var id = svc.GenerateBackupId();
|
||||
Assert.Matches(@"^\d{8}T\d{6}Z_[0-9a-f]{8}$", id);
|
||||
Assert.StartsWith("20260522T040000Z_", id);
|
||||
}
|
||||
|
||||
// ── Fakes ───────────────────────────────────────────────────────────────
|
||||
|
||||
private sealed class FakeTableStorage : TableStorageService
|
||||
{
|
||||
private readonly Dictionary<string, List<TableEntity>> _data;
|
||||
private readonly bool _defaultEmpty;
|
||||
private readonly string? _throwOnTable;
|
||||
|
||||
public FakeTableStorage(Dictionary<string, List<TableEntity>> data, bool defaultEmpty = false, string? throwOnTable = null)
|
||||
: base(new TableServiceClient(new Uri("https://example.invalid"), new TableSharedKeyCredential("x", Convert.ToBase64String(new byte[32]))), NullLogger<TableStorageService>.Instance)
|
||||
{
|
||||
_data = data;
|
||||
_defaultEmpty = defaultEmpty;
|
||||
_throwOnTable = throwOnTable;
|
||||
}
|
||||
|
||||
public override async IAsyncEnumerable<TableEntity> EnumerateAllAsync(string tableName, [EnumeratorCancellation] CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (string.Equals(tableName, _throwOnTable, StringComparison.Ordinal))
|
||||
throw new InvalidOperationException($"simulated table failure: {tableName}");
|
||||
|
||||
if (_data.TryGetValue(tableName, out var rows))
|
||||
{
|
||||
foreach (var row in rows)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
yield return row;
|
||||
}
|
||||
yield break;
|
||||
}
|
||||
if (_defaultEmpty) yield break;
|
||||
throw new InvalidOperationException($"unexpected table: {tableName}");
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class FakeBlobStore : BlobBackupStore
|
||||
{
|
||||
public bool ManifestWritten { get; private set; }
|
||||
public readonly List<string> WrittenNdjsonBlobs = new();
|
||||
public readonly Dictionary<string, byte[]> NdjsonBytes = new();
|
||||
|
||||
public FakeBlobStore()
|
||||
: base(new FakeBlobStorageService(), NullLogger<BlobBackupStore>.Instance)
|
||||
{
|
||||
}
|
||||
|
||||
public override Task EnsureContainerAsync(CancellationToken ct = default) => Task.CompletedTask;
|
||||
public override Task EnsureMaintenanceLockSentinelAsync(CancellationToken ct = default) => Task.CompletedTask;
|
||||
|
||||
public override Task<Stream> OpenNdjsonWriteStreamAsync(string backupId, string tableName, CancellationToken ct = default)
|
||||
{
|
||||
var name = BuildNdjsonBlobName(backupId, tableName);
|
||||
WrittenNdjsonBlobs.Add(name);
|
||||
var ms = new RecordingMemoryStream(b => NdjsonBytes[name] = b);
|
||||
return Task.FromResult<Stream>(ms);
|
||||
}
|
||||
|
||||
public override Task<bool> TryWriteManifestAsync(string backupId, byte[] manifestBytes, CancellationToken ct = default)
|
||||
{
|
||||
// Plan invariant: every NDJSON blob must be committed (= disposed) before the manifest is written.
|
||||
// Our recording stream finalises on dispose; if a stream is still open at this point a bug exists.
|
||||
ManifestWritten = true;
|
||||
return Task.FromResult(true);
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class RecordingMemoryStream : MemoryStream
|
||||
{
|
||||
private readonly Action<byte[]> _onClose;
|
||||
private bool _closed;
|
||||
public RecordingMemoryStream(Action<byte[]> onClose) { _onClose = onClose; }
|
||||
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
if (!_closed)
|
||||
{
|
||||
_closed = true;
|
||||
_onClose(this.ToArray());
|
||||
}
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Minimal stub — never actually used because all BlobBackupStore methods are overridden.</summary>
|
||||
private sealed class FakeBlobStorageService : BlobStorageService
|
||||
{
|
||||
public FakeBlobStorageService()
|
||||
: base(new Azure.Storage.Blobs.BlobServiceClient(new Uri("https://example.invalid"), new Azure.Storage.StorageSharedKeyCredential("x", Convert.ToBase64String(new byte[32]))), NullLogger<BlobStorageService>.Instance) { }
|
||||
}
|
||||
|
||||
private sealed class FakeClock : TimeProvider
|
||||
{
|
||||
private DateTime _now;
|
||||
private int _readCount;
|
||||
/// <summary>Time to advance AFTER the Nth read (default: after the 1st).</summary>
|
||||
public TimeSpan AdvanceOnNextRead { get; set; } = TimeSpan.Zero;
|
||||
public int AdvanceAfterReadNumber { get; set; } = 1;
|
||||
public FakeClock(DateTime startUtc) { _now = DateTime.SpecifyKind(startUtc, DateTimeKind.Utc); }
|
||||
public override DateTimeOffset GetUtcNow()
|
||||
{
|
||||
var currentValue = new DateTimeOffset(_now, TimeSpan.Zero);
|
||||
_readCount++;
|
||||
if (_readCount == AdvanceAfterReadNumber && AdvanceOnNextRead != TimeSpan.Zero)
|
||||
{
|
||||
_now += AdvanceOnNextRead;
|
||||
AdvanceOnNextRead = TimeSpan.Zero;
|
||||
}
|
||||
return currentValue;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -122,8 +122,8 @@ public class MaintenanceServicePoisonQueueTests
|
||||
public void MonitoredPoisonQueues_CoversAllProducerQueues()
|
||||
{
|
||||
// Guard against silently adding a new producer queue without a poison entry.
|
||||
// If a fourth queue is introduced, MonitoredPoisonQueues + this assertion must move together.
|
||||
Assert.Equal(3, MaintenanceService.MonitoredPoisonQueues.Length);
|
||||
// If a fifth queue is introduced, MonitoredPoisonQueues + this assertion must move together.
|
||||
Assert.Equal(4, MaintenanceService.MonitoredPoisonQueues.Length);
|
||||
Assert.Contains(
|
||||
Constants.QueueNames.AnalyzeOnEnrollmentEnd + "-poison",
|
||||
MaintenanceService.MonitoredPoisonQueues);
|
||||
@@ -133,6 +133,10 @@ public class MaintenanceServicePoisonQueueTests
|
||||
Assert.Contains(
|
||||
Constants.QueueNames.TelemetryIndexReconcile + "-poison",
|
||||
MaintenanceService.MonitoredPoisonQueues);
|
||||
// PR1 critical-table backup queue (plan §Wave5 #2).
|
||||
Assert.Contains(
|
||||
Constants.QueueNames.CriticalTableBackupPoison,
|
||||
MaintenanceService.MonitoredPoisonQueues);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
|
||||
@@ -208,8 +208,10 @@ Generated_Code/
|
||||
# Backup & report files from converting an old project file
|
||||
# to a newer Visual Studio version. Backup files are not needed,
|
||||
# because we have git ;-)
|
||||
# Pinned to top-level so Services/Backup, Functions/Backup, etc. for the
|
||||
# critical-table backup feature (plan §PR1) are not accidentally excluded.
|
||||
_UpgradeReport_Files/
|
||||
Backup*/
|
||||
/Backup*/
|
||||
UpgradeLog*.XML
|
||||
UpgradeLog*.htm
|
||||
|
||||
|
||||
+63
@@ -0,0 +1,63 @@
|
||||
using System;
|
||||
using Azure;
|
||||
using Azure.Data.Tables;
|
||||
|
||||
namespace AutopilotMonitor.Functions.DataAccess.TableStorage
|
||||
{
|
||||
/// <summary>
|
||||
/// Azure Table Storage row for the <c>BackupJobs</c> table. Kind / State / BackupOutcome
|
||||
/// are persisted as strings because Azure Tables has no EDM enum type; the
|
||||
/// <see cref="BackupJobsRepository"/> maps to and from the strongly-typed
|
||||
/// <c>BackupJobStatus</c> DTO in <c>AutopilotMonitor.Shared.Models.Backup</c>.
|
||||
/// Unknown strings on read throw so enum refactorings cannot silently shift
|
||||
/// the persisted state.
|
||||
/// <para>
|
||||
/// Lives in the Functions project (not Shared) because <c>ITableEntity</c> is an
|
||||
/// Azure-SDK dependency we deliberately keep out of the Shared layer.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Single-partition design (<c>PartitionKey="BackupJobs"</c>): job IDs are guids,
|
||||
/// the table never grows large enough to need partitioning, and a single PK lets
|
||||
/// the watchdog enumerate the whole table cheaply.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class BackupJobStatusEntity : ITableEntity
|
||||
{
|
||||
/// <summary>Always <c>"BackupJobs"</c>.</summary>
|
||||
public string PartitionKey { get; set; } = "BackupJobs";
|
||||
|
||||
/// <summary>Job id, Guid-N format.</summary>
|
||||
public string RowKey { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Filled by the SDK on read; ignored on write.</summary>
|
||||
public DateTimeOffset? Timestamp { get; set; }
|
||||
|
||||
/// <summary>Filled by the SDK on read; used by callers for CAS updates.</summary>
|
||||
public ETag ETag { get; set; }
|
||||
|
||||
/// <summary>"Backup" or "RestoreTable".</summary>
|
||||
public string Kind { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>"Queued" | "Running" | "Completed" | "Failed" | "Skipped" | "BlockedTerminal".</summary>
|
||||
public string State { get; set; } = string.Empty;
|
||||
|
||||
public string RequestedBy { get; set; } = string.Empty;
|
||||
|
||||
public DateTime QueuedAtUtc { get; set; }
|
||||
public DateTime? StartedAtUtc { get; set; }
|
||||
public DateTime? CompletedAtUtc { get; set; }
|
||||
|
||||
/// <summary>Watchdog stale-detection. Refreshed every renewal-tick and per-phase progress write.</summary>
|
||||
public DateTime LastHeartbeatUtc { get; set; }
|
||||
|
||||
public string? BackupId { get; set; }
|
||||
public string? SourceBackupId { get; set; }
|
||||
public string? TableName { get; set; }
|
||||
public string? Strategy { get; set; }
|
||||
public string? Progress { get; set; }
|
||||
public string? Error { get; set; }
|
||||
|
||||
/// <summary>"Success" or "Partial" (only when Kind=Backup AND State=Completed). Null otherwise.</summary>
|
||||
public string? BackupOutcome { get; set; }
|
||||
}
|
||||
}
|
||||
+189
@@ -0,0 +1,189 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Functions.Services;
|
||||
using AutopilotMonitor.Shared;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
using Azure;
|
||||
using Azure.Data.Tables;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.DataAccess.TableStorage
|
||||
{
|
||||
/// <summary>
|
||||
/// Per-job state machine persistence for the critical-table backup feature.
|
||||
/// Single partition (<c>"BackupJobs"</c>), RowKey = jobId (Guid-N). Lifecycle is
|
||||
/// driven by ETag-CAS updates so the worker, the watchdog, and the duplicate-detection
|
||||
/// path cannot race each other into an inconsistent state.
|
||||
/// <para>
|
||||
/// Enum-to-string mapping for Kind / State / BackupOutcome lives here in the
|
||||
/// repository because Azure Tables has no EDM enum and enum refactorings must not
|
||||
/// silently shift the persisted value. Unknown strings on read throw with a clear
|
||||
/// message — no silent fallback.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public class BackupJobsRepository
|
||||
{
|
||||
private const string PartitionKey = "BackupJobs";
|
||||
|
||||
private readonly TableClient _table;
|
||||
private readonly ILogger<BackupJobsRepository> _logger;
|
||||
|
||||
public BackupJobsRepository(TableStorageService storage, ILogger<BackupJobsRepository> logger)
|
||||
{
|
||||
_logger = logger;
|
||||
_table = storage.GetTableClient(Constants.TableNames.BackupJobs);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Initial-insert of a freshly-created job. Returns false on 409 — Guid-N jobIds
|
||||
/// should never collide, so a 409 is an operator-visible bug.
|
||||
/// </summary>
|
||||
public virtual async Task<bool> CreateAsync(BackupJobStatus job, CancellationToken ct = default)
|
||||
{
|
||||
if (job == null) throw new ArgumentNullException(nameof(job));
|
||||
if (string.IsNullOrEmpty(job.JobId)) throw new ArgumentException("JobId required", nameof(job));
|
||||
|
||||
try
|
||||
{
|
||||
await _table.AddEntityAsync(MapToEntity(job), ct).ConfigureAwait(false);
|
||||
return true;
|
||||
}
|
||||
catch (RequestFailedException ex) when (ex.Status == 409)
|
||||
{
|
||||
_logger.LogError(
|
||||
"BackupJobsRepository.CreateAsync: jobId={JobId} already exists — possible Guid collision",
|
||||
job.JobId);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Returns (null, null) if the job does not exist.</summary>
|
||||
public virtual async Task<(BackupJobStatus? Job, ETag? ETag)> GetWithETagAsync(string jobId, CancellationToken ct = default)
|
||||
{
|
||||
if (string.IsNullOrEmpty(jobId)) throw new ArgumentException("jobId required", nameof(jobId));
|
||||
|
||||
try
|
||||
{
|
||||
var response = await _table.GetEntityAsync<BackupJobStatusEntity>(PartitionKey, jobId, cancellationToken: ct).ConfigureAwait(false);
|
||||
return (MapToDomain(response.Value), response.Value.ETag);
|
||||
}
|
||||
catch (RequestFailedException ex) when (ex.Status == 404)
|
||||
{
|
||||
return (null, null);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// CAS-update with If-Match. Returns true on success, false on 412 (concurrent
|
||||
/// modification) or 404 (job vanished). Any other failure propagates.
|
||||
/// </summary>
|
||||
public virtual async Task<bool> TryUpdateWithCasAsync(BackupJobStatus job, ETag ifMatch, CancellationToken ct = default)
|
||||
{
|
||||
if (job == null) throw new ArgumentNullException(nameof(job));
|
||||
if (string.IsNullOrEmpty(job.JobId)) throw new ArgumentException("JobId required", nameof(job));
|
||||
|
||||
try
|
||||
{
|
||||
await _table.UpdateEntityAsync(MapToEntity(job), ifMatch, TableUpdateMode.Replace, ct).ConfigureAwait(false);
|
||||
return true;
|
||||
}
|
||||
catch (RequestFailedException ex) when (ex.Status == 412 || ex.Status == 404)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Streams every job, optionally filtered server-side. Used by the watchdog
|
||||
/// (state Queued OR Running) and the listing API. ETag is per-row from the SDK.
|
||||
/// </summary>
|
||||
public virtual async IAsyncEnumerable<(BackupJobStatus Job, ETag ETag)> QueryAsync(
|
||||
string? filter = null,
|
||||
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default)
|
||||
{
|
||||
await foreach (var entity in _table.QueryAsync<BackupJobStatusEntity>(filter: filter, cancellationToken: ct).ConfigureAwait(false))
|
||||
{
|
||||
yield return (MapToDomain(entity), entity.ETag);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Mapping: Entity ↔ Domain ───────────────────────────────────────────
|
||||
|
||||
internal static BackupJobStatusEntity MapToEntity(BackupJobStatus job)
|
||||
{
|
||||
return new BackupJobStatusEntity
|
||||
{
|
||||
PartitionKey = PartitionKey,
|
||||
RowKey = job.JobId,
|
||||
Kind = job.Kind.ToString(),
|
||||
State = job.State.ToString(),
|
||||
RequestedBy = job.RequestedBy,
|
||||
QueuedAtUtc = job.QueuedAtUtc,
|
||||
StartedAtUtc = job.StartedAtUtc,
|
||||
CompletedAtUtc = job.CompletedAtUtc,
|
||||
LastHeartbeatUtc = job.LastHeartbeatUtc,
|
||||
BackupId = job.BackupId,
|
||||
SourceBackupId = job.SourceBackupId,
|
||||
TableName = job.TableName,
|
||||
Strategy = job.Strategy,
|
||||
Progress = job.Progress,
|
||||
Error = job.Error,
|
||||
BackupOutcome = job.BackupOutcome?.ToString(),
|
||||
};
|
||||
}
|
||||
|
||||
internal static BackupJobStatus MapToDomain(BackupJobStatusEntity entity)
|
||||
{
|
||||
return new BackupJobStatus
|
||||
{
|
||||
JobId = entity.RowKey,
|
||||
Kind = ParseKind(entity.Kind),
|
||||
State = ParseState(entity.State),
|
||||
RequestedBy = entity.RequestedBy,
|
||||
QueuedAtUtc = entity.QueuedAtUtc,
|
||||
StartedAtUtc = entity.StartedAtUtc,
|
||||
CompletedAtUtc = entity.CompletedAtUtc,
|
||||
LastHeartbeatUtc = entity.LastHeartbeatUtc,
|
||||
BackupId = entity.BackupId,
|
||||
SourceBackupId = entity.SourceBackupId,
|
||||
TableName = entity.TableName,
|
||||
Strategy = entity.Strategy,
|
||||
Progress = entity.Progress,
|
||||
Error = entity.Error,
|
||||
BackupOutcome = ParseBackupOutcome(entity.BackupOutcome),
|
||||
};
|
||||
}
|
||||
|
||||
private static BackupJobKind ParseKind(string value)
|
||||
{
|
||||
if (string.IsNullOrEmpty(value))
|
||||
throw new InvalidOperationException("BackupJobStatusEntity.Kind is empty — entity corrupt or partially written");
|
||||
if (!Enum.TryParse<BackupJobKind>(value, ignoreCase: false, out var kind))
|
||||
throw new InvalidOperationException($"BackupJobStatusEntity.Kind has unknown value '{value}' — enum refactoring or hand-edit detected");
|
||||
return kind;
|
||||
}
|
||||
|
||||
private static BackupJobState ParseState(string value)
|
||||
{
|
||||
if (string.IsNullOrEmpty(value))
|
||||
throw new InvalidOperationException("BackupJobStatusEntity.State is empty — entity corrupt or partially written");
|
||||
if (!Enum.TryParse<BackupJobState>(value, ignoreCase: false, out var state))
|
||||
throw new InvalidOperationException($"BackupJobStatusEntity.State has unknown value '{value}' — enum refactoring or hand-edit detected");
|
||||
return state;
|
||||
}
|
||||
|
||||
private static BackupOutcome? ParseBackupOutcome(string? value)
|
||||
{
|
||||
if (string.IsNullOrEmpty(value)) return null;
|
||||
// Legacy "Failed" outcomes from pre-Wave21 plans must reject (BackupOutcome is now
|
||||
// Success | Partial only; real failures live on JobState=Failed with BackupOutcome=null).
|
||||
if (!Enum.TryParse<BackupOutcome>(value, ignoreCase: false, out var outcome))
|
||||
throw new InvalidOperationException(
|
||||
$"BackupJobStatusEntity.BackupOutcome has unknown value '{value}' — must be 'Success' or 'Partial' (no 'Failed' in current schema)");
|
||||
return outcome;
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
+40
@@ -0,0 +1,40 @@
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Functions.Services.Backup;
|
||||
using Microsoft.Azure.Functions.Worker;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Functions.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// Drives <see cref="BackupJobWatchdog.SweepAsync"/> every 30 minutes so stuck
|
||||
/// Queued / Running rows in <c>BackupJobs</c> are eventually transitioned to
|
||||
/// Failed (with the watchdog's lease-probe + ETag-CAS safeguards). Independent
|
||||
/// timer because the cadence is shorter than the daily backup itself, and the
|
||||
/// watchdog is cheap (single filtered table query + 0..1 lease probe per sweep).
|
||||
/// </summary>
|
||||
public class BackupJobWatchdogTimerFunction
|
||||
{
|
||||
private readonly BackupJobWatchdog _watchdog;
|
||||
private readonly ILogger<BackupJobWatchdogTimerFunction> _logger;
|
||||
|
||||
public BackupJobWatchdogTimerFunction(
|
||||
BackupJobWatchdog watchdog,
|
||||
ILogger<BackupJobWatchdogTimerFunction> logger)
|
||||
{
|
||||
_watchdog = watchdog;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <summary>NCRONTAB "0 */30 * * * *" — every 30 minutes on the half-hour.</summary>
|
||||
[Function("BackupJobWatchdogTimer")]
|
||||
public async Task Run([TimerTrigger("0 */30 * * * *")] object timer, CancellationToken ct)
|
||||
{
|
||||
var transitioned = await _watchdog.SweepAsync(ct).ConfigureAwait(false);
|
||||
if (transitioned > 0)
|
||||
{
|
||||
_logger.LogWarning("BackupJobWatchdogTimer: {Count} stale job(s) transitioned to Failed", transitioned);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+118
@@ -0,0 +1,118 @@
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Functions.Services;
|
||||
using AutopilotMonitor.Functions.Services.Backup;
|
||||
using AutopilotMonitor.Shared;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
using Microsoft.Azure.Functions.Worker;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Functions.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// Daily 04:00 UTC trigger of the critical-table backup feature. The TIMER is
|
||||
/// worker-equivalent topology-wise (Lease → Service → Release) but has no
|
||||
/// JobStatus + no queue pop: failure paths emit OpsEvents only, and the
|
||||
/// daily re-run is the natural recovery mechanism (no per-fire retries).
|
||||
/// </summary>
|
||||
public class CriticalTableBackupTimerFunction
|
||||
{
|
||||
private readonly BlobBackupStore _blobStore;
|
||||
private readonly ICriticalTableBackupService _backupService;
|
||||
private readonly OpsEventService _opsEvents;
|
||||
private readonly ILogger<CriticalTableBackupTimerFunction> _logger;
|
||||
|
||||
public CriticalTableBackupTimerFunction(
|
||||
BlobBackupStore blobStore,
|
||||
ICriticalTableBackupService backupService,
|
||||
OpsEventService opsEvents,
|
||||
ILogger<CriticalTableBackupTimerFunction> logger)
|
||||
{
|
||||
_blobStore = blobStore;
|
||||
_backupService = backupService;
|
||||
_opsEvents = opsEvents;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// NCRONTAB "0 0 4 * * *" — every day at 04:00 UTC. After the daily
|
||||
/// VulnerabilityDataSync (03:00) and well outside the agent ingest peak.
|
||||
/// </summary>
|
||||
[Function("CriticalTableBackupTimer")]
|
||||
public async Task Run([TimerTrigger("0 0 4 * * *")] object timer, CancellationToken ct)
|
||||
{
|
||||
_logger.LogInformation("CriticalTableBackupTimer fired");
|
||||
|
||||
string triggeredBy = "Timer";
|
||||
string? backupId = null;
|
||||
|
||||
// ── Acquire maintenance lease ───────────────────────────────────────
|
||||
await _blobStore.EnsureMaintenanceLockSentinelAsync(ct).ConfigureAwait(false);
|
||||
|
||||
Azure.Storage.Blobs.Specialized.BlobLeaseClient? lease;
|
||||
try
|
||||
{
|
||||
lease = await _blobStore.AcquireMaintenanceLeaseAsync(BlobBackupStore.MaintenanceLeaseDuration, ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (LeaseHeldException)
|
||||
{
|
||||
_logger.LogInformation("CriticalTableBackupTimer: maintenance lease held by another op — skipping today's run");
|
||||
await TryRecordAsync(() => _opsEvents.RecordCriticalTableBackupSkippedLockedAsync(
|
||||
"another maintenance operation holds the lease at timer fire", triggeredBy)).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
backupId = _backupService.GenerateBackupId();
|
||||
BackupRunResult? result;
|
||||
try
|
||||
{
|
||||
result = await _backupService.RunBackupUnderLeaseAsync(backupId, triggeredBy, ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||||
{
|
||||
_logger.LogError(ex, "CriticalTableBackupTimer: backup run failed (backupId={BackupId})", backupId);
|
||||
await TryRecordAsync(() => _opsEvents.RecordCriticalTableBackupFailedAsync(backupId, ex.Message, triggeredBy)).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
var durationMs = (int)(result.Manifest.CompletedAtUtc - result.Manifest.StartedAtUtc).TotalMilliseconds;
|
||||
var failedOrSkipped = 0;
|
||||
foreach (var t in result.Manifest.Tables)
|
||||
{
|
||||
if (t.Status == TableBackupStatus.Failed || t.Status == TableBackupStatus.Skipped) failedOrSkipped++;
|
||||
}
|
||||
|
||||
if (result.Outcome == BackupOutcome.Success)
|
||||
{
|
||||
await TryRecordAsync(() => _opsEvents.RecordCriticalTableBackupCompletedAsync(
|
||||
result.Manifest.BackupId, result.Manifest.Tables.Count, durationMs,
|
||||
Constants.BlobContainers.CriticalTableBackups, result.ManifestBlobName, triggeredBy)).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
await TryRecordAsync(() => _opsEvents.RecordCriticalTableBackupPartialAsync(
|
||||
result.Manifest.BackupId, result.Manifest.Tables.Count, failedOrSkipped, durationMs,
|
||||
Constants.BlobContainers.CriticalTableBackups, result.ManifestBlobName, triggeredBy)).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
try { await lease.ReleaseAsync(cancellationToken: ct).ConfigureAwait(false); }
|
||||
catch (Exception ex) { _logger.LogWarning(ex, "CriticalTableBackupTimer: lease release failed (auto-expires)"); }
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>OpsEvent write must not bubble — backup itself already succeeded (or failed by other means).</summary>
|
||||
private async Task TryRecordAsync(Func<Task> writeAsync)
|
||||
{
|
||||
try { await writeAsync().ConfigureAwait(false); }
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "CriticalTableBackupTimer: ops-event recording failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
using System.Net;
|
||||
using System.Text.Json;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Functions.DataAccess.TableStorage;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
using Microsoft.Azure.Functions.Worker;
|
||||
using Microsoft.Azure.Functions.Worker.Http;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Functions.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// <c>GET /api/global/backups/jobs/{jobId}</c> — polling endpoint for the
|
||||
/// 202-Accepted async backup workflow. GA-only via <c>EndpointAccessPolicyCatalog</c>.
|
||||
/// Returns the live <see cref="BackupJobStatus"/> DTO including
|
||||
/// <c>backupOutcome</c> (Success / Partial) and <c>lastHeartbeatUtc</c> so the
|
||||
/// UI can render Completed / Partial / Failed banners and the operator can spot
|
||||
/// stalled jobs at a glance.
|
||||
/// </summary>
|
||||
public class GetBackupJobStatusFunction
|
||||
{
|
||||
private readonly BackupJobsRepository _jobs;
|
||||
private readonly ILogger<GetBackupJobStatusFunction> _logger;
|
||||
|
||||
public GetBackupJobStatusFunction(BackupJobsRepository jobs, ILogger<GetBackupJobStatusFunction> logger)
|
||||
{
|
||||
_jobs = jobs;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
[Function("GetBackupJobStatus")]
|
||||
public async Task<HttpResponseData> Run(
|
||||
[HttpTrigger(AuthorizationLevel.Anonymous, "get", Route = "global/backups/jobs/{jobId}")] HttpRequestData req,
|
||||
string jobId)
|
||||
{
|
||||
var (job, _) = await _jobs.GetWithETagAsync(jobId, req.FunctionContext.CancellationToken);
|
||||
if (job is null)
|
||||
{
|
||||
return await WriteJsonAsync(req, HttpStatusCode.NotFound, new { error = "JobNotFound", jobId });
|
||||
}
|
||||
return await WriteJsonAsync(req, HttpStatusCode.OK, job);
|
||||
}
|
||||
|
||||
private static async Task<HttpResponseData> WriteJsonAsync(HttpRequestData req, HttpStatusCode status, object body)
|
||||
{
|
||||
var response = req.CreateResponse(status);
|
||||
response.Headers.Add("Content-Type", "application/json; charset=utf-8");
|
||||
var json = JsonSerializer.Serialize(body, BackupManifestJson.SerializerOptions);
|
||||
await response.WriteStringAsync(json);
|
||||
return response;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
using System.Net;
|
||||
using System.Text.Json;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Functions.Services.Backup;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
using Microsoft.Azure.Functions.Worker;
|
||||
using Microsoft.Azure.Functions.Worker.Http;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Functions.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// <c>GET /api/global/backups/{backupId}</c> — fetches the manifest JSON for a
|
||||
/// specific backup run. 404 if the manifest is missing (i.e. the prefix exists but
|
||||
/// the durability anchor never got written — treated as <c>Incomplete</c> per plan).
|
||||
/// GA-only.
|
||||
/// </summary>
|
||||
public class GetBackupManifestFunction
|
||||
{
|
||||
private readonly BlobBackupStore _store;
|
||||
private readonly ILogger<GetBackupManifestFunction> _logger;
|
||||
|
||||
public GetBackupManifestFunction(BlobBackupStore store, ILogger<GetBackupManifestFunction> logger)
|
||||
{
|
||||
_store = store;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
[Function("GetBackupManifest")]
|
||||
public async Task<HttpResponseData> Run(
|
||||
[HttpTrigger(AuthorizationLevel.Anonymous, "get", Route = "global/backups/{backupId}")] HttpRequestData req,
|
||||
string backupId)
|
||||
{
|
||||
var (payload, _) = await _store.ReadManifestAsync(backupId, req.FunctionContext.CancellationToken);
|
||||
if (payload is null)
|
||||
{
|
||||
var notFound = req.CreateResponse(HttpStatusCode.NotFound);
|
||||
notFound.Headers.Add("Content-Type", "application/json; charset=utf-8");
|
||||
await notFound.WriteStringAsync(JsonSerializer.Serialize(new { error = "ManifestNotFound", backupId }, BackupManifestJson.SerializerOptions));
|
||||
return notFound;
|
||||
}
|
||||
|
||||
// Stream the bytes verbatim — the manifest is already JSON, no need to deserialise + re-serialise.
|
||||
var response = req.CreateResponse(HttpStatusCode.OK);
|
||||
response.Headers.Add("Content-Type", "application/json; charset=utf-8");
|
||||
await response.Body.WriteAsync(payload, req.FunctionContext.CancellationToken);
|
||||
return response;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Net;
|
||||
using System.Text.Json;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Functions.Services.Backup;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
using Microsoft.Azure.Functions.Worker;
|
||||
using Microsoft.Azure.Functions.Worker.Http;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Functions.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// <c>GET /api/global/backups</c> — lists all backupIds in the
|
||||
/// <c>critical-table-backups</c> container. PR1 returns just the ids; the manifest
|
||||
/// detail (table breakdown, sizes, SHAs) is fetched separately via
|
||||
/// <c>GET /api/global/backups/{backupId}</c>. GA-only.
|
||||
/// </summary>
|
||||
public class ListBackupsFunction
|
||||
{
|
||||
private readonly BlobBackupStore _store;
|
||||
private readonly ILogger<ListBackupsFunction> _logger;
|
||||
|
||||
public ListBackupsFunction(BlobBackupStore store, ILogger<ListBackupsFunction> logger)
|
||||
{
|
||||
_store = store;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
[Function("ListBackups")]
|
||||
public async Task<HttpResponseData> Run(
|
||||
[HttpTrigger(AuthorizationLevel.Anonymous, "get", Route = "global/backups")] HttpRequestData req)
|
||||
{
|
||||
var ids = new List<string>();
|
||||
await foreach (var id in _store.ListBackupIdsAsync(req.FunctionContext.CancellationToken))
|
||||
{
|
||||
ids.Add(id);
|
||||
}
|
||||
// newest first — ids are yyyyMMddTHHmmssZ_{guid8} so reverse-string-sort = reverse-time-sort.
|
||||
ids.Sort(StringComparer.Ordinal);
|
||||
ids.Reverse();
|
||||
|
||||
var response = req.CreateResponse(HttpStatusCode.OK);
|
||||
response.Headers.Add("Content-Type", "application/json; charset=utf-8");
|
||||
var json = JsonSerializer.Serialize(new { backupIds = ids }, BackupManifestJson.SerializerOptions);
|
||||
await response.WriteStringAsync(json);
|
||||
return response;
|
||||
}
|
||||
}
|
||||
}
|
||||
+118
@@ -0,0 +1,118 @@
|
||||
using System;
|
||||
using System.Net;
|
||||
using System.Text.Json;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Functions.DataAccess.TableStorage;
|
||||
using AutopilotMonitor.Functions.Helpers;
|
||||
using AutopilotMonitor.Functions.Services.Backup.Queue;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
using Microsoft.Azure.Functions.Worker;
|
||||
using Microsoft.Azure.Functions.Worker.Http;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Functions.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// <c>POST /api/global/backups/trigger</c> — manual GA trigger for a one-off
|
||||
/// critical-table backup run. GA-only via <c>EndpointAccessPolicyCatalog</c>.
|
||||
/// <para>
|
||||
/// Semantics (plan Wave11 #2 — fail-hard producer): on successful enqueue we
|
||||
/// return <c>202 Accepted</c> with the jobId; on enqueue failure we transition
|
||||
/// the BackupJobs row to <c>Failed</c> and return <c>500</c> so the operator
|
||||
/// sees an immediate error instead of an apparent success with no work scheduled.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public class TriggerCriticalTableBackupFunction
|
||||
{
|
||||
private readonly ICriticalTableBackupProducer _producer;
|
||||
private readonly BackupJobsRepository _jobs;
|
||||
private readonly ILogger<TriggerCriticalTableBackupFunction> _logger;
|
||||
private readonly TimeProvider _clock;
|
||||
|
||||
public TriggerCriticalTableBackupFunction(
|
||||
ICriticalTableBackupProducer producer,
|
||||
BackupJobsRepository jobs,
|
||||
ILogger<TriggerCriticalTableBackupFunction> logger,
|
||||
TimeProvider? clock = null)
|
||||
{
|
||||
_producer = producer;
|
||||
_jobs = jobs;
|
||||
_logger = logger;
|
||||
_clock = clock ?? TimeProvider.System;
|
||||
}
|
||||
|
||||
[Function("TriggerCriticalTableBackup")]
|
||||
public async Task<HttpResponseData> Run(
|
||||
[HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "global/backups/trigger")] HttpRequestData req)
|
||||
{
|
||||
var actor = TenantHelper.GetUserIdentifier(req) ?? "GlobalAdmin";
|
||||
var jobId = Guid.NewGuid().ToString("N");
|
||||
var nowUtc = _clock.GetUtcNow().UtcDateTime;
|
||||
|
||||
var job = new BackupJobStatus
|
||||
{
|
||||
JobId = jobId,
|
||||
Kind = BackupJobKind.Backup,
|
||||
State = BackupJobState.Queued,
|
||||
RequestedBy = actor,
|
||||
QueuedAtUtc = nowUtc,
|
||||
LastHeartbeatUtc = nowUtc,
|
||||
};
|
||||
|
||||
var inserted = await _jobs.CreateAsync(job, req.FunctionContext.CancellationToken);
|
||||
if (!inserted)
|
||||
{
|
||||
// Guid-N collision (effectively impossible) — surface so the operator can retry.
|
||||
return await WriteJsonAsync(req, HttpStatusCode.Conflict, new
|
||||
{
|
||||
error = "JobIdCollision",
|
||||
message = "Failed to allocate a unique jobId — please retry.",
|
||||
});
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await _producer.EnqueueAsync(new CriticalTableBackupEnvelope { JobId = jobId }, req.FunctionContext.CancellationToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "TriggerCriticalTableBackup: enqueue failed for jobId {JobId} — rolling status to Failed", jobId);
|
||||
|
||||
// Roll the row forward so the operator sees the failure immediately
|
||||
// (otherwise the BackupJobsWatchdog would eventually pick this up
|
||||
// after 60min — but for HTTP this is too slow).
|
||||
var (fresh, etag) = await _jobs.GetWithETagAsync(jobId, req.FunctionContext.CancellationToken);
|
||||
if (fresh is not null && etag is not null)
|
||||
{
|
||||
fresh.State = BackupJobState.Failed;
|
||||
fresh.CompletedAtUtc = _clock.GetUtcNow().UtcDateTime;
|
||||
fresh.LastHeartbeatUtc = fresh.CompletedAtUtc.Value;
|
||||
fresh.Error = $"enqueue failed: {ex.Message}";
|
||||
await _jobs.TryUpdateWithCasAsync(fresh, etag.Value, req.FunctionContext.CancellationToken);
|
||||
}
|
||||
|
||||
return await WriteJsonAsync(req, HttpStatusCode.InternalServerError, new
|
||||
{
|
||||
error = "EnqueueFailed",
|
||||
message = "Failed to enqueue backup job — job status has been marked Failed.",
|
||||
jobId,
|
||||
});
|
||||
}
|
||||
|
||||
return await WriteJsonAsync(req, HttpStatusCode.Accepted, new
|
||||
{
|
||||
jobId,
|
||||
statusUrl = $"/api/global/backups/jobs/{jobId}",
|
||||
});
|
||||
}
|
||||
|
||||
private static async Task<HttpResponseData> WriteJsonAsync(HttpRequestData req, HttpStatusCode status, object body)
|
||||
{
|
||||
var response = req.CreateResponse(status);
|
||||
response.Headers.Add("Content-Type", "application/json; charset=utf-8");
|
||||
var json = JsonSerializer.Serialize(body, BackupManifestJson.SerializerOptions);
|
||||
await response.WriteStringAsync(json);
|
||||
return response;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -41,6 +41,12 @@ public class NoStoreCacheMiddleware : IFunctionsWorkerMiddleware
|
||||
"/api/global/sessions",
|
||||
"/api/global/distress-reports",
|
||||
"/api/global/session-reports", // covers list; /download-url + /{id}/note via prefix
|
||||
// Critical-table backup feature — manifest + job-status responses can carry
|
||||
// property values from AdminConfiguration / TenantConfiguration (SAS URLs,
|
||||
// webhook secrets, API keys) and must never be cached. Trigger endpoint also
|
||||
// here so a returned jobId / statusUrl doesn't end up in proxy cache.
|
||||
"/api/global/backups",
|
||||
"/api/global/backups/trigger",
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
@@ -56,6 +62,7 @@ public class NoStoreCacheMiddleware : IFunctionsWorkerMiddleware
|
||||
"/api/global/raw/", // GA cross-tenant raw event/session/table surfaces
|
||||
"/api/global/search/", // GA cross-tenant search results
|
||||
"/api/global/session-reports/", // /download-url (SAS) + /{reportId}/note
|
||||
"/api/global/backups/", // covers /{backupId}, /jobs/{jobId} — same secret-bearing rationale
|
||||
};
|
||||
|
||||
public async Task Invoke(FunctionContext context, FunctionExecutionDelegate next)
|
||||
|
||||
@@ -277,6 +277,25 @@ builder.Services.AddSingleton<
|
||||
builder.Services.AddHostedService<
|
||||
AutopilotMonitor.Functions.Services.Vulnerability.VulnerabilityCorrelateQueueWorker>();
|
||||
|
||||
// ── Critical-Table Backup (plan §PR1) ─────────────────────────────────────────
|
||||
// Lease-aware service registered separately from the timer/worker so the lease
|
||||
// boundary is testable in isolation. JobsRepository is the canonical persistence
|
||||
// for the 6-state machine.
|
||||
builder.Services.AddSingleton<AutopilotMonitor.Functions.DataAccess.TableStorage.BackupJobsRepository>();
|
||||
builder.Services.AddSingleton<AutopilotMonitor.Functions.Services.Backup.BlobBackupStore>();
|
||||
builder.Services.AddSingleton<
|
||||
AutopilotMonitor.Functions.Services.Backup.ICriticalTableBackupService,
|
||||
AutopilotMonitor.Functions.Services.Backup.CriticalTableBackupService>();
|
||||
builder.Services.AddSingleton<AutopilotMonitor.Functions.Services.Backup.BackupJobWatchdog>();
|
||||
// Producer = fail-hard SendMessage (HTTP trigger handles the rollback to Failed).
|
||||
builder.Services.AddSingleton<
|
||||
AutopilotMonitor.Functions.Services.Backup.Queue.ICriticalTableBackupProducer,
|
||||
AutopilotMonitor.Functions.Services.Backup.Queue.AzureQueueCriticalTableBackupProducer>();
|
||||
// Worker = BackgroundService, not QueueTrigger (matches AnalyzeOnEnrollmentEndQueueWorker
|
||||
// pattern per plan §Wave4 #1 — avoids the QueueTrigger-specific connection app-setting).
|
||||
builder.Services.AddHostedService<
|
||||
AutopilotMonitor.Functions.Services.Backup.Queue.CriticalTableBackupQueueWorker>();
|
||||
|
||||
// Programmatic SignalR push for background tasks (rule engine, vulnerability correlation)
|
||||
builder.Services.AddSingleton<SignalRNotificationService>();
|
||||
builder.Services.AddSingleton<ISignalRNotificationService>(sp => sp.GetRequiredService<SignalRNotificationService>());
|
||||
|
||||
@@ -288,6 +288,12 @@ public static class EndpointAccessPolicyCatalog
|
||||
new("DELETE", "versions/block/{encodedPattern}", EndpointPolicy.GlobalAdminOnly),
|
||||
new("POST", "maintenance/trigger", EndpointPolicy.GlobalAdminOnly),
|
||||
|
||||
// ── Critical-Table Backup (plan §PR1) ──────────────────────────────────
|
||||
new("POST", "global/backups/trigger", EndpointPolicy.GlobalAdminOnly),
|
||||
new("GET", "global/backups", EndpointPolicy.GlobalAdminOnly),
|
||||
new("GET", "global/backups/{backupId}", EndpointPolicy.GlobalAdminOnly),
|
||||
new("GET", "global/backups/jobs/{jobId}", EndpointPolicy.GlobalAdminOnly),
|
||||
|
||||
// ── Tenant Offboarding Customs Archive (PR3.B) ─────────────────────────
|
||||
// Snapshot of each tenant's custom GatherRules / AnalyzeRules / ImeLogPatterns
|
||||
// rows from prior offboarding runs. Operator-driven cleanup: GA reviews + deletes.
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Functions.DataAccess.TableStorage;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// Periodic safety-net for stuck backup jobs. Two failure modes are healed:
|
||||
/// <list type="bullet">
|
||||
/// <item><b>Queued</b> longer than <see cref="QueuedStaleThreshold"/>: producer wrote
|
||||
/// the row but the queue never delivered (worker cold-start lag, storage outage).
|
||||
/// No heartbeat is expected on Queued, so the only signal is wall-clock age.</item>
|
||||
/// <item><b>Running</b> with stale <c>LastHeartbeatUtc</c> AND a free maintenance lease.
|
||||
/// A live worker holds the lease while running, so a free lease is the proof that
|
||||
/// the previous worker is gone. Without the lease-probe a watchdog flip would race
|
||||
/// a still-running worker into a Failed → Completed flap.</item>
|
||||
/// </list>
|
||||
/// All transitions are ETag-CAS; 412 is silently absorbed (worker beat us to the punch).
|
||||
/// </summary>
|
||||
public class BackupJobWatchdog
|
||||
{
|
||||
/// <summary>How long a Queued job is allowed to sit before the watchdog calls it dead.</summary>
|
||||
public static readonly TimeSpan QueuedStaleThreshold = TimeSpan.FromMinutes(60);
|
||||
|
||||
/// <summary>How long a Running job may go without a heartbeat before the watchdog inspects the lease.</summary>
|
||||
public static readonly TimeSpan RunningHeartbeatStaleThreshold = TimeSpan.FromMinutes(5);
|
||||
|
||||
private readonly BackupJobsRepository _jobs;
|
||||
private readonly BlobBackupStore _blobStore;
|
||||
private readonly ILogger<BackupJobWatchdog> _logger;
|
||||
private readonly TimeProvider _clock;
|
||||
|
||||
public BackupJobWatchdog(
|
||||
BackupJobsRepository jobs,
|
||||
BlobBackupStore blobStore,
|
||||
ILogger<BackupJobWatchdog> logger,
|
||||
TimeProvider? clock = null)
|
||||
{
|
||||
_jobs = jobs;
|
||||
_blobStore = blobStore;
|
||||
_logger = logger;
|
||||
_clock = clock ?? TimeProvider.System;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Single sweep — invoked by a timer (every ~30min) or on demand from a test.
|
||||
/// Returns the number of jobs transitioned to Failed.
|
||||
/// </summary>
|
||||
public async Task<int> SweepAsync(CancellationToken ct = default)
|
||||
{
|
||||
var nowUtc = _clock.GetUtcNow().UtcDateTime;
|
||||
var queuedCutoff = nowUtc - QueuedStaleThreshold;
|
||||
var runningCutoff = nowUtc - RunningHeartbeatStaleThreshold;
|
||||
var transitioned = 0;
|
||||
bool leaseProbedThisSweep = false;
|
||||
bool leaseHeldThisSweep = false;
|
||||
|
||||
// Query non-terminal jobs only — keeps the watchdog cheap and avoids
|
||||
// touching Completed/Failed rows whose lifecycle is done.
|
||||
var filter = $"PartitionKey eq 'BackupJobs' and (State eq 'Queued' or State eq 'Running')";
|
||||
|
||||
await foreach (var (job, etag) in _jobs.QueryAsync(filter, ct).ConfigureAwait(false))
|
||||
{
|
||||
ct.ThrowIfCancellationRequested();
|
||||
|
||||
bool shouldFail;
|
||||
string reason;
|
||||
|
||||
if (job.State == BackupJobState.Queued && job.QueuedAtUtc < queuedCutoff)
|
||||
{
|
||||
shouldFail = true;
|
||||
reason = $"watchdog: Queued > {QueuedStaleThreshold.TotalMinutes}min without worker pickup";
|
||||
}
|
||||
else if (job.State == BackupJobState.Running && job.LastHeartbeatUtc < runningCutoff)
|
||||
{
|
||||
// Probe the maintenance lease at most once per sweep — cheap, but each
|
||||
// probe is a real Azure call.
|
||||
if (!leaseProbedThisSweep)
|
||||
{
|
||||
leaseProbedThisSweep = true;
|
||||
try
|
||||
{
|
||||
var probe = await _blobStore.AcquireMaintenanceLeaseAsync(TimeSpan.FromSeconds(15), ct).ConfigureAwait(false);
|
||||
try { await probe.ReleaseAsync(cancellationToken: ct).ConfigureAwait(false); }
|
||||
catch (Exception ex) { _logger.LogWarning(ex, "BackupJobWatchdog: probe-lease release failed (auto-expires)"); }
|
||||
}
|
||||
catch (LeaseHeldException)
|
||||
{
|
||||
leaseHeldThisSweep = true;
|
||||
}
|
||||
}
|
||||
|
||||
if (leaseHeldThisSweep)
|
||||
{
|
||||
// Live worker still holds the lease — leave the job alone.
|
||||
continue;
|
||||
}
|
||||
|
||||
shouldFail = true;
|
||||
reason = $"watchdog: Running with stale heartbeat (> {RunningHeartbeatStaleThreshold.TotalMinutes}min) and free lease — previous worker likely died";
|
||||
}
|
||||
else
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!shouldFail) continue;
|
||||
|
||||
job.State = BackupJobState.Failed;
|
||||
job.CompletedAtUtc = nowUtc;
|
||||
job.LastHeartbeatUtc = nowUtc;
|
||||
job.Error = reason;
|
||||
var ok = await _jobs.TryUpdateWithCasAsync(job, etag, ct).ConfigureAwait(false);
|
||||
if (ok)
|
||||
{
|
||||
transitioned++;
|
||||
_logger.LogWarning("BackupJobWatchdog: transitioned job {JobId} to Failed — {Reason}", job.JobId, reason);
|
||||
}
|
||||
else
|
||||
{
|
||||
// 412 — a worker raced us with a legitimate state update. Drop and move on.
|
||||
_logger.LogInformation("BackupJobWatchdog: CAS 412 on jobId {JobId} — worker beat us, skipping", job.JobId);
|
||||
}
|
||||
}
|
||||
|
||||
return transitioned;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,219 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Shared;
|
||||
using Azure;
|
||||
using Azure.Storage.Blobs;
|
||||
using Azure.Storage.Blobs.Models;
|
||||
using Azure.Storage.Blobs.Specialized;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// Blob I/O for the critical-table backup feature (plan §PR1). Wraps the
|
||||
/// <c>critical-table-backups</c> container with a small surface tailored to the
|
||||
/// service + worker + timer flows: lease sentinel, per-table streaming uploads,
|
||||
/// manifest read/write, listing.
|
||||
/// <para>
|
||||
/// All write paths take the maintenance lease as a precondition — the WORKER /
|
||||
/// TIMER acquires + releases it; this store never owns lease lifecycle (plan
|
||||
/// §Wave17/18 Lease-Ownership-Boundary).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public class BlobBackupStore
|
||||
{
|
||||
/// <summary>Sentinel blob path — leased by the worker/timer to serialize maintenance ops.</summary>
|
||||
public const string MaintenanceLockBlobName = "_lock/maintenance.lock";
|
||||
|
||||
/// <summary>Default lease duration when the timer or worker takes the maintenance lock.</summary>
|
||||
public static readonly TimeSpan MaintenanceLeaseDuration = TimeSpan.FromSeconds(60);
|
||||
|
||||
private readonly BlobStorageService _blobs;
|
||||
private readonly ILogger<BlobBackupStore> _logger;
|
||||
private int _containerEnsured;
|
||||
|
||||
public BlobBackupStore(BlobStorageService blobs, ILogger<BlobBackupStore> logger)
|
||||
{
|
||||
_blobs = blobs;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
// ── Container + sentinel bootstrap ──────────────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// Idempotently ensures the container exists. Process-local once-flag avoids hammering
|
||||
/// CreateIfNotExists on every call.
|
||||
/// </summary>
|
||||
public virtual async Task EnsureContainerAsync(CancellationToken ct = default)
|
||||
{
|
||||
if (_containerEnsured == 1) return;
|
||||
var container = GetContainer();
|
||||
await container.CreateIfNotExistsAsync(cancellationToken: ct).ConfigureAwait(false);
|
||||
Interlocked.Exchange(ref _containerEnsured, 1);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Idempotently ensures the maintenance-lease sentinel blob exists. Without this an
|
||||
/// AcquireLease call on a missing blob throws 404. Uses If-None-Match=* so concurrent
|
||||
/// callers cooperate; both 409 BlobAlreadyExists and 412 ConditionNotMet are no-ops.
|
||||
/// </summary>
|
||||
public virtual async Task EnsureMaintenanceLockSentinelAsync(CancellationToken ct = default)
|
||||
{
|
||||
await EnsureContainerAsync(ct).ConfigureAwait(false);
|
||||
var blob = GetContainer().GetBlobClient(MaintenanceLockBlobName);
|
||||
var options = new BlobUploadOptions
|
||||
{
|
||||
HttpHeaders = new BlobHttpHeaders { ContentType = "application/octet-stream" },
|
||||
Conditions = new BlobRequestConditions { IfNoneMatch = ETag.All },
|
||||
};
|
||||
try
|
||||
{
|
||||
using var empty = new MemoryStream(Array.Empty<byte>());
|
||||
await blob.UploadAsync(empty, options, ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (RequestFailedException ex) when (IsAlreadyExists(ex))
|
||||
{
|
||||
// Sentinel exists already — perfect, that's the steady state.
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Attempts to acquire the maintenance lease for <paramref name="leaseDuration"/>
|
||||
/// seconds (15..60 per Azure spec; null = MaintenanceLeaseDuration). Throws
|
||||
/// <see cref="LeaseHeldException"/> on 409 (another holder); other errors propagate.
|
||||
/// Caller is responsible for renewing (every <duration) and releasing.
|
||||
/// </summary>
|
||||
public virtual async Task<BlobLeaseClient> AcquireMaintenanceLeaseAsync(TimeSpan? leaseDuration = null, CancellationToken ct = default)
|
||||
{
|
||||
await EnsureMaintenanceLockSentinelAsync(ct).ConfigureAwait(false);
|
||||
|
||||
var blob = GetContainer().GetBlobClient(MaintenanceLockBlobName);
|
||||
var leaseClient = blob.GetBlobLeaseClient();
|
||||
try
|
||||
{
|
||||
await leaseClient.AcquireAsync(leaseDuration ?? MaintenanceLeaseDuration, conditions: null, ct).ConfigureAwait(false);
|
||||
return leaseClient;
|
||||
}
|
||||
catch (RequestFailedException ex) when (ex.Status == 409)
|
||||
{
|
||||
throw new LeaseHeldException("maintenance lease is held by another operation", ex);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Per-table dump I/O ──────────────────────────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// Opens a writable stream for the per-table NDJSON dump. The block-blob is committed
|
||||
/// when the returned stream is disposed (await using), so the caller MUST wrap it in
|
||||
/// an <c>await using</c> block and dispose it BEFORE the manifest is written —
|
||||
/// otherwise the "manifest last" invariant breaks.
|
||||
/// </summary>
|
||||
public virtual async Task<Stream> OpenNdjsonWriteStreamAsync(string backupId, string tableName, CancellationToken ct = default)
|
||||
{
|
||||
if (string.IsNullOrEmpty(backupId)) throw new ArgumentException("backupId required", nameof(backupId));
|
||||
if (string.IsNullOrEmpty(tableName)) throw new ArgumentException("tableName required", nameof(tableName));
|
||||
|
||||
await EnsureContainerAsync(ct).ConfigureAwait(false);
|
||||
var blob = GetContainer().GetBlockBlobClient(BuildNdjsonBlobName(backupId, tableName));
|
||||
return await blob.OpenWriteAsync(overwrite: true, options: new BlockBlobOpenWriteOptions
|
||||
{
|
||||
HttpHeaders = new BlobHttpHeaders { ContentType = "application/x-ndjson" },
|
||||
}, cancellationToken: ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Final manifest write — fails (returns false) on If-None-Match collision so a
|
||||
/// backupId reuse is detected loudly. Manifest is the durability anchor (plan §3).
|
||||
/// </summary>
|
||||
public virtual async Task<bool> TryWriteManifestAsync(string backupId, byte[] manifestBytes, CancellationToken ct = default)
|
||||
{
|
||||
if (string.IsNullOrEmpty(backupId)) throw new ArgumentException("backupId required", nameof(backupId));
|
||||
if (manifestBytes == null) throw new ArgumentNullException(nameof(manifestBytes));
|
||||
|
||||
await EnsureContainerAsync(ct).ConfigureAwait(false);
|
||||
var blob = GetContainer().GetBlobClient(BuildManifestBlobName(backupId));
|
||||
var options = new BlobUploadOptions
|
||||
{
|
||||
HttpHeaders = new BlobHttpHeaders { ContentType = "application/json" },
|
||||
Conditions = new BlobRequestConditions { IfNoneMatch = ETag.All },
|
||||
};
|
||||
try
|
||||
{
|
||||
using var ms = new MemoryStream(manifestBytes);
|
||||
await blob.UploadAsync(ms, options, ct).ConfigureAwait(false);
|
||||
return true;
|
||||
}
|
||||
catch (RequestFailedException ex) when (IsAlreadyExists(ex))
|
||||
{
|
||||
_logger.LogError("BlobBackupStore.TryWriteManifestAsync: manifest for backupId {BackupId} already exists — backupId collision", backupId);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Reads the manifest bytes + ETag, or returns (null, null) if absent.</summary>
|
||||
public virtual async Task<(byte[]? Payload, ETag? ETag)> ReadManifestAsync(string backupId, CancellationToken ct = default)
|
||||
{
|
||||
if (string.IsNullOrEmpty(backupId)) throw new ArgumentException("backupId required", nameof(backupId));
|
||||
|
||||
await EnsureContainerAsync(ct).ConfigureAwait(false);
|
||||
var blob = GetContainer().GetBlobClient(BuildManifestBlobName(backupId));
|
||||
try
|
||||
{
|
||||
using var ms = new MemoryStream();
|
||||
await blob.DownloadToAsync(ms, ct).ConfigureAwait(false);
|
||||
var props = await blob.GetPropertiesAsync(cancellationToken: ct).ConfigureAwait(false);
|
||||
return (ms.ToArray(), props.Value.ETag);
|
||||
}
|
||||
catch (RequestFailedException ex) when (ex.Status == 404)
|
||||
{
|
||||
return (null, null);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Lists all backupIds present in the container (one entry per distinct first-level
|
||||
/// prefix), ignoring the <c>_lock/</c> prefix. Whether the manifest is actually
|
||||
/// present is up to the caller — this method returns ids of all <c>{id}/</c>
|
||||
/// prefixes regardless.
|
||||
/// </summary>
|
||||
public virtual async IAsyncEnumerable<string> ListBackupIdsAsync(
|
||||
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default)
|
||||
{
|
||||
await EnsureContainerAsync(ct).ConfigureAwait(false);
|
||||
var container = GetContainer();
|
||||
// GetBlobsByHierarchyAsync with delimiter="/" returns the top-level prefixes.
|
||||
var seen = new HashSet<string>(StringComparer.Ordinal);
|
||||
await foreach (var item in container.GetBlobsByHierarchyAsync(traits: BlobTraits.None, states: BlobStates.None, prefix: null, delimiter: "/", cancellationToken: ct).ConfigureAwait(false))
|
||||
{
|
||||
if (item.IsPrefix && item.Prefix is { Length: > 0 } p)
|
||||
{
|
||||
var trimmed = p.TrimEnd('/');
|
||||
if (trimmed.StartsWith("_lock", StringComparison.Ordinal)) continue;
|
||||
if (seen.Add(trimmed)) yield return trimmed;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Internals ───────────────────────────────────────────────────────────
|
||||
|
||||
internal static string BuildNdjsonBlobName(string backupId, string tableName) => $"{backupId}/{tableName}.ndjson";
|
||||
internal static string BuildManifestBlobName(string backupId) => $"{backupId}/manifest.json";
|
||||
|
||||
private BlobContainerClient GetContainer() => _blobs.GetContainerClient(Constants.BlobContainers.CriticalTableBackups);
|
||||
|
||||
private static bool IsAlreadyExists(RequestFailedException ex)
|
||||
=> ex.Status == 409
|
||||
|| ex.Status == 412
|
||||
|| ex.ErrorCode == BlobErrorCode.BlobAlreadyExists
|
||||
|| ex.ErrorCode == BlobErrorCode.ConditionNotMet;
|
||||
}
|
||||
|
||||
/// <summary>Thrown by <see cref="BlobBackupStore.AcquireMaintenanceLeaseAsync"/> on 409 conflict.</summary>
|
||||
public sealed class LeaseHeldException : Exception
|
||||
{
|
||||
public LeaseHeldException(string message, Exception inner) : base(message, inner) { }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,257 @@
|
||||
using System;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Security.Cryptography;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Functions.Services.Tables;
|
||||
using AutopilotMonitor.Shared;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// Streams all <see cref="Constants.CriticalBackupTables.All"/> tables into per-table
|
||||
/// NDJSON blobs, then writes a <c>manifest.json</c> LAST as the durability anchor.
|
||||
/// Plan §PR1.
|
||||
/// <para>
|
||||
/// LEASE OWNERSHIP boundary: this service neither acquires nor releases the
|
||||
/// maintenance lease. It assumes the caller (timer or queue worker) already holds it.
|
||||
/// The service also does not write to <c>BackupJobStatus</c> — that responsibility
|
||||
/// lives with the caller too, so the service stays Unit-testable without touching
|
||||
/// the table-storage layer.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class CriticalTableBackupService : ICriticalTableBackupService
|
||||
{
|
||||
/// <summary>Per-run wall-clock budget. 10 min safety headroom below the 60 min Function timeout.</summary>
|
||||
internal static readonly TimeSpan PerRunBudget = TimeSpan.FromMinutes(50);
|
||||
|
||||
private readonly TableStorageService _tables;
|
||||
private readonly BlobBackupStore _store;
|
||||
private readonly ILogger<CriticalTableBackupService> _logger;
|
||||
private readonly TimeProvider _clock;
|
||||
|
||||
public CriticalTableBackupService(
|
||||
TableStorageService tables,
|
||||
BlobBackupStore store,
|
||||
ILogger<CriticalTableBackupService> logger,
|
||||
TimeProvider? clock = null)
|
||||
{
|
||||
_tables = tables;
|
||||
_store = store;
|
||||
_logger = logger;
|
||||
_clock = clock ?? TimeProvider.System;
|
||||
}
|
||||
|
||||
public string GenerateBackupId()
|
||||
{
|
||||
var now = _clock.GetUtcNow().UtcDateTime;
|
||||
var stamp = now.ToString("yyyyMMddTHHmmssZ", System.Globalization.CultureInfo.InvariantCulture);
|
||||
var guidPart = Guid.NewGuid().ToString("N").Substring(0, 8);
|
||||
return $"{stamp}_{guidPart}";
|
||||
}
|
||||
|
||||
public async Task<BackupRunResult> RunBackupUnderLeaseAsync(string backupId, string triggeredBy, CancellationToken ct = default)
|
||||
{
|
||||
if (string.IsNullOrEmpty(backupId)) throw new ArgumentException("backupId required", nameof(backupId));
|
||||
if (string.IsNullOrEmpty(triggeredBy)) triggeredBy = "Unknown";
|
||||
|
||||
var startedAtUtc = _clock.GetUtcNow().UtcDateTime;
|
||||
var deadlineUtc = startedAtUtc + PerRunBudget;
|
||||
|
||||
var manifest = new CriticalTableBackupManifest
|
||||
{
|
||||
SchemaVersion = 1,
|
||||
BackupId = backupId,
|
||||
StartedAtUtc = startedAtUtc,
|
||||
TriggeredBy = triggeredBy,
|
||||
};
|
||||
|
||||
var degraded = false;
|
||||
|
||||
foreach (var tableName in Constants.CriticalBackupTables.All)
|
||||
{
|
||||
ct.ThrowIfCancellationRequested();
|
||||
|
||||
if (_clock.GetUtcNow().UtcDateTime >= deadlineUtc)
|
||||
{
|
||||
manifest.Tables.Add(new CriticalTableBackupTableEntry
|
||||
{
|
||||
TableName = tableName,
|
||||
Status = TableBackupStatus.Skipped,
|
||||
BlobName = BlobBackupStore.BuildNdjsonBlobName(backupId, tableName),
|
||||
ErrorMessage = "per-run budget exceeded before table started",
|
||||
});
|
||||
degraded = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
var entry = await DumpOneTableAsync(backupId, tableName, deadlineUtc, ct).ConfigureAwait(false);
|
||||
manifest.Tables.Add(entry);
|
||||
if (entry.Status == TableBackupStatus.Failed || entry.Status == TableBackupStatus.Skipped)
|
||||
{
|
||||
degraded = true;
|
||||
}
|
||||
}
|
||||
|
||||
manifest.CompletedAtUtc = _clock.GetUtcNow().UtcDateTime;
|
||||
manifest.Outcome = degraded ? BackupOutcome.Partial : BackupOutcome.Success;
|
||||
|
||||
// Manifest LAST. If this throws, caller treats run as Failed (no JobState=Completed,
|
||||
// no BackupOutcome). Existence of manifest.json is the durability anchor.
|
||||
var manifestBytes = JsonSerializer.SerializeToUtf8Bytes(manifest, BackupManifestJson.SerializerOptions);
|
||||
var ok = await _store.TryWriteManifestAsync(backupId, manifestBytes, ct).ConfigureAwait(false);
|
||||
if (!ok)
|
||||
{
|
||||
throw new InvalidOperationException($"Manifest for backupId {backupId} already exists — collision");
|
||||
}
|
||||
|
||||
return new BackupRunResult
|
||||
{
|
||||
Outcome = manifest.Outcome,
|
||||
Manifest = manifest,
|
||||
ManifestBlobName = BlobBackupStore.BuildManifestBlobName(backupId),
|
||||
};
|
||||
}
|
||||
|
||||
// ── Per-table loop body ─────────────────────────────────────────────────
|
||||
|
||||
private async Task<CriticalTableBackupTableEntry> DumpOneTableAsync(
|
||||
string backupId, string tableName, DateTime deadlineUtc, CancellationToken ct)
|
||||
{
|
||||
var entry = new CriticalTableBackupTableEntry
|
||||
{
|
||||
TableName = tableName,
|
||||
BlobName = BlobBackupStore.BuildNdjsonBlobName(backupId, tableName),
|
||||
};
|
||||
|
||||
long rows = 0;
|
||||
long byteSize = 0;
|
||||
string sha256Hex;
|
||||
|
||||
try
|
||||
{
|
||||
using var sha = SHA256.Create();
|
||||
// await using ensures the block-blob CommitBlockList runs BEFORE the next
|
||||
// table starts and BEFORE the manifest is written (plan Wave14 #3).
|
||||
await using var blobStream = await _store.OpenNdjsonWriteStreamAsync(backupId, tableName, ct).ConfigureAwait(false);
|
||||
// CountingStream tracks byte count without buffering; CryptoStream feeds the hash;
|
||||
// both wrap the actual blob stream so writes pass through each layer once.
|
||||
await using var counting = new CountingStream(blobStream);
|
||||
await using var crypto = new CryptoStream(counting, sha, CryptoStreamMode.Write, leaveOpen: true);
|
||||
|
||||
await foreach (var entity in _tables.EnumerateAllAsync(tableName, ct).ConfigureAwait(false))
|
||||
{
|
||||
ct.ThrowIfCancellationRequested();
|
||||
if (_clock.GetUtcNow().UtcDateTime >= deadlineUtc)
|
||||
throw new BudgetExceededException("per-run budget exceeded during table dump");
|
||||
|
||||
var dump = TableEntityDumpConverter.MapEntityToDump(entity);
|
||||
var line = JsonSerializer.SerializeToUtf8Bytes(dump, BackupManifestJson.SerializerOptions);
|
||||
await crypto.WriteAsync(line, ct).ConfigureAwait(false);
|
||||
await crypto.WriteAsync(NewlineBytes, ct).ConfigureAwait(false);
|
||||
rows++;
|
||||
}
|
||||
|
||||
// Final-block must flush BEFORE we read sha.Hash; CryptoStream does this on dispose,
|
||||
// but we need the bytes BEFORE the CountingStream/blobStream dispose. Manual flush:
|
||||
crypto.FlushFinalBlock();
|
||||
sha256Hex = Convert.ToHexString(sha.Hash ?? Array.Empty<byte>());
|
||||
byteSize = counting.BytesWritten;
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Renewal-loss / external cancel — never swallow into a per-table Failed.
|
||||
throw;
|
||||
}
|
||||
catch (BudgetExceededException ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "CriticalTableBackupService: budget exceeded during {Table}", tableName);
|
||||
entry.Status = TableBackupStatus.Skipped;
|
||||
entry.ErrorMessage = ex.Message;
|
||||
return entry;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "CriticalTableBackupService: table {Table} dump failed", tableName);
|
||||
entry.Status = TableBackupStatus.Failed;
|
||||
entry.ErrorMessage = ex.Message;
|
||||
return entry;
|
||||
}
|
||||
|
||||
entry.RowCount = rows;
|
||||
entry.ByteSize = byteSize;
|
||||
entry.Sha256Hex = sha256Hex;
|
||||
entry.Status = rows == 0 ? TableBackupStatus.Empty : TableBackupStatus.Ok;
|
||||
return entry;
|
||||
}
|
||||
|
||||
private static readonly byte[] NewlineBytes = Encoding.UTF8.GetBytes("\n");
|
||||
|
||||
// ── Counting stream wrapper ─────────────────────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// Tracks the byte-count of writes that flow through it. Wraps the blob stream so
|
||||
/// the service can stamp ByteSize on the manifest entry without re-reading the blob.
|
||||
/// </summary>
|
||||
private sealed class CountingStream : Stream
|
||||
{
|
||||
private readonly Stream _inner;
|
||||
private long _bytesWritten;
|
||||
|
||||
public CountingStream(Stream inner) { _inner = inner; }
|
||||
public long BytesWritten => _bytesWritten;
|
||||
|
||||
public override bool CanRead => false;
|
||||
public override bool CanSeek => false;
|
||||
public override bool CanWrite => true;
|
||||
public override long Length => _bytesWritten;
|
||||
public override long Position { get => _bytesWritten; set => throw new NotSupportedException(); }
|
||||
|
||||
public override void Flush() => _inner.Flush();
|
||||
public override Task FlushAsync(CancellationToken ct) => _inner.FlushAsync(ct);
|
||||
|
||||
public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
|
||||
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
|
||||
public override void SetLength(long value) => throw new NotSupportedException();
|
||||
|
||||
public override void Write(byte[] buffer, int offset, int count)
|
||||
{
|
||||
_inner.Write(buffer, offset, count);
|
||||
_bytesWritten += count;
|
||||
}
|
||||
|
||||
public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken ct)
|
||||
{
|
||||
await _inner.WriteAsync(buffer.AsMemory(offset, count), ct).ConfigureAwait(false);
|
||||
_bytesWritten += count;
|
||||
}
|
||||
|
||||
public override async ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken ct = default)
|
||||
{
|
||||
await _inner.WriteAsync(buffer, ct).ConfigureAwait(false);
|
||||
_bytesWritten += buffer.Length;
|
||||
}
|
||||
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
// Don't dispose _inner — the outer await using on the blob stream owns disposal.
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Thrown when the per-run wall-clock budget is exceeded mid-loop. Routed to
|
||||
/// <see cref="TableBackupStatus.Skipped"/> by its own catch — distinct from generic
|
||||
/// failures so the Outcome math (Partial) and operator action items stay clear.
|
||||
/// </summary>
|
||||
public sealed class BudgetExceededException : Exception
|
||||
{
|
||||
public BudgetExceededException(string message) : base(message) { }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// Backup orchestrator contract. The lease is ALWAYS owned by the caller (worker
|
||||
/// or timer); the service is intentionally lease-unaware so it can be tested in
|
||||
/// isolation. <see cref="GenerateBackupId"/> is split out so the caller can stamp
|
||||
/// the id onto <c>BackupJobStatus</c> before the heavy work begins (UI deep-link).
|
||||
/// </summary>
|
||||
public interface ICriticalTableBackupService
|
||||
{
|
||||
/// <summary>Pure id generator — no side effects. Format <c>yyyyMMddTHHmmssZ_{guid8}</c>.</summary>
|
||||
string GenerateBackupId();
|
||||
|
||||
/// <summary>
|
||||
/// Runs the full per-table backup loop under an externally-held maintenance lease.
|
||||
/// Returns a populated <see cref="BackupRunResult"/>; per-table errors are folded
|
||||
/// into the result with <see cref="BackupOutcome.Partial"/>. Fatal failures
|
||||
/// (manifest write throws, stream infra dies) propagate to the caller.
|
||||
/// </summary>
|
||||
Task<BackupRunResult> RunBackupUnderLeaseAsync(string backupId, string triggeredBy, CancellationToken ct = default);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Outcome of a single <c>RunBackupUnderLeaseAsync</c> call. The worker maps
|
||||
/// <see cref="BackupOutcome.Success"/>/<see cref="BackupOutcome.Partial"/> onto the
|
||||
/// matching JobStatus + OpsEvent; a thrown exception maps to JobState=Failed without
|
||||
/// a result object.
|
||||
/// </summary>
|
||||
public sealed class BackupRunResult
|
||||
{
|
||||
public BackupOutcome Outcome { get; set; }
|
||||
public CriticalTableBackupManifest Manifest { get; set; } = new();
|
||||
public string ManifestBlobName { get; set; } = string.Empty;
|
||||
}
|
||||
}
|
||||
+93
@@ -0,0 +1,93 @@
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Shared;
|
||||
using Azure.Identity;
|
||||
using Azure.Storage.Queues;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Newtonsoft.Json;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services.Backup.Queue
|
||||
{
|
||||
/// <summary>
|
||||
/// Azure-Storage-Queue producer for the critical-table backup feature.
|
||||
/// Mirrors <c>AzureQueueAnalyzeOnEnrollmentEndProducer</c>'s connection setup
|
||||
/// (Managed Identity preferred, connection string fallback) but flips the failure
|
||||
/// semantics: SendMessageAsync errors propagate so the HTTP trigger can mark
|
||||
/// the job Failed + return 5xx instead of leaving the operator with a 202 and
|
||||
/// no actual work scheduled.
|
||||
/// </summary>
|
||||
public sealed class AzureQueueCriticalTableBackupProducer : ICriticalTableBackupProducer
|
||||
{
|
||||
private readonly QueueClient _queueClient;
|
||||
private readonly ILogger<AzureQueueCriticalTableBackupProducer> _logger;
|
||||
private int _queueEnsured;
|
||||
|
||||
public AzureQueueCriticalTableBackupProducer(
|
||||
IConfiguration configuration,
|
||||
ILogger<AzureQueueCriticalTableBackupProducer> logger)
|
||||
{
|
||||
_logger = logger;
|
||||
|
||||
var options = new QueueClientOptions
|
||||
{
|
||||
MessageEncoding = QueueMessageEncoding.Base64,
|
||||
};
|
||||
|
||||
var storageAccountName = configuration["AzureStorageAccountName"];
|
||||
var connectionString = configuration["AzureTableStorageConnectionString"];
|
||||
|
||||
if (!string.IsNullOrEmpty(storageAccountName))
|
||||
{
|
||||
var queueUri = new Uri(
|
||||
$"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.CriticalTableBackup}");
|
||||
_queueClient = new QueueClient(queueUri, new DefaultAzureCredential(), options);
|
||||
_logger.LogInformation(
|
||||
"CriticalTableBackup producer initialized with Managed Identity (account: {Account})",
|
||||
storageAccountName);
|
||||
}
|
||||
else if (!string.IsNullOrEmpty(connectionString))
|
||||
{
|
||||
_queueClient = new QueueClient(
|
||||
connectionString, Constants.QueueNames.CriticalTableBackup, options);
|
||||
_logger.LogInformation("CriticalTableBackup producer initialized with connection string");
|
||||
}
|
||||
else
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
"Queue Storage not configured. Set either 'AzureStorageAccountName' or 'AzureTableStorageConnectionString'.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task EnqueueAsync(CriticalTableBackupEnvelope envelope, CancellationToken ct = default)
|
||||
{
|
||||
if (envelope == null) throw new ArgumentNullException(nameof(envelope));
|
||||
if (string.IsNullOrEmpty(envelope.JobId)) throw new ArgumentException("JobId required", nameof(envelope));
|
||||
|
||||
await EnsureQueueExistsAsync(ct).ConfigureAwait(false);
|
||||
|
||||
// Fail-hard: do NOT swallow. Caller (HTTP trigger) catches and rolls the
|
||||
// BackupJobStatus row to Failed + returns 500.
|
||||
var body = JsonConvert.SerializeObject(envelope);
|
||||
await _queueClient.SendMessageAsync(body, ct).ConfigureAwait(false);
|
||||
_logger.LogInformation("CriticalTableBackup enqueued (jobId={JobId})", envelope.JobId);
|
||||
}
|
||||
|
||||
private async Task EnsureQueueExistsAsync(CancellationToken ct)
|
||||
{
|
||||
if (_queueEnsured == 1) return;
|
||||
try
|
||||
{
|
||||
await _queueClient.CreateIfNotExistsAsync(cancellationToken: ct).ConfigureAwait(false);
|
||||
Interlocked.Exchange(ref _queueEnsured, 1);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Ensure failure during enqueue is itself a fail-hard signal — surface to the caller.
|
||||
_logger.LogError(ex, "CriticalTableBackup queue CreateIfNotExists failed");
|
||||
throw;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+438
@@ -0,0 +1,438 @@
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Functions.DataAccess.TableStorage;
|
||||
using AutopilotMonitor.Shared;
|
||||
using AutopilotMonitor.Shared.Models.Backup;
|
||||
using Azure;
|
||||
using Azure.Identity;
|
||||
using Azure.Storage.Blobs.Specialized;
|
||||
using Azure.Storage.Queues;
|
||||
using Azure.Storage.Queues.Models;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Newtonsoft.Json;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services.Backup.Queue
|
||||
{
|
||||
/// <summary>
|
||||
/// Background worker for the <c>critical-table-backup-jobs</c> queue. Drains messages,
|
||||
/// holds the maintenance lease for the duration of the per-job work, and runs
|
||||
/// <see cref="ICriticalTableBackupService.RunBackupUnderLeaseAsync"/> with the lease held.
|
||||
/// <para>
|
||||
/// <b>State-machine boundary (plan Wave17/18/19/22/23):</b> the WORKER owns the lease
|
||||
/// and JobStatus; the SERVICE is intentionally lease-unaware and stateless. State
|
||||
/// transitions follow this normative order: Duplicate-Detection → Acquire-Lease →
|
||||
/// CAS Queued→Running → Run (under lease) → CAS Running→Completed (under lease) →
|
||||
/// Release Lease → SafeDelete. Service-throw of a retryable exception triggers
|
||||
/// CAS Running→Queued (so the next dequeue sees Queued, not Running+free-lease) plus
|
||||
/// rethrow so the outer poll-loop applies VisibilityTimeout retry semantics.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Queue-config differs from short-lived workers</b> (plan Wave6): BatchSize=1,
|
||||
/// VisibilityTimeout=60min. The per-run service budget is 50min so the 60min visibility
|
||||
/// gives a 10min cushion — no explicit renewal loop is needed in PR1 (deferred as a
|
||||
/// hardening follow-up; Renewal-Loop adds ~150 LOC of race-handling that is only
|
||||
/// load-bearing for runs > 50min).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class CriticalTableBackupQueueWorker : BackgroundService
|
||||
{
|
||||
internal const int BatchSize = 1;
|
||||
internal static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(60);
|
||||
internal static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(10);
|
||||
internal static readonly TimeSpan ErrorBackoff = TimeSpan.FromSeconds(30);
|
||||
internal const int MaxDequeueCount = 5;
|
||||
|
||||
private const string PoisonQueueSuffix = "-poison";
|
||||
|
||||
private readonly QueueClient _mainQueue;
|
||||
private readonly QueueClient _poisonQueue;
|
||||
private readonly ICriticalTableBackupService _backupService;
|
||||
private readonly BlobBackupStore _blobStore;
|
||||
private readonly BackupJobsRepository _jobs;
|
||||
private readonly OpsEventService _opsEvents;
|
||||
private readonly ILogger<CriticalTableBackupQueueWorker> _logger;
|
||||
private readonly TimeProvider _clock;
|
||||
|
||||
public CriticalTableBackupQueueWorker(
|
||||
IConfiguration configuration,
|
||||
ICriticalTableBackupService backupService,
|
||||
BlobBackupStore blobStore,
|
||||
BackupJobsRepository jobs,
|
||||
OpsEventService opsEvents,
|
||||
ILogger<CriticalTableBackupQueueWorker> logger,
|
||||
TimeProvider? clock = null)
|
||||
{
|
||||
_backupService = backupService ?? throw new ArgumentNullException(nameof(backupService));
|
||||
_blobStore = blobStore ?? throw new ArgumentNullException(nameof(blobStore));
|
||||
_jobs = jobs ?? throw new ArgumentNullException(nameof(jobs));
|
||||
_opsEvents = opsEvents ?? throw new ArgumentNullException(nameof(opsEvents));
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
_clock = clock ?? TimeProvider.System;
|
||||
|
||||
var options = new QueueClientOptions { MessageEncoding = QueueMessageEncoding.Base64 };
|
||||
var storageAccountName = configuration["AzureStorageAccountName"];
|
||||
var connectionString = configuration["AzureTableStorageConnectionString"];
|
||||
|
||||
if (!string.IsNullOrEmpty(storageAccountName))
|
||||
{
|
||||
var mainUri = new Uri($"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.CriticalTableBackup}");
|
||||
var poisonUri = new Uri($"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.CriticalTableBackup}{PoisonQueueSuffix}");
|
||||
var credential = new DefaultAzureCredential();
|
||||
_mainQueue = new QueueClient(mainUri, credential, options);
|
||||
_poisonQueue = new QueueClient(poisonUri, credential, options);
|
||||
_logger.LogInformation(
|
||||
"CriticalTableBackupQueueWorker initialized with Managed Identity (account: {Account})",
|
||||
storageAccountName);
|
||||
}
|
||||
else if (!string.IsNullOrEmpty(connectionString))
|
||||
{
|
||||
_mainQueue = new QueueClient(connectionString, Constants.QueueNames.CriticalTableBackup, options);
|
||||
_poisonQueue = new QueueClient(connectionString, Constants.QueueNames.CriticalTableBackup + PoisonQueueSuffix, options);
|
||||
_logger.LogInformation("CriticalTableBackupQueueWorker initialized with connection string");
|
||||
}
|
||||
else
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
"Queue Storage not configured. Set either 'AzureStorageAccountName' or 'AzureTableStorageConnectionString'.");
|
||||
}
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
await TryCreateQueueAsync(_mainQueue, "main", stoppingToken).ConfigureAwait(false);
|
||||
await TryCreateQueueAsync(_poisonQueue, "poison", stoppingToken).ConfigureAwait(false);
|
||||
|
||||
_logger.LogInformation("CriticalTableBackupQueueWorker: poll loop started");
|
||||
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
var batch = await _mainQueue
|
||||
.ReceiveMessagesAsync(BatchSize, VisibilityTimeout, stoppingToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
if (batch?.Value is null || batch.Value.Length == 0)
|
||||
{
|
||||
await Task.Delay(PollInterval, stoppingToken).ConfigureAwait(false);
|
||||
continue;
|
||||
}
|
||||
|
||||
foreach (var msg in batch.Value)
|
||||
{
|
||||
if (stoppingToken.IsCancellationRequested) break;
|
||||
await ProcessOneAsync(msg, stoppingToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
break;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "CriticalTableBackupQueueWorker: poll-loop error — backing off {Backoff}", ErrorBackoff);
|
||||
try { await Task.Delay(ErrorBackoff, stoppingToken).ConfigureAwait(false); }
|
||||
catch (OperationCanceledException) { break; }
|
||||
}
|
||||
}
|
||||
|
||||
_logger.LogInformation("CriticalTableBackupQueueWorker: poll loop stopped");
|
||||
}
|
||||
|
||||
private async Task ProcessOneAsync(QueueMessage msg, CancellationToken ct)
|
||||
{
|
||||
if (msg.DequeueCount > MaxDequeueCount)
|
||||
{
|
||||
await MoveToPoisonAsync(msg, ct).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
CriticalTableBackupEnvelope? envelope;
|
||||
try
|
||||
{
|
||||
envelope = JsonConvert.DeserializeObject<CriticalTableBackupEnvelope>(msg.Body.ToString());
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "CriticalTableBackupQueueWorker: malformed envelope — dropping (msg {Id})", msg.MessageId);
|
||||
await SafeDeleteAsync(msg, ct).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
if (envelope is null || string.IsNullOrEmpty(envelope.JobId))
|
||||
{
|
||||
_logger.LogWarning("CriticalTableBackupQueueWorker: empty envelope — dropping (msg {Id})", msg.MessageId);
|
||||
await SafeDeleteAsync(msg, ct).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
await ProcessJobAsync(envelope.JobId, msg, ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task ProcessJobAsync(string jobId, QueueMessage msg, CancellationToken ct)
|
||||
{
|
||||
// ── Duplicate-Detection (plan Wave8 #2 + Wave9 #1) ──────────────────
|
||||
var (job, jobETag) = await _jobs.GetWithETagAsync(jobId, ct).ConfigureAwait(false);
|
||||
if (job is null)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"CriticalTableBackupQueueWorker: jobId {JobId} not in BackupJobs — dropping orphan message {Id}",
|
||||
jobId, msg.MessageId);
|
||||
await SafeDeleteAsync(msg, ct).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
if (IsTerminal(job.State))
|
||||
{
|
||||
_logger.LogInformation(
|
||||
"CriticalTableBackupQueueWorker: jobId {JobId} already terminal ({State}) — dropping reappearance",
|
||||
jobId, job.State);
|
||||
await SafeDeleteAsync(msg, ct).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
// Running on dequeue is rare but possible (timer race, prior worker crashed
|
||||
// post-CAS pre-release). Defer-visibility instead of clobbering a possibly
|
||||
// still-running peer's status (plan Wave9 #1 / Wave10 #4).
|
||||
if (job.State == BackupJobState.Running)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"CriticalTableBackupQueueWorker: jobId {JobId} state=Running on dequeue — deferring visibility, not mutating status",
|
||||
jobId);
|
||||
try
|
||||
{
|
||||
await _mainQueue.UpdateMessageAsync(msg.MessageId, msg.PopReceipt, visibilityTimeout: VisibilityTimeout, cancellationToken: ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex,
|
||||
"CriticalTableBackupQueueWorker: defer-visibility failed for jobId {JobId} — message will reappear naturally",
|
||||
jobId);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// ── Acquire maintenance lease BEFORE CAS Running ────────────────────
|
||||
BlobLeaseClient? lease = null;
|
||||
try
|
||||
{
|
||||
lease = await _blobStore.AcquireMaintenanceLeaseAsync(BlobBackupStore.MaintenanceLeaseDuration, ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (LeaseHeldException)
|
||||
{
|
||||
_logger.LogInformation(
|
||||
"CriticalTableBackupQueueWorker: maintenance lease held by another op — jobId {JobId} → Skipped",
|
||||
jobId);
|
||||
var skippedJob = job;
|
||||
skippedJob.State = BackupJobState.Skipped;
|
||||
skippedJob.LastHeartbeatUtc = _clock.GetUtcNow().UtcDateTime;
|
||||
skippedJob.CompletedAtUtc = _clock.GetUtcNow().UtcDateTime;
|
||||
skippedJob.Error = "maintenance lease held by another operation";
|
||||
await _jobs.TryUpdateWithCasAsync(skippedJob, jobETag!.Value, ct).ConfigureAwait(false);
|
||||
await TryRecordOpsEventAsync(() => _opsEvents.RecordCriticalTableBackupSkippedLockedAsync(skippedJob.Error, skippedJob.RequestedBy)).ConfigureAwait(false);
|
||||
await SafeDeleteAsync(msg, ct).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
// ── CAS Queued→Running + heartbeat (plan Wave14 #1) ─────────────
|
||||
var nowUtc = _clock.GetUtcNow().UtcDateTime;
|
||||
var runningJob = job;
|
||||
runningJob.State = BackupJobState.Running;
|
||||
runningJob.StartedAtUtc = nowUtc;
|
||||
runningJob.LastHeartbeatUtc = nowUtc;
|
||||
var casOk = await _jobs.TryUpdateWithCasAsync(runningJob, jobETag!.Value, ct).ConfigureAwait(false);
|
||||
if (!casOk)
|
||||
{
|
||||
// Concurrent state change (watchdog, parallel worker). Re-read + dispatch.
|
||||
var (fresh, _) = await _jobs.GetWithETagAsync(jobId, ct).ConfigureAwait(false);
|
||||
if (fresh is not null && IsTerminal(fresh.State))
|
||||
{
|
||||
await SafeDeleteAsync(msg, ct).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger.LogWarning("CriticalTableBackupQueueWorker: CAS Queued→Running 412 with non-terminal fresh state — leaving message for retry");
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// ── Service-call: backup itself (Wave22 #3 — service is lease-unaware) ──
|
||||
BackupRunResult? result = null;
|
||||
Exception? retryableException = null;
|
||||
try
|
||||
{
|
||||
var backupId = _backupService.GenerateBackupId();
|
||||
|
||||
// Stamp BackupId via CAS so UI can deep-link to the manifest the moment the worker has chosen the id.
|
||||
var (currentJob, currentETag) = await _jobs.GetWithETagAsync(jobId, ct).ConfigureAwait(false);
|
||||
if (currentJob is not null && currentETag is not null)
|
||||
{
|
||||
currentJob.BackupId = backupId;
|
||||
currentJob.LastHeartbeatUtc = _clock.GetUtcNow().UtcDateTime;
|
||||
await _jobs.TryUpdateWithCasAsync(currentJob, currentETag.Value, ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
result = await _backupService.RunBackupUnderLeaseAsync(backupId, runningJob.RequestedBy, ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Retryable: storage 5xx / network / generic bug. Roll JobState back to Queued
|
||||
// so the next dequeue sees a clean state (plan Wave25 #1+#2). Throw to outer
|
||||
// so visibility-timeout-retry kicks in. After 5 attempts the poison path
|
||||
// converts to JobState=Failed.
|
||||
retryableException = ex;
|
||||
}
|
||||
|
||||
if (retryableException is not null)
|
||||
{
|
||||
var (rb, rbETag) = await _jobs.GetWithETagAsync(jobId, ct).ConfigureAwait(false);
|
||||
if (rb is not null && rbETag is not null)
|
||||
{
|
||||
rb.State = BackupJobState.Queued;
|
||||
rb.BackupId = null; // next attempt gets a fresh id (orphan blobs are cleaned by lifecycle)
|
||||
rb.LastHeartbeatUtc = _clock.GetUtcNow().UtcDateTime;
|
||||
rb.Error = retryableException.Message;
|
||||
await _jobs.TryUpdateWithCasAsync(rb, rbETag.Value, ct).ConfigureAwait(false);
|
||||
}
|
||||
_logger.LogWarning(retryableException,
|
||||
"CriticalTableBackupQueueWorker: jobId {JobId} backup attempt failed — rolling to Queued for retry (dequeue {N})",
|
||||
jobId, msg.DequeueCount);
|
||||
throw retryableException; // outer loop applies visibility-timeout retry
|
||||
}
|
||||
|
||||
// ── Terminal CAS Completed (under lease) ────────────────────────
|
||||
var completedAt = _clock.GetUtcNow().UtcDateTime;
|
||||
var (finalJob, finalETag) = await _jobs.GetWithETagAsync(jobId, ct).ConfigureAwait(false);
|
||||
if (finalJob is not null && finalETag is not null)
|
||||
{
|
||||
finalJob.State = BackupJobState.Completed;
|
||||
finalJob.CompletedAtUtc = completedAt;
|
||||
finalJob.LastHeartbeatUtc = completedAt;
|
||||
finalJob.BackupOutcome = result!.Outcome;
|
||||
finalJob.Error = null;
|
||||
await _jobs.TryUpdateWithCasAsync(finalJob, finalETag.Value, ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// ── OpsEvent (outside the service try — non-throwing per Wave25 #3) ──
|
||||
var durationMs = (int)(completedAt - runningJob.StartedAtUtc!.Value).TotalMilliseconds;
|
||||
var failedOrSkipped = 0;
|
||||
foreach (var t in result!.Manifest.Tables)
|
||||
{
|
||||
if (t.Status == TableBackupStatus.Failed || t.Status == TableBackupStatus.Skipped) failedOrSkipped++;
|
||||
}
|
||||
await TryRecordOpsEventAsync(() => result.Outcome == BackupOutcome.Success
|
||||
? _opsEvents.RecordCriticalTableBackupCompletedAsync(
|
||||
result.Manifest.BackupId, result.Manifest.Tables.Count, durationMs,
|
||||
Constants.BlobContainers.CriticalTableBackups, result.ManifestBlobName, runningJob.RequestedBy)
|
||||
: _opsEvents.RecordCriticalTableBackupPartialAsync(
|
||||
result.Manifest.BackupId, result.Manifest.Tables.Count, failedOrSkipped, durationMs,
|
||||
Constants.BlobContainers.CriticalTableBackups, result.ManifestBlobName, runningJob.RequestedBy))
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
{
|
||||
// Plan Wave16 #2 / Wave17 #1: terminal status set BEFORE release; release before delete.
|
||||
if (lease is not null)
|
||||
{
|
||||
try { await lease.ReleaseAsync(cancellationToken: ct).ConfigureAwait(false); }
|
||||
catch (Exception ex) { _logger.LogWarning(ex, "CriticalTableBackupQueueWorker: lease release failed (auto-expires)"); }
|
||||
}
|
||||
}
|
||||
|
||||
// Successful path: delete the message.
|
||||
await SafeDeleteAsync(msg, ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// ── Helpers ─────────────────────────────────────────────────────────────
|
||||
|
||||
private async Task MoveToPoisonAsync(QueueMessage msg, CancellationToken ct)
|
||||
{
|
||||
// Plan Wave5 #1 + Wave22 #2: persist Failed BEFORE poison-move so the job
|
||||
// status never hangs in Running while the message is already poison.
|
||||
try
|
||||
{
|
||||
var envelope = JsonConvert.DeserializeObject<CriticalTableBackupEnvelope>(msg.Body.ToString());
|
||||
if (envelope is not null && !string.IsNullOrEmpty(envelope.JobId))
|
||||
{
|
||||
var (job, etag) = await _jobs.GetWithETagAsync(envelope.JobId, ct).ConfigureAwait(false);
|
||||
if (job is not null && etag is not null && !IsTerminal(job.State))
|
||||
{
|
||||
job.State = BackupJobState.Failed;
|
||||
job.CompletedAtUtc = _clock.GetUtcNow().UtcDateTime;
|
||||
job.LastHeartbeatUtc = job.CompletedAtUtc.Value;
|
||||
job.Error = $"poison-move after {msg.DequeueCount - 1} failed attempts";
|
||||
await _jobs.TryUpdateWithCasAsync(job, etag.Value, ct).ConfigureAwait(false);
|
||||
await TryRecordOpsEventAsync(() => _opsEvents.RecordCriticalTableBackupFailedAsync(job.BackupId, job.Error!, job.RequestedBy)).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "CriticalTableBackupQueueWorker: failed-state persistence before poison-move threw — proceeding with poison move anyway");
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await _poisonQueue.SendMessageAsync(msg.Body.ToString(), ct).ConfigureAwait(false);
|
||||
await SafeDeleteAsync(msg, ct).ConfigureAwait(false);
|
||||
_logger.LogWarning(
|
||||
"CriticalTableBackupQueueWorker: moved message {Id} to poison queue after {N} failed attempts",
|
||||
msg.MessageId, msg.DequeueCount - 1);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "CriticalTableBackupQueueWorker: poison move failed (will retry)");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task SafeDeleteAsync(QueueMessage msg, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
await _mainQueue.DeleteMessageAsync(msg.MessageId, msg.PopReceipt, ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex,
|
||||
"CriticalTableBackupQueueWorker: delete failed for message {Id} — will reappear after visibility-timeout",
|
||||
msg.MessageId);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task TryCreateQueueAsync(QueueClient queue, string label, CancellationToken ct)
|
||||
{
|
||||
try { await queue.CreateIfNotExistsAsync(cancellationToken: ct).ConfigureAwait(false); }
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex,
|
||||
"CriticalTableBackupQueueWorker: CreateIfNotExists failed for {Label} queue — send/receive will retry",
|
||||
label);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>OpsEvent write must not bubble — backup itself already succeeded (or failed by other means).</summary>
|
||||
private async Task TryRecordOpsEventAsync(Func<Task> writeAsync)
|
||||
{
|
||||
try { await writeAsync().ConfigureAwait(false); }
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "CriticalTableBackupQueueWorker: ops-event recording failed — backup state remains authoritative");
|
||||
}
|
||||
}
|
||||
|
||||
private static bool IsTerminal(BackupJobState state)
|
||||
=> state == BackupJobState.Completed
|
||||
|| state == BackupJobState.Failed
|
||||
|| state == BackupJobState.Skipped
|
||||
|| state == BackupJobState.BlockedTerminal;
|
||||
}
|
||||
}
|
||||
+26
@@ -0,0 +1,26 @@
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services.Backup.Queue
|
||||
{
|
||||
/// <summary>
|
||||
/// Envelope for a single backup-queue message. Tiny on purpose — all state lives
|
||||
/// in <c>BackupJobs</c>, the worker re-reads it on dequeue.
|
||||
/// </summary>
|
||||
public sealed class CriticalTableBackupEnvelope
|
||||
{
|
||||
public string JobId { get; set; } = string.Empty;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Producer for the <c>critical-table-backup-jobs</c> queue. <b>Fail-hard</b>:
|
||||
/// SendMessageAsync exceptions propagate so the HTTP trigger can roll the
|
||||
/// BackupJobStatus row forward to Failed and return 5xx — silent enqueue loss
|
||||
/// would leave the job stuck in Queued forever (no Manual-Analyze-Now-style
|
||||
/// fallback exists here).
|
||||
/// </summary>
|
||||
public interface ICriticalTableBackupProducer
|
||||
{
|
||||
Task EnqueueAsync(CriticalTableBackupEnvelope envelope, CancellationToken ct = default);
|
||||
}
|
||||
}
|
||||
@@ -34,6 +34,7 @@ namespace AutopilotMonitor.Functions.Services
|
||||
Constants.QueueNames.AnalyzeOnEnrollmentEnd + "-poison",
|
||||
Constants.QueueNames.VulnerabilityCorrelate + "-poison",
|
||||
Constants.QueueNames.TelemetryIndexReconcile + "-poison",
|
||||
Constants.QueueNames.CriticalTableBackupPoison,
|
||||
};
|
||||
|
||||
/// <summary>Default warning threshold — every poison message is a 5x-failed handler call.</summary>
|
||||
|
||||
@@ -254,6 +254,36 @@ namespace AutopilotMonitor.Functions.Services
|
||||
null, "System.Maintenance",
|
||||
new { queueName, count, threshold });
|
||||
|
||||
// ── Critical-Table Backup ─────────────────────────────────────────────
|
||||
|
||||
/// <summary>Backup run finished with all tables successfully captured. Info-level — visible in the timeline, not alertable by default.</summary>
|
||||
public Task RecordCriticalTableBackupCompletedAsync(string backupId, int tableCount, int durationMs, string container, string manifestBlobName, string triggeredBy)
|
||||
=> WriteAsync(OpsEventCategory.Maintenance, "CriticalTableBackupCompleted", OpsEventSeverity.Info,
|
||||
$"Critical-table backup {backupId} completed: {tableCount} tables, {durationMs}ms (triggeredBy={triggeredBy})",
|
||||
null, "System.Maintenance",
|
||||
new { backupId, tableCount, durationMs, container, manifestBlobName, triggeredBy });
|
||||
|
||||
/// <summary>Backup run wrote a manifest but at least one table Failed or Skipped. Warning-level — operator should inspect manifest perTableFailures.</summary>
|
||||
public Task RecordCriticalTableBackupPartialAsync(string backupId, int totalTables, int failedOrSkipped, int durationMs, string container, string manifestBlobName, string triggeredBy)
|
||||
=> WriteAsync(OpsEventCategory.Maintenance, "CriticalTableBackupPartial", OpsEventSeverity.Warning,
|
||||
$"Critical-table backup {backupId} PARTIAL: {failedOrSkipped}/{totalTables} tables failed or skipped, manifest written ({durationMs}ms, triggeredBy={triggeredBy})",
|
||||
null, "System.Maintenance",
|
||||
new { backupId, totalTables, failedOrSkipped, durationMs, container, manifestBlobName, triggeredBy });
|
||||
|
||||
/// <summary>Backup run never produced a valid manifest (fatal exception, storage outage). Error-level. Queue-path emits this AFTER 5x retry + poison-move; timer emits immediately.</summary>
|
||||
public Task RecordCriticalTableBackupFailedAsync(string? backupId, string errorMessage, string triggeredBy)
|
||||
=> WriteAsync(OpsEventCategory.Maintenance, "CriticalTableBackupFailed", OpsEventSeverity.Error,
|
||||
$"Critical-table backup FAILED (backupId={backupId ?? "n/a"}, triggeredBy={triggeredBy}): {errorMessage}",
|
||||
null, "System.Maintenance",
|
||||
new { backupId, errorMessage, triggeredBy });
|
||||
|
||||
/// <summary>Backup or restore was skipped because the maintenance lease was already held by another job. Info-level — not a failure.</summary>
|
||||
public Task RecordCriticalTableBackupSkippedLockedAsync(string reason, string triggeredBy)
|
||||
=> WriteAsync(OpsEventCategory.Maintenance, "CriticalTableBackupSkippedLocked", OpsEventSeverity.Info,
|
||||
$"Critical-table backup skipped — {reason} (triggeredBy={triggeredBy})",
|
||||
null, "System.Maintenance",
|
||||
new { reason, triggeredBy });
|
||||
|
||||
// ── Tenant ─────────────────────────────────────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Azure.Data.Tables;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services
|
||||
{
|
||||
/// <summary>
|
||||
/// Critical-table backup primitives on <see cref="TableStorageService"/> (plan §PR1).
|
||||
/// </summary>
|
||||
public partial class TableStorageService
|
||||
{
|
||||
/// <summary>
|
||||
/// Streams every row of <paramref name="tableName"/> as a generic
|
||||
/// <see cref="TableEntity"/>. Used by <c>CriticalTableBackupService</c> to feed
|
||||
/// the per-table NDJSON dump without loading the whole table into memory.
|
||||
/// Fail-loud per <c>feedback_storage_helpers_fail_soft</c>: storage errors
|
||||
/// propagate so the caller can decide whether to mark the table Failed and
|
||||
/// continue or to bubble.
|
||||
/// </summary>
|
||||
public virtual async IAsyncEnumerable<TableEntity> EnumerateAllAsync(
|
||||
string tableName,
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (string.IsNullOrEmpty(tableName))
|
||||
throw new ArgumentException("tableName is required", nameof(tableName));
|
||||
|
||||
var tableClient = _tableServiceClient.GetTableClient(tableName);
|
||||
await foreach (var entity in tableClient.QueryAsync<TableEntity>(cancellationToken: cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
yield return entity;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -215,6 +215,12 @@ namespace AutopilotMonitor.Shared
|
||||
|
||||
// Pre-auth distress channel (no authentication required)
|
||||
public const string ReportDistress = "/api/agent/distress";
|
||||
|
||||
// Critical-table backup feature (plan §PR1+, all GA-only)
|
||||
public const string GlobalBackupsTrigger = "/api/global/backups/trigger"; // POST → 202 + jobId
|
||||
public const string GlobalBackupsList = "/api/global/backups"; // GET → list of backups
|
||||
public const string GlobalBackupsManifest = "/api/global/backups/{backupId}"; // GET → manifest detail
|
||||
public const string GlobalBackupsJobStatus = "/api/global/backups/jobs/{jobId}"; // GET → BackupJobStatus
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
@@ -670,6 +676,17 @@ namespace AutopilotMonitor.Shared
|
||||
/// the blob in the customer's own storage and never touches this container.
|
||||
/// </summary>
|
||||
public const string HostedDiagnostics = "diagnostics";
|
||||
|
||||
/// <summary>
|
||||
/// Critical-table backup destination. Daily timer + manual GA trigger write
|
||||
/// per-table NDJSON dumps under <c>{backupId}/{tableName}.ndjson</c> plus a
|
||||
/// final <c>{backupId}/manifest.json</c>. Maintenance lease sentinel lives
|
||||
/// at <c>_lock/maintenance.lock</c> in the same container — coordinates
|
||||
/// backup + restore jobs across worker/timer/single-row paths.
|
||||
/// Lifecycle: 90-day delete on prefix <c>critical-table-backups/</c>; account
|
||||
/// settings (Soft-Delete + Versioning) provide defence-in-depth.
|
||||
/// </summary>
|
||||
public const string CriticalTableBackups = "critical-table-backups";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -729,6 +746,73 @@ namespace AutopilotMonitor.Shared
|
||||
|
||||
/// <summary>Poison sibling of <see cref="TenantOffboarding"/>.</summary>
|
||||
public const string TenantOffboardingPoison = "tenant-offboarding-poison";
|
||||
|
||||
/// <summary>
|
||||
/// Critical-table backup job queue. Producer = HTTP trigger
|
||||
/// <c>/api/global/backups/trigger</c> (fail-hard); consumer =
|
||||
/// <c>CriticalTableBackupQueueWorker</c> (BackgroundService, BatchSize=1,
|
||||
/// VisibilityTimeout=60min with 25min PopReceipt-renewal). Carries a small
|
||||
/// envelope <c>{ jobId }</c> — full state lives in <see cref="TableNames.BackupJobs"/>.
|
||||
/// Poison-suffix <c>-poison</c>, max-dequeue 5; Failed-state is persisted on
|
||||
/// Poison-Move (NOT on first throw — that would defeat retry).
|
||||
/// </summary>
|
||||
public const string CriticalTableBackup = "critical-table-backup-jobs";
|
||||
|
||||
/// <summary>Poison sibling of <see cref="CriticalTableBackup"/>.</summary>
|
||||
public const string CriticalTableBackupPoison = "critical-table-backup-jobs-poison";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Critical tables that the daily backup feature snapshots. Order matters only for
|
||||
/// deterministic test output; the per-table loop is independent so a failure in
|
||||
/// one table does NOT block the others (Outcome=Partial covers that case).
|
||||
/// Three of these (<c>GlobalAdmins</c>, <c>TenantAdmins</c>, <c>McpUsers</c>)
|
||||
/// are gated to single-row restore in the API layer — full-table restore would
|
||||
/// reactivate disabled or re-create removed identities.
|
||||
/// <c>OffboardingAudit</c> additionally blocks replace-all (append-only audit).
|
||||
/// </summary>
|
||||
public static class CriticalBackupTables
|
||||
{
|
||||
public static readonly string[] All = new[]
|
||||
{
|
||||
TableNames.AdminConfiguration,
|
||||
TableNames.AnalyzeRules,
|
||||
TableNames.Feedback,
|
||||
TableNames.GatherRules,
|
||||
TableNames.GlobalAdmins,
|
||||
TableNames.ImeLogPatterns,
|
||||
TableNames.ImeVersionHistory,
|
||||
TableNames.McpUsers,
|
||||
TableNames.OffboardingAudit,
|
||||
TableNames.PreviewConfig,
|
||||
TableNames.PreviewWhitelist,
|
||||
TableNames.RuleStates,
|
||||
TableNames.TenantAdmins,
|
||||
TableNames.TenantConfiguration,
|
||||
TableNames.TenantOffboardingCustomsArchive,
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// Full-table restore is API-blocked on these — the rows carry IsEnabled
|
||||
/// semantics whose accidental re-activation is a security incident.
|
||||
/// Single-row restore (with mandatory diff preview) remains allowed.
|
||||
/// </summary>
|
||||
public static readonly string[] AuthTablesFullRestoreForbidden = new[]
|
||||
{
|
||||
TableNames.GlobalAdmins,
|
||||
TableNames.TenantAdmins,
|
||||
TableNames.McpUsers,
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// replace-all strategy is API-blocked on these — append-only audit semantics
|
||||
/// (orphan-deletion would erase audit history added after the backup).
|
||||
/// upsert-only remains allowed.
|
||||
/// </summary>
|
||||
public static readonly string[] AuditTablesReplaceAllForbidden = new[]
|
||||
{
|
||||
TableNames.OffboardingAudit,
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
using System;
|
||||
|
||||
namespace AutopilotMonitor.Shared.Models.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// Per-job state machine for asynchronous backup + restore work. Persisted as
|
||||
/// <see cref="BackupJobStatusEntity"/> in <c>BackupJobs</c> (PartitionKey="BackupJobs",
|
||||
/// RowKey={jobId}). The domain DTO carries strongly-typed enums; the Repository
|
||||
/// layer maps from/to the string-typed Azure Table columns (Azure Tables has no
|
||||
/// EDM enum). Unknown strings on read MUST throw — silent fallback is a regression
|
||||
/// risk after enum refactorings.
|
||||
/// </summary>
|
||||
public sealed class BackupJobStatus
|
||||
{
|
||||
public string JobId { get; set; } = string.Empty;
|
||||
public BackupJobKind Kind { get; set; }
|
||||
public BackupJobState State { get; set; }
|
||||
|
||||
/// <summary>UPN for manual triggers; <c>"Timer"</c> never reaches this DTO — Timer skips the queue.</summary>
|
||||
public string RequestedBy { get; set; } = string.Empty;
|
||||
|
||||
public DateTime QueuedAtUtc { get; set; }
|
||||
public DateTime? StartedAtUtc { get; set; }
|
||||
public DateTime? CompletedAtUtc { get; set; }
|
||||
|
||||
/// <summary>Updated by the renewal-loop and per-phase progress writes; drives the watchdog.</summary>
|
||||
public DateTime LastHeartbeatUtc { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// For Kind=Backup: the worker-generated id of the manifest this job produced
|
||||
/// (null until the worker has acquired the lease and stamped it). UI uses this
|
||||
/// to deep-link from a completed job to its manifest detail page.
|
||||
/// </summary>
|
||||
public string? BackupId { get; set; }
|
||||
|
||||
/// <summary>For Kind=RestoreTable: the manifest id the operator is restoring FROM.</summary>
|
||||
public string? SourceBackupId { get; set; }
|
||||
|
||||
/// <summary>For Kind=RestoreTable: the table being restored.</summary>
|
||||
public string? TableName { get; set; }
|
||||
|
||||
/// <summary>For Kind=RestoreTable: <c>upsert-only</c> | <c>replace-all</c>.</summary>
|
||||
public string? Strategy { get; set; }
|
||||
|
||||
/// <summary>Free-form JSON for in-flight worker progress (phase, counters). Informational only — recovery does not depend on it.</summary>
|
||||
public string? Progress { get; set; }
|
||||
|
||||
public string? Error { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// For Kind=Backup with State=Completed: <c>Success</c> | <c>Partial</c>.
|
||||
/// Null for any other (Kind, State) combination.
|
||||
/// </summary>
|
||||
public BackupOutcome? BackupOutcome { get; set; }
|
||||
}
|
||||
|
||||
public enum BackupJobKind
|
||||
{
|
||||
Backup,
|
||||
RestoreTable,
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Six-state job machine. Terminal states (Completed / Failed / Skipped / BlockedTerminal)
|
||||
/// short-circuit the worker's duplicate-detection on reappearance: message is dropped, no
|
||||
/// re-run. Failed vs BlockedTerminal vs Skipped is intentional so operator-action follow-up
|
||||
/// can be distinguished (logs / re-trigger / nothing-to-do).
|
||||
/// </summary>
|
||||
public enum BackupJobState
|
||||
{
|
||||
Queued,
|
||||
Running,
|
||||
|
||||
/// <summary>Manifest written (Backup) or restore phase succeeded (RestoreTable).</summary>
|
||||
Completed,
|
||||
|
||||
/// <summary>Unexpected abort: storage 5xx / network / bug. After 5 retries → poison-move sets this.</summary>
|
||||
Failed,
|
||||
|
||||
/// <summary>Maintenance-lease held by another job — no work performed. Operator retriggers when the parallel job is done.</summary>
|
||||
Skipped,
|
||||
|
||||
/// <summary>Deterministic domain error: manifest missing, SHA mismatch, auth-table block, confirmation-token typo. Not retryable.</summary>
|
||||
BlockedTerminal,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Serialization;
|
||||
|
||||
namespace AutopilotMonitor.Shared.Models.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// Centralised JSON options for the critical-table backup feature: manifest write,
|
||||
/// NDJSON-line serialisation, HTTP response bodies. Deliberately distinct from
|
||||
/// <c>DeletionManifestJson.SerializerOptions</c> because the backup pipeline needs a
|
||||
/// <see cref="JsonStringEnumConverter"/> with <c>allowIntegerValues=false</c> so
|
||||
/// legacy or hand-crafted manifests with numeric enum values fail loudly instead of
|
||||
/// silently mapping to the wrong member.
|
||||
/// <para>
|
||||
/// <b>DictionaryKeyPolicy is intentionally NOT set</b> — <c>DeletionRowDump.Props</c>
|
||||
/// uses the original Azure Table column names (case-sensitive) as dictionary keys;
|
||||
/// CamelCasing them would create a new column on restore.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public static class BackupManifestJson
|
||||
{
|
||||
public static readonly JsonSerializerOptions SerializerOptions = new()
|
||||
{
|
||||
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
|
||||
// DictionaryKeyPolicy intentionally unset (see remarks above).
|
||||
Converters =
|
||||
{
|
||||
new JsonStringEnumConverter(namingPolicy: null, allowIntegerValues: false),
|
||||
},
|
||||
DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
|
||||
WriteIndented = false,
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace AutopilotMonitor.Shared.Models.Backup
|
||||
{
|
||||
/// <summary>
|
||||
/// Manifest written as the LAST blob in a backup run. Existence of
|
||||
/// <c>{backupId}/manifest.json</c> is the durability anchor: orphan NDJSON
|
||||
/// blobs under a prefix without a manifest are treated as <c>Incomplete</c>
|
||||
/// by the listing API and never offered for restore.
|
||||
/// <para>
|
||||
/// Schema-version is intentionally bumpable but starts at 1; restore code
|
||||
/// rejects unknown EDM types and unknown outcome strings with a clear error.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class CriticalTableBackupManifest
|
||||
{
|
||||
public int SchemaVersion { get; set; } = 1;
|
||||
|
||||
/// <summary>Format <c>yyyyMMddTHHmmssZ_{guid8}</c>.</summary>
|
||||
public string BackupId { get; set; } = string.Empty;
|
||||
|
||||
public DateTime StartedAtUtc { get; set; }
|
||||
public DateTime CompletedAtUtc { get; set; }
|
||||
|
||||
/// <summary>UPN of the operator who manually triggered the job, or <c>"Timer"</c> for the daily cron.</summary>
|
||||
public string TriggeredBy { get; set; } = "Timer";
|
||||
|
||||
/// <summary>Manifest-level outcome. NOT a JobState — see plan §JobStatus.</summary>
|
||||
public BackupOutcome Outcome { get; set; }
|
||||
|
||||
public List<CriticalTableBackupTableEntry> Tables { get; set; } = new();
|
||||
}
|
||||
|
||||
public sealed class CriticalTableBackupTableEntry
|
||||
{
|
||||
public string TableName { get; set; } = string.Empty;
|
||||
public TableBackupStatus Status { get; set; }
|
||||
public long RowCount { get; set; }
|
||||
public long ByteSize { get; set; }
|
||||
public string Sha256Hex { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Canonical: <c>{backupId}/{TableName}.ndjson</c>.</summary>
|
||||
public string BlobName { get; set; } = string.Empty;
|
||||
|
||||
public string? ErrorMessage { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Manifest-level outcome. Job-level state (Failed for fatal errors) is tracked
|
||||
/// separately on <c>BackupJobStatus</c>; a manifest only exists when the run
|
||||
/// reached the "all tables attempted, manifest write succeeded" milestone.
|
||||
/// </summary>
|
||||
public enum BackupOutcome
|
||||
{
|
||||
/// <summary>All tables backed up successfully.</summary>
|
||||
Success,
|
||||
|
||||
/// <summary>At least one table Failed or Skipped, but the manifest was still written
|
||||
/// for the remaining tables. Operationally valuable — 14 of 15 tables are still restorable.</summary>
|
||||
Partial,
|
||||
}
|
||||
|
||||
/// <summary>Per-table outcome inside a manifest.</summary>
|
||||
public enum TableBackupStatus
|
||||
{
|
||||
/// <summary>NDJSON blob written, SHA matched, RowCount > 0.</summary>
|
||||
Ok,
|
||||
|
||||
/// <summary>Table had 0 rows — empty NDJSON blob (and SHA-of-empty-string) is still committed for symmetry.</summary>
|
||||
Empty,
|
||||
|
||||
/// <summary>Per-run budget exceeded before this table was attempted (or finished). Operator can retry next run.</summary>
|
||||
Skipped,
|
||||
|
||||
/// <summary>An unexpected exception aborted this table's dump. Other tables continued.</summary>
|
||||
Failed,
|
||||
}
|
||||
}
|
||||
@@ -28,6 +28,15 @@ const OPS_EVENT_TYPES: Record<string, string[]> = {
|
||||
// tenant admins see only the lifecycle endpoints (deletion_started/completed/restored).
|
||||
// Operators wire Telegram on this event + read it in the Session Cleanup admin page.
|
||||
"SessionDeletionPoisoned",
|
||||
// Critical-table backup (plan §PR1) — dual-register per memory feedback_ops_event_types_dual_register.
|
||||
// Dispatched by CriticalTableBackupQueueWorker + CriticalTableBackupTimerFunction.
|
||||
// Backend helpers: RecordCriticalTableBackup{Completed,Partial,Failed,SkippedLocked}Async.
|
||||
// Partial is a distinct type (Warning severity) so operators can wire a softer
|
||||
// Telegram rule than for Failed (Error severity, fatal manifest-write failure).
|
||||
"CriticalTableBackupCompleted",
|
||||
"CriticalTableBackupPartial",
|
||||
"CriticalTableBackupFailed",
|
||||
"CriticalTableBackupSkippedLocked",
|
||||
],
|
||||
Security: [
|
||||
"DeviceBlocked",
|
||||
|
||||
Reference in New Issue
Block a user