From a833842e57830d5ba0a7de7b74147766ee1b4a13 Mon Sep 17 00:00:00 2001 From: Zacgoose <107489668+Zacgoose@users.noreply.github.com> Date: Fri, 21 Aug 2026 23:28:17 +0800 Subject: [PATCH 1/2] feat(orchestration): sweep orchestrator tables on a schedule and in full CleanupOldRunsAsync only ran from the startup recovery pass, only removed Completed/CompletedWithErrors/Failed runs that carried a CompletedUtc, and walked nothing but the Run partition. On a host that is not restarted the three tables grew without bound, and even across restarts Cancelled runs, runs nobody was driving, and Tasks/Results partitions whose Run row was already gone lived forever. The sweep now removes every terminal status (Cancelled included) once past retention, treats a non-terminal run as abandoned when it is not in _activeRuns and nothing about it - its !!run-counter heartbeat, Run row or StartedUtc - has been written within retention, and removes Tasks/Results partitions with no Run row once their newest row is past retention. The orphan scan reads keys and Timestamp only, through a new projected QueryTableAsync overload, so a Results partition's payload is never pulled back just to learn its age. Abandoned runs also lose their durable queue rows. It runs at the end of startup recovery and then every Orchestrator:CleanupIntervalHours (default 4) with Orchestrator:RetentionHours (default 48; it was a fixed 7 days). Both keys are documented in appsettings.example.jsonc and docs/configuration.md and pinned by ConfigurationReferenceTests. OrchestratorRetentionTests covers each rule against the in-memory store; OrchestratorRetentionAzuriteTests proves the projection and the partition deletes against real tables. --- .../Configuration/OrchestratorSettings.cs | 14 + Services/Orchestration/OrchestratorService.cs | 77 ++++- Services/Orchestration/SchedulerService.cs | 4 + Services/Storage/AzureTableStore.cs | 12 + Services/Storage/ICraftTableStore.cs | 13 + Services/Storage/OrchestratorCleanupResult.cs | 15 + Services/Storage/OrchestratorTableStore.cs | 131 +++++++- appsettings.example.jsonc | 5 + docs/configuration.md | 10 +- .../ConfigurationReferenceTests.cs | 2 + .../OrchestratorRetentionAzuriteTests.cs | 162 +++++++++ .../Craft.Tests/OrchestratorRetentionTests.cs | 314 ++++++++++++++++++ 12 files changed, 742 insertions(+), 17 deletions(-) create mode 100644 Services/Storage/OrchestratorCleanupResult.cs create mode 100644 tests/Craft.Tests/OrchestratorRetentionAzuriteTests.cs create mode 100644 tests/Craft.Tests/OrchestratorRetentionTests.cs diff --git a/Services/Configuration/OrchestratorSettings.cs b/Services/Configuration/OrchestratorSettings.cs index 786132b..d49004f 100644 --- a/Services/Configuration/OrchestratorSettings.cs +++ b/Services/Configuration/OrchestratorSettings.cs @@ -79,4 +79,18 @@ public class OrchestratorSettings /// Maximum number of times a task can be interrupted before being marked Failed. public int MaxRetries { get; set; } = 3; + + /// + /// How long a run's rows outlive it (hours, default 48). A run that finished — or that nothing is + /// driving and that last wrote to storage — longer ago than this is removed from all three tables, + /// together with any Tasks/Results partition whose Run row is already gone. Craft itself needs the + /// rows only while a run is live; they stay this long for operators reading recent history. + /// + public int RetentionHours { get; set; } = 48; + + /// + /// How often the retention sweep runs after the one at startup (hours, default 4). 0 disables the + /// periodic sweep; the startup pass, which follows crash recovery, still runs. + /// + public int CleanupIntervalHours { get; set; } = 4; } diff --git a/Services/Orchestration/OrchestratorService.cs b/Services/Orchestration/OrchestratorService.cs index 4cd4272..53c8f8c 100644 --- a/Services/Orchestration/OrchestratorService.cs +++ b/Services/Orchestration/OrchestratorService.cs @@ -538,14 +538,85 @@ public async Task ResumeInterruptedRunsAsync(CancellationToken ct) } } - // Cleanup old runs (older than 7 days) + // First retention pass, now that every run that could be resumed is back in _activeRuns and so + // exempt from the abandoned-run rule. The scheduler keeps it going on an interval from here. try { - await _store.CleanupOldRunsAsync(TimeSpan.FromDays(7)); + await RunRetentionSweepAsync(ct); } catch (Exception ex) { - _logger.LogWarning(ex, "[Scheduler] Failed to cleanup old runs"); + _logger.LogWarning(ex, "[Scheduler] Startup retention sweep failed"); + } + } + + /// + /// One retention pass over the orchestrator tables: finished runs past + /// Orchestrator:RetentionHours, runs nobody is driving that have not been written to for that + /// long, and Tasks/Results partitions whose Run row is already gone. Runs at the end of startup + /// recovery and then every Orchestrator:CleanupIntervalHours via . + /// + public async Task RunRetentionSweepAsync(CancellationToken ct) + { + var retention = TimeSpan.FromHours(Math.Max(1, _settings.Orchestrator.RetentionHours)); + var active = _activeRuns.Keys.ToHashSet(StringComparer.Ordinal); + var result = await _store.CleanupOldRunsAsync(retention, active, ct); + + // An abandoned run can still have rows in the durable queue. The pump would drop each as a + // stale descriptor when it came to claim it — but only after paying for the claim. + foreach (var name in result.AbandonedRuns) + { + try + { + await _queue.RemoveRunAsync(name, ct); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "[Scheduler] Could not remove queue rows for abandoned run {Name}", name); + } + } + + return result; + } + + /// + /// Periodic retention sweeps for the life of the host, started by the scheduler once recovery has + /// run. A sweep that fails is logged and tried again next interval; CleanupIntervalHours of 0 + /// leaves only the startup pass. + /// + public async Task RunRetentionLoopAsync(CancellationToken ct) + { + var hours = _settings.Orchestrator.CleanupIntervalHours; + if (hours <= 0) + { + _logger.LogInformation( + "[Scheduler] Periodic retention sweep disabled (CleanupIntervalHours={Hours}); only the startup pass runs", hours); + return; + } + + var interval = TimeSpan.FromHours(hours); + using var timer = new PeriodicTimer(interval); + try + { + while (await timer.WaitForNextTickAsync(ct)) + { + try + { + await RunRetentionSweepAsync(ct); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + return; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "[Scheduler] Retention sweep failed; next attempt in {Interval}", interval); + } + } + } + catch (OperationCanceledException) + { + // Host shutdown. } } diff --git a/Services/Orchestration/SchedulerService.cs b/Services/Orchestration/SchedulerService.cs index 0ce3fd6..d85d952 100644 --- a/Services/Orchestration/SchedulerService.cs +++ b/Services/Orchestration/SchedulerService.cs @@ -94,6 +94,10 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) _logger.LogError(ex, "[Scheduler] Failed to resume interrupted orchestrator runs"); } + // Retention sweeps for the rest of the process lifetime; recovery ran the first one. Fire and + // forget is deliberate: the loop handles its own failures and ends with the stopping token. + _ = _orchestrator.RunRetentionLoopAsync(stoppingToken); + while (!stoppingToken.IsCancellationRequested) { var now = DateTimeOffset.UtcNow; diff --git a/Services/Storage/AzureTableStore.cs b/Services/Storage/AzureTableStore.cs index 18f2899..e2f430c 100644 --- a/Services/Storage/AzureTableStore.cs +++ b/Services/Storage/AzureTableStore.cs @@ -209,6 +209,18 @@ public async IAsyncEnumerable QueryTableAsync(string table, string? fi yield return ToRow(entity); } + /// + /// The filtered scan with a $select, for callers that want keys and a stamp rather than the + /// row. The retention sweep reads every Results row's partition this way, and a Results row is a + /// 64 KiB chunk of payload it has no use for. + /// + public async IAsyncEnumerable QueryTableAsync(string table, string? filter, IReadOnlyList? properties, + [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default) + { + await foreach (var entity in Client(table).QueryAsync(filter: filter, select: properties, cancellationToken: ct)) + yield return ToRow(entity); + } + public async Task DeleteAsync(string table, string partitionKey, string rowKey, CancellationToken ct = default) { try diff --git a/Services/Storage/ICraftTableStore.cs b/Services/Storage/ICraftTableStore.cs index 783b838..9a8f883 100644 --- a/Services/Storage/ICraftTableStore.cs +++ b/Services/Storage/ICraftTableStore.cs @@ -70,6 +70,19 @@ Task TryReplaceBatchAsync(string table, string partitionKey, IReadOnlyList IAsyncEnumerable QueryTableAsync(string table, string? filter, CancellationToken ct = default) => QueryTableAsync(table, ct); + /// + /// The filtered scan, additionally projected to (a backend that + /// honours it still returns the keys and Timestamp) so that wide rows are not shipped just to read + /// their keys. + /// + /// Same contract as the filter: an optimisation a backend may ignore. This default returns full + /// rows, so a caller must only ever READ the properties it asked for and must not take a property's + /// absence to mean anything. + /// + IAsyncEnumerable QueryTableAsync(string table, string? filter, IReadOnlyList? properties, + CancellationToken ct = default) + => QueryTableAsync(table, filter, ct); + /// Delete a single row. A missing row is not an error. Task DeleteAsync(string table, string partitionKey, string rowKey, CancellationToken ct = default); diff --git a/Services/Storage/OrchestratorCleanupResult.cs b/Services/Storage/OrchestratorCleanupResult.cs new file mode 100644 index 0000000..959bb54 --- /dev/null +++ b/Services/Storage/OrchestratorCleanupResult.cs @@ -0,0 +1,15 @@ +namespace Craft.Storage; + +/// +/// What one pass removed, so the caller can +/// log it and take the follow-up that is not the store's business (an abandoned run's queue rows). +/// +/// Run rows scanned. +/// Runs that had finished and were past retention. +/// Runs that had not finished, were not active, and had not been written to within retention. +/// Tasks/Results partitions with no Run row whose newest row was past retention. +public sealed record OrchestratorCleanupResult( + int RunsExamined, + IReadOnlyList ExpiredRuns, + IReadOnlyList AbandonedRuns, + int OrphanPartitionsRemoved); diff --git a/Services/Storage/OrchestratorTableStore.cs b/Services/Storage/OrchestratorTableStore.cs index 7880d7d..75b6523 100644 --- a/Services/Storage/OrchestratorTableStore.cs +++ b/Services/Storage/OrchestratorTableStore.cs @@ -794,31 +794,136 @@ public async Task CleanupRunAsync(string runName) } } - /// Delete all runs (and their tasks/results) older than the retention period. - public async Task CleanupOldRunsAsync(TimeSpan retention) + /// The statuses a RUN ends in. Distinct from , which is about tasks. + private static bool IsTerminalRun(string? status) => + status is "Completed" or "CompletedWithErrors" or "Failed" or "Cancelled"; + + /// Keys and Timestamp only — what the orphan scan needs, and nothing a Results chunk carries. + private static readonly string[] s_keysAndTimestamp = ["PartitionKey", "RowKey", "Timestamp"]; + + /// + /// Retention sweep over the three tables. Everything removed is decided per run: + /// + /// A run in a terminal status (Completed, CompletedWithErrors, Failed, Cancelled) whose + /// CompletedUtc — or StartedUtc, for a row written before completion was stamped — is older than + /// loses its Run row and its Tasks and Results partitions. + /// A run in any other status is exempt while it is in (this + /// process is driving it). Otherwise it is abandoned once nothing about it has been written for + /// : task completion is written in one transaction with the + /// '!!run-counter' row, so that row's Timestamp is the heartbeat, and the Run row's own Timestamp + /// and StartedUtc count too. That covers runs recovery could not resume (task script gone), runs + /// queued but never dispatched, and — on a host that shares the tables — runs another process + /// stopped driving. + /// A Tasks or Results partition with no Run row at all is removed once its newest row is + /// older than . Those come from racing a + /// late status write, and from a Run row deleted while its partitions were still being written; + /// nothing else ever looked at them. + /// + /// This used to consider only Completed/CompletedWithErrors/Failed runs that carried a CompletedUtc, + /// and ran only from the startup recovery pass — so Cancelled runs, abandoned runs and orphaned + /// partitions lived forever, and on a host that was not restarted so did everything else. + /// + public async Task CleanupOldRunsAsync(TimeSpan retention, + IReadOnlySet? activeRuns = null, CancellationToken ct = default) { var cutoff = DateTimeOffset.UtcNow - retention; + var known = new HashSet(StringComparer.Ordinal); + var expired = new List(); + var abandoned = new List(); // Collect first, then delete — avoids mutating the "Run" partition while enumerating it. - var toClean = new List(); - await foreach (var row in _store.QueryPartitionAsync(_runsTable, "Run")) + var runs = new List(); + await foreach (var row in _store.QueryPartitionAsync(_runsTable, "Run", ct)) + runs.Add(row); + + foreach (var row in runs) { - var completedUtc = row.GetDateTimeOffset("CompletedUtc"); - var status = row.GetString("Status"); + known.Add(row.RowKey); - if (status is "Completed" or "CompletedWithErrors" or "Failed" - && completedUtc.HasValue && completedUtc.Value < cutoff) + if (IsTerminalRun(row.GetString("Status"))) { - toClean.Add(row.RowKey); + var ended = row.GetDateTimeOffset("CompletedUtc") + ?? row.GetDateTimeOffset("StartedUtc") + ?? row.Timestamp; + if (ended < cutoff) expired.Add(row.RowKey); + continue; } + + if (activeRuns != null && activeRuns.Contains(row.RowKey)) continue; + + var counter = await _store.GetAsync(_tasksTable, row.RowKey, CounterRowKey, ct); + var lastActivity = Newest(counter?.Timestamp, row.Timestamp, row.GetDateTimeOffset("StartedUtc")); + if (lastActivity < cutoff) abandoned.Add(row.RowKey); } - foreach (var name in toClean) + foreach (var name in expired) await CleanupRunAsync(name); + foreach (var name in abandoned) + { + _logger.LogInformation( + "[OrchestratorStore] Run {Name} is not active and has not been written to for {Hours:F0}h — treating it as abandoned", + name, retention.TotalHours); + await CleanupRunAsync(name); + } + + var orphans = 0; + foreach (var table in new[] { _tasksTable, _resultsTable }) + orphans += await CleanupOrphanPartitionsAsync(table, known, cutoff, ct); + + if (expired.Count + abandoned.Count + orphans > 0) + _logger.LogInformation( + "[OrchestratorStore] Retention sweep removed {Expired} finished run(s), {Abandoned} abandoned run(s) and {Orphans} orphaned partition(s) older than {Hours:F0}h ({Examined} runs examined)", + expired.Count, abandoned.Count, orphans, retention.TotalHours, runs.Count); + + return new OrchestratorCleanupResult(runs.Count, expired, abandoned, orphans); + } + + private async Task CleanupOrphanPartitionsAsync(string table, HashSet knownRuns, + DateTimeOffset cutoff, CancellationToken ct) + { + // Newest row per partition that has no Run row. Only keys and Timestamp travel: a Results + // partition IS the run's payload, and reading that back every sweep would be the cost this + // sweep exists to avoid. A backend that ignores the projection still answers correctly, just + // expensively. + var newest = new Dictionary(StringComparer.Ordinal); + var unstamped = new HashSet(StringComparer.Ordinal); + await foreach (var row in _store.QueryTableAsync(table, null, s_keysAndTimestamp, ct)) + { + if (knownRuns.Contains(row.PartitionKey)) continue; + if (row.Timestamp is not { } stamped) + { + // No way to tell how old it is — never guess in the direction of deleting. + unstamped.Add(row.PartitionKey); + continue; + } + if (!newest.TryGetValue(row.PartitionKey, out var current) || stamped > current) + newest[row.PartitionKey] = stamped; + } - if (toClean.Count > 0) - _logger.LogInformation("[OrchestratorStore] Cleaned up {Count} old runs (retention: {Days}d)", - toClean.Count, retention.TotalDays); + var removed = 0; + foreach (var (partition, stamped) in newest) + { + if (stamped >= cutoff || unstamped.Contains(partition)) continue; + try + { + await _store.DeletePartitionAsync(table, partition, ct); + removed++; + _logger.LogInformation("[OrchestratorStore] Removed orphaned partition {Table}/{Partition}", table, partition); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "[OrchestratorStore] Failed to remove orphaned partition {Table}/{Partition}", table, partition); + } + } + return removed; + } + + private static DateTimeOffset? Newest(params DateTimeOffset?[] candidates) + { + DateTimeOffset? newest = null; + foreach (var candidate in candidates) + if (candidate.HasValue && (!newest.HasValue || candidate.Value > newest.Value)) newest = candidate; + return newest; } /// Split a string into chunks of at most maxChars characters, avoiding surrogate splits. diff --git a/appsettings.example.jsonc b/appsettings.example.jsonc index b6bd88d..d6159b4 100644 --- a/appsettings.example.jsonc +++ b/appsettings.example.jsonc @@ -378,6 +378,11 @@ "TablePrefix": "Orchestrator", // Max task interruptions (crash/restart) before marking Failed "MaxRetries": 3, + // Retention sweep over the three tables: runs that finished (or, if nothing is driving them, last + // wrote to storage) longer ago than RetentionHours go, as do Tasks/Results partitions whose Run row + // is gone. Once at startup after crash recovery, then every CleanupIntervalHours (0 = startup only). + "RetentionHours": 48, + "CleanupIntervalHours": 4, // Batch + coalesce per-task/run status writes off the fan-out critical path (results are never batched). // Default true. Removes the per-task Azure Table write that gates worker throughput — see // docs/orch-analysis.md. Set false for the original per-task writes. diff --git a/docs/configuration.md b/docs/configuration.md index 3b6ceb2..25f013a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -408,7 +408,15 @@ Fan-out/fan-in task execution with crash recovery. "PostExecFunction": "Invoke-CraftPostExecution", // Max task interruptions (host crash/restart) before marking Failed. - "MaxRetries": 3 + "MaxRetries": 3, + + // Retention sweep over the three tables. A run that finished — or that nothing is driving and that + // last wrote to storage — longer ago than RetentionHours is removed together with its Tasks/Results + // partitions, as is any Tasks/Results partition whose Run row is already gone. Runs once at startup + // (after crash recovery) and then every CleanupIntervalHours; 0 keeps only the startup pass. Craft + // needs the rows only while a run is live — the retention is for operators reading recent history. + "RetentionHours": 48, + "CleanupIntervalHours": 4 } ``` diff --git a/tests/Craft.Tests/ConfigurationReferenceTests.cs b/tests/Craft.Tests/ConfigurationReferenceTests.cs index 8b032cc..435bd0d 100644 --- a/tests/Craft.Tests/ConfigurationReferenceTests.cs +++ b/tests/Craft.Tests/ConfigurationReferenceTests.cs @@ -73,6 +73,8 @@ public static TheoryData DocumentedDefaults() { "App:Scheduler:ApplyTZOffset", settings.Scheduler.ApplyTZOffset }, { "App:Orchestrator:TablePrefix", settings.Orchestrator.TablePrefix }, { "App:Orchestrator:MaxRetries", settings.Orchestrator.MaxRetries }, + { "App:Orchestrator:RetentionHours", settings.Orchestrator.RetentionHours }, + { "App:Orchestrator:CleanupIntervalHours", settings.Orchestrator.CleanupIntervalHours }, { "App:Cache:MaxEntries", settings.Cache.MaxEntries }, { "App:Cache:DefaultTtlSeconds", settings.Cache.DefaultTtlSeconds }, { "App:Scripts:PermissionExtraction:Enabled", settings.Scripts.PermissionExtraction.Enabled }, diff --git a/tests/Craft.Tests/OrchestratorRetentionAzuriteTests.cs b/tests/Craft.Tests/OrchestratorRetentionAzuriteTests.cs new file mode 100644 index 0000000..a86369f --- /dev/null +++ b/tests/Craft.Tests/OrchestratorRetentionAzuriteTests.cs @@ -0,0 +1,162 @@ +using Azure; +using Azure.Data.Tables; +using Craft.Configuration; +using Craft.Orchestration; +using Craft.Storage; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Craft.Tests; + +/// +/// The retention sweep against real tables (Azurite, or a storage account via +/// CRAFT_TEST_TABLE_CONNECTION). Two things only the real backend can prove: the orphan scan's +/// $select projection is accepted and still yields PartitionKey and Timestamp, and +/// DeletePartitionAsync really empties a partition written through the normal paths. Skipped, not +/// failed, when no backend is reachable — with the same caveat as +/// : a skip looks like a pass. +/// +public class OrchestratorRetentionAzuriteTests +{ + private sealed class Fixture : IAsyncDisposable + { + public required OrchestratorTableStore Store { get; init; } + public required AzureTableStore Backing { get; init; } + public required CraftSettings Settings { get; init; } + public required string Connection { get; init; } + + public static async Task TryConnectAsync() + { + var settings = new CraftSettings(); + var connection = Environment.GetEnvironmentVariable("CRAFT_TEST_TABLE_CONNECTION"); + if (!string.IsNullOrWhiteSpace(connection)) + settings.Auth.UserStorageConnection = connection; + else + { + settings.Storage.AllowDevelopmentStorage = true; + connection = "UseDevelopmentStorage=true"; + } + + // Unique per run so repeated runs cannot see each other's rows. + settings.Orchestrator.TablePrefix = "azrt" + Guid.NewGuid().ToString("N")[..8]; + + var backing = new AzureTableStore(settings); + try + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(3)); + await backing.PingAsync(cts.Token); + } + catch + { + return null; + } + + var store = new OrchestratorTableStore(NullLogger.Instance, settings, backing); + await store.InitializeAsync(); + return new Fixture { Store = store, Backing = backing, Settings = settings, Connection = connection }; + } + + public string Table(string suffix) => Settings.Orchestrator.TablePrefix + suffix; + + public async Task CountAsync(string suffix, string partition) + { + var n = 0; + await foreach (var _ in Backing.QueryPartitionAsync(Table(suffix), partition)) n++; + return n; + } + + public Task AddRunAsync(string name, string status, DateTime started, DateTime? completed) => + Store.UpsertRunAsync(new OrchestratorRun { Name = name, Status = status, StartedUtc = started, CompletedUtc = completed }); + + public Task AddTasksAsync(string run, int count) => + Store.UpsertTaskBatchAsync(run, Enumerable.Range(1, count) + .Select(i => new OrchestratorTaskItem { Id = $"task-{i}", Status = "Completed" }).ToList()); + + public async ValueTask DisposeAsync() + { + // Drop the per-run tables so repeated local runs do not pile fixtures into the emulator. + var service = new TableServiceClient(Connection); + foreach (var suffix in new[] { "Runs", "Tasks", "Results" }) + { + try { await service.DeleteTableAsync(Table(suffix)); } + catch (RequestFailedException) { /* already gone */ } + } + } + } + + [Fact] + public async Task Sweep_RemovesFinishedRunsPastRetention_AndKeepsEverythingStillWanted() + { + await using var fx = await Fixture.TryConnectAsync(); + if (fx == null) return; + + var old = DateTime.UtcNow.AddDays(-3); + var recent = DateTime.UtcNow.AddHours(-1); + + await fx.AddRunAsync("done-old", "Completed", old, old); + await fx.AddTasksAsync("done-old", 3); + await fx.Store.StoreResultAsync("done-old", "task-1", "{\"ok\":true}"); + + await fx.AddRunAsync("cancelled-old", "Cancelled", old, old); + await fx.AddTasksAsync("cancelled-old", 1); + + await fx.AddRunAsync("failed-recent", "Failed", recent, recent); + await fx.AddTasksAsync("failed-recent", 2); + + // Started long ago, nobody in this process drives it, but its counter row was written just + // now — the heartbeat that says another process (or this one, moments ago) is still at it. + await fx.AddRunAsync("running", "Running", DateTime.UtcNow.AddDays(-10), null); + await fx.AddTasksAsync("running", 2); + await fx.Store.InitRemainingAsync("running", 2); + + // No Run row, but written moments ago: an orphan, just not an old one. + await fx.AddTasksAsync("ghost", 2); + + var result = await fx.Store.CleanupOldRunsAsync(TimeSpan.FromHours(48), new HashSet()); + + Assert.Collection(result.ExpiredRuns.OrderBy(n => n, StringComparer.Ordinal), + n => Assert.Equal("cancelled-old", n), + n => Assert.Equal("done-old", n)); + Assert.Empty(result.AbandonedRuns); + Assert.Equal(0, result.OrphanPartitionsRemoved); + Assert.Equal(4, result.RunsExamined); + + Assert.Null(await fx.Store.GetRunAsync("done-old")); + Assert.Null(await fx.Store.GetRunAsync("cancelled-old")); + Assert.Equal(0, await fx.CountAsync("Tasks", "done-old")); + Assert.Equal(0, await fx.CountAsync("Results", "done-old")); + Assert.Equal(0, await fx.CountAsync("Tasks", "cancelled-old")); + + Assert.NotNull(await fx.Store.GetRunAsync("failed-recent")); + Assert.Equal(2, await fx.CountAsync("Tasks", "failed-recent")); + Assert.Equal(2, (await fx.Store.GetRunAsync("running"))!.Tasks.Count); + Assert.Equal(2, await fx.CountAsync("Tasks", "ghost")); + } + + [Fact] + public async Task Sweep_RemovesOrphanedPartitions_OnceNothingInThemIsRecent() + { + await using var fx = await Fixture.TryConnectAsync(); + if (fx == null) return; + + await fx.AddTasksAsync("ghost", 3); + await fx.Store.StoreResultAsync("ghost", "task-1", "{\"ok\":true}"); + + // A run this process is driving: exempt from the abandoned rule, and its partition is never + // an orphan because its Run row is there. + await fx.AddRunAsync("kept", "Running", DateTime.UtcNow, null); + await fx.AddTasksAsync("kept", 2); + + // Every row was stamped by the service a moment ago. A cutoff in the future is the only way + // to make them "old" without waiting, and it also absorbs any skew between the emulator's + // clock and this process's. + var result = await fx.Store.CleanupOldRunsAsync(TimeSpan.FromMinutes(-5), new HashSet { "kept" }); + + Assert.Equal(2, result.OrphanPartitionsRemoved); + Assert.Empty(result.ExpiredRuns); + Assert.Empty(result.AbandonedRuns); + Assert.Equal(0, await fx.CountAsync("Tasks", "ghost")); + Assert.Equal(0, await fx.CountAsync("Results", "ghost")); + Assert.Equal(2, await fx.CountAsync("Tasks", "kept")); + Assert.NotNull(await fx.Store.GetRunAsync("kept")); + } +} diff --git a/tests/Craft.Tests/OrchestratorRetentionTests.cs b/tests/Craft.Tests/OrchestratorRetentionTests.cs new file mode 100644 index 0000000..4f833a2 --- /dev/null +++ b/tests/Craft.Tests/OrchestratorRetentionTests.cs @@ -0,0 +1,314 @@ +using Craft.Configuration; +using Craft.Orchestration; +using Craft.Storage; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Craft.Tests; + +/// +/// The retention sweep is the only thing that bounds the orchestrator tables on a host that is not +/// restarted, and every rule in it is a rule about what NOT to delete while Craft is live. These pin +/// those rules against an in-memory backend; proves +/// the same sweep against real tables, where the projection and the partition deletes are real. +/// +public class OrchestratorRetentionTests +{ + private sealed class FakeStore : ICraftTableStore + { + private readonly Dictionary> _tables = new(); + + public Task TryReplaceBatchAsync(string table, string partitionKey, IReadOnlyList rows, + CancellationToken ct = default) => throw new NotSupportedException(); + + public Task PingAsync(CancellationToken ct = default) => Task.CompletedTask; + + public Task EnsureTableAsync(string table, CancellationToken ct = default) + { + if (!_tables.ContainsKey(table)) _tables[table] = new(); + return Task.CompletedTask; + } + + public Task UpsertAsync(string table, StoreRow row, CancellationToken ct = default) + { + _tables[table][(row.PartitionKey, row.RowKey)] = row; + return Task.CompletedTask; + } + + public Task UpsertBatchAsync(string table, string partitionKey, IReadOnlyList rows, + CancellationToken ct = default) + { + foreach (var r in rows) _tables[table][(r.PartitionKey, r.RowKey)] = r; + return Task.CompletedTask; + } + + public Task GetAsync(string table, string partitionKey, string rowKey, CancellationToken ct = default) + => Task.FromResult(_tables[table].TryGetValue((partitionKey, rowKey), out var r) ? r : null); + + public async IAsyncEnumerable QueryPartitionAsync(string table, string partitionKey, + [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default) + { + foreach (var kv in _tables[table].Where(k => k.Key.Item1 == partitionKey).ToList()) + { + yield return kv.Value; + await Task.Yield(); + } + } + + public async IAsyncEnumerable QueryTableAsync(string table, + [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default) + { + foreach (var kv in _tables[table].ToList()) + { + yield return kv.Value; + await Task.Yield(); + } + } + + public Task DeleteAsync(string table, string partitionKey, string rowKey, CancellationToken ct = default) + { + _tables[table].Remove((partitionKey, rowKey)); + return Task.CompletedTask; + } + + public Task DeletePartitionAsync(string table, string partitionKey, CancellationToken ct = default) + { + foreach (var k in _tables[table].Keys.Where(k => k.Item1 == partitionKey).ToList()) + _tables[table].Remove(k); + return Task.CompletedTask; + } + } + + private static readonly TimeSpan Retention = TimeSpan.FromHours(48); + private static readonly HashSet NothingActive = new(StringComparer.Ordinal); + private static DateTime Old => DateTime.UtcNow.AddDays(-3); + private static DateTime Recent => DateTime.UtcNow.AddHours(-1); + + private sealed record Harness(OrchestratorTableStore Store, FakeStore Backing, CraftSettings Settings) + { + public string Runs => Settings.Orchestrator.TablePrefix + "Runs"; + public string Tasks => Settings.Orchestrator.TablePrefix + "Tasks"; + public string Results => Settings.Orchestrator.TablePrefix + "Results"; + + public Task AddRunAsync(string name, string status, DateTime started, DateTime? completed) => + Store.UpsertRunAsync(new OrchestratorRun { Name = name, Status = status, StartedUtc = started, CompletedUtc = completed }); + + public Task AddTasksAsync(string run, int count) => + Store.UpsertTaskBatchAsync(run, Enumerable.Range(1, count) + .Select(i => new OrchestratorTaskItem { Id = $"task-{i}", Status = "Completed" }).ToList()); + + /// A raw row carrying a Timestamp, the way a real backend returns every row. + public Task AddStampedRowAsync(string table, string partition, string rowKey, DateTime stamp) => + Backing.UpsertAsync(table, new StoreRow(partition, rowKey) { Timestamp = new DateTimeOffset(stamp, TimeSpan.Zero) }); + + public async Task CountAsync(string table, string partition) + { + var n = 0; + await foreach (var _ in Backing.QueryPartitionAsync(table, partition)) n++; + return n; + } + } + + private static async Task NewHarnessAsync() + { + var settings = new CraftSettings(); + var backing = new FakeStore(); + var store = new OrchestratorTableStore(NullLogger.Instance, settings, backing); + await store.InitializeAsync(); + return new Harness(store, backing, settings); + } + + [Fact] + public async Task CancelledRun_PastRetention_IsRemovedWithItsPartitions() + { + var h = await NewHarnessAsync(); + await h.AddRunAsync("cancelled", "Cancelled", Old, Old); + await h.AddTasksAsync("cancelled", 3); + await h.Store.StoreResultAsync("cancelled", "task-1", "{\"ok\":true}"); + + var result = await h.Store.CleanupOldRunsAsync(Retention); + + Assert.Equal("cancelled", Assert.Single(result.ExpiredRuns)); + Assert.Null(await h.Store.GetRunAsync("cancelled")); + Assert.Equal(0, await h.CountAsync(h.Tasks, "cancelled")); + Assert.Equal(0, await h.CountAsync(h.Results, "cancelled")); + } + + [Theory] + [InlineData("Completed")] + [InlineData("CompletedWithErrors")] + [InlineData("Failed")] + [InlineData("Cancelled")] + public async Task EveryTerminalStatus_PastRetention_IsRemoved(string status) + { + var h = await NewHarnessAsync(); + await h.AddRunAsync("run", status, Old, Old); + + var result = await h.Store.CleanupOldRunsAsync(Retention); + + Assert.Equal("run", Assert.Single(result.ExpiredRuns)); + Assert.Equal(1, result.RunsExamined); + Assert.Null(await h.Store.GetRunAsync("run")); + } + + [Fact] + public async Task FinishedRun_InsideRetention_IsKept_WithItsRows() + { + var h = await NewHarnessAsync(); + await h.AddRunAsync("recent", "Completed", Old, Recent); + await h.AddTasksAsync("recent", 2); + + var result = await h.Store.CleanupOldRunsAsync(Retention); + + Assert.Empty(result.ExpiredRuns); + Assert.NotNull(await h.Store.GetRunAsync("recent")); + Assert.Equal(2, await h.CountAsync(h.Tasks, "recent")); + } + + [Fact] + public async Task FinishedRun_WithoutCompletedUtc_IsJudgedByStartedUtc() + { + var h = await NewHarnessAsync(); + await h.AddRunAsync("failed-old", "Failed", Old, null); + await h.AddRunAsync("failed-recent", "Failed", Recent, null); + + var result = await h.Store.CleanupOldRunsAsync(Retention); + + Assert.Equal("failed-old", Assert.Single(result.ExpiredRuns)); + Assert.NotNull(await h.Store.GetRunAsync("failed-recent")); + } + + [Fact] + public async Task ActiveRun_IsKept_HoweverLongAgoItStarted() + { + var h = await NewHarnessAsync(); + await h.AddRunAsync("long", "Running", DateTime.UtcNow.AddDays(-10), null); + await h.AddTasksAsync("long", 2); + + var result = await h.Store.CleanupOldRunsAsync(Retention, new HashSet { "long" }); + + Assert.Empty(result.AbandonedRuns); + Assert.NotNull(await h.Store.GetRunAsync("long")); + Assert.Equal(2, await h.CountAsync(h.Tasks, "long")); + } + + [Fact] + public async Task RunNobodyIsDriving_WithNoWritesWithinRetention_IsAbandoned() + { + var h = await NewHarnessAsync(); + await h.AddRunAsync("stuck", "Pending", Old, null); + await h.AddTasksAsync("stuck", 2); + + var result = await h.Store.CleanupOldRunsAsync(Retention, NothingActive); + + Assert.Equal("stuck", Assert.Single(result.AbandonedRuns)); + Assert.Empty(result.ExpiredRuns); + Assert.Null(await h.Store.GetRunAsync("stuck")); + Assert.Equal(0, await h.CountAsync(h.Tasks, "stuck")); + } + + [Fact] + public async Task RunNobodyIsDriving_WithAFreshHeartbeat_IsKept() + { + // Started long ago, but a task completed an hour ago: the counter row is the heartbeat. + var h = await NewHarnessAsync(); + await h.AddRunAsync("slow", "Running", DateTime.UtcNow.AddDays(-10), null); + await h.AddTasksAsync("slow", 2); + await h.AddStampedRowAsync(h.Tasks, "slow", "!!run-counter", Recent); + + var result = await h.Store.CleanupOldRunsAsync(Retention, NothingActive); + + Assert.Empty(result.AbandonedRuns); + Assert.NotNull(await h.Store.GetRunAsync("slow")); + Assert.Equal(3, await h.CountAsync(h.Tasks, "slow")); + } + + [Fact] + public async Task UnknownStatus_IsTreatedAsNotFinished() + { + var h = await NewHarnessAsync(); + await h.AddRunAsync("odd-stale", "Suspended", Old, null); + await h.AddRunAsync("odd-fresh", "Suspended", Old, null); + await h.AddStampedRowAsync(h.Tasks, "odd-fresh", "!!run-counter", Recent); + + var result = await h.Store.CleanupOldRunsAsync(Retention, NothingActive); + + Assert.Equal("odd-stale", Assert.Single(result.AbandonedRuns)); + Assert.Empty(result.ExpiredRuns); + Assert.NotNull(await h.Store.GetRunAsync("odd-fresh")); + } + + [Fact] + public async Task OrphanedPartitions_WithOnlyOldRows_AreRemoved() + { + var h = await NewHarnessAsync(); + await h.AddStampedRowAsync(h.Tasks, "ghost", "task-1", Old); + await h.AddStampedRowAsync(h.Tasks, "ghost", "task-2", Old); + await h.AddStampedRowAsync(h.Results, "ghost", "task-1", Old); + + var result = await h.Store.CleanupOldRunsAsync(Retention); + + Assert.Equal(2, result.OrphanPartitionsRemoved); + Assert.Equal(0, await h.CountAsync(h.Tasks, "ghost")); + Assert.Equal(0, await h.CountAsync(h.Results, "ghost")); + } + + [Fact] + public async Task OrphanedPartition_WithAFreshRow_IsKeptWhole() + { + var h = await NewHarnessAsync(); + await h.AddStampedRowAsync(h.Tasks, "ghost", "task-1", Old); + await h.AddStampedRowAsync(h.Tasks, "ghost", "task-2", Old); + await h.AddStampedRowAsync(h.Tasks, "ghost", "task-3", Recent); + + var result = await h.Store.CleanupOldRunsAsync(Retention); + + Assert.Equal(0, result.OrphanPartitionsRemoved); + Assert.Equal(3, await h.CountAsync(h.Tasks, "ghost")); + } + + [Fact] + public async Task PartitionOfARetainedRun_IsNotAnOrphan_EvenWhenItsRowsAreOld() + { + var h = await NewHarnessAsync(); + await h.AddRunAsync("resumed", "Completed", Old, Recent); + await h.AddStampedRowAsync(h.Tasks, "resumed", "task-1", Old); + await h.AddStampedRowAsync(h.Results, "resumed", "task-1", Old); + + var result = await h.Store.CleanupOldRunsAsync(Retention); + + Assert.Equal(0, result.OrphanPartitionsRemoved); + Assert.Equal(1, await h.CountAsync(h.Tasks, "resumed")); + Assert.Equal(1, await h.CountAsync(h.Results, "resumed")); + } + + [Fact] + public async Task RowsWithoutATimestamp_AreNeverJudgedOrphans() + { + // A backend that cannot say how old a row is gets the benefit of the doubt. + var h = await NewHarnessAsync(); + await h.AddTasksAsync("ghost", 2); + + var result = await h.Store.CleanupOldRunsAsync(Retention); + + Assert.Equal(0, result.OrphanPartitionsRemoved); + Assert.Equal(2, await h.CountAsync(h.Tasks, "ghost")); + } + + [Fact] + public async Task Sweep_ReportsEverythingItExamined() + { + var h = await NewHarnessAsync(); + await h.AddRunAsync("done", "Completed", Old, Old); + await h.AddRunAsync("live", "Running", Recent, null); + await h.AddRunAsync("stuck", "Pending", Old, null); + await h.AddStampedRowAsync(h.Results, "ghost", "r", Old); + + var result = await h.Store.CleanupOldRunsAsync(Retention, new HashSet { "live" }); + + Assert.Equal(3, result.RunsExamined); + Assert.Equal("done", Assert.Single(result.ExpiredRuns)); + Assert.Equal("stuck", Assert.Single(result.AbandonedRuns)); + Assert.Equal(1, result.OrphanPartitionsRemoved); + Assert.NotNull(await h.Store.GetRunAsync("live")); + } +} From a54231fc09ca09397c4b0d6e30b945be972e6379 Mon Sep 17 00:00:00 2001 From: Zacgoose <107489668+Zacgoose@users.noreply.github.com> Date: Sat, 22 Aug 2026 00:34:54 +0800 Subject: [PATCH 2/2] fix(storage): recreate a missing table and retry the operation that found it gone A table deleted out from under a live host - through table maintenance, or by a reset that cleared the orchestrator's state - took every write to it down until the next restart, and silently: the batch path's per-entity fallback failed the same way and swallowed each failure, so status writes simply vanished. AzureTableStore now treats 404 TableNotFound (as opposed to a row's ResourceNotFound) as the signal to create the table and run the operation again: single and batch upserts retry, reads and the partition scan start over against the fresh table, and a conditional claim returns false after bringing the table back since its rows cannot exist any more. EnsureTableAsync shares the create path. On the real service a just-deleted table refuses creation with 409 TableBeingDeleted for about a minute (measured 62 s; Azure documents at least 40 s), and during that window it still accepts and then discards writes, so the create polls every 3 s for up to 2 minutes under the caller's cancellation token. A recreate logs a warning naming the table. AzureTableStoreRecreateTests proves each operation kind against a never-created table, and one test gated on CRAFT_TEST_INCLUDE_SLOW=1 proves the deletion window is ridden out against a real account. --- Services/Storage/AzureTableStore.cs | 181 ++++++++++++++-- docs/configuration.md | 6 + .../AzureTableStoreRecreateTests.cs | 202 ++++++++++++++++++ 3 files changed, 367 insertions(+), 22 deletions(-) create mode 100644 tests/Craft.Tests/AzureTableStoreRecreateTests.cs diff --git a/Services/Storage/AzureTableStore.cs b/Services/Storage/AzureTableStore.cs index e2f430c..a2693e1 100644 --- a/Services/Storage/AzureTableStore.cs +++ b/Services/Storage/AzureTableStore.cs @@ -33,7 +33,9 @@ public sealed class AzureTableStore : ICraftTableStore "PartitionKey", "RowKey", "Timestamp", "odata.etag" }; - public AzureTableStore(CraftSettings settings) + private readonly ILogger? _logger; + + public AzureTableStore(CraftSettings settings, ILogger? logger = null) { // Resolved lazily so constructing the store on a role that never touches storage does not // require a connection string — it is only resolved on first actual use. Prefers the explicit @@ -41,6 +43,7 @@ public AzureTableStore(CraftSettings settings) _connectionString = new Lazy(() => settings.Storage.ResolveConnection(settings.Auth.UserStorageConnection, "table storage")); _clientOptions = BuildClientOptions(settings.Storage); + _logger = logger; } /// @@ -85,11 +88,79 @@ public async Task PingAsync(CancellationToken ct = default) break; } - public Task EnsureTableAsync(string table, CancellationToken ct = default) => - Client(table).CreateIfNotExistsAsync(ct); + /// + /// How long an operation that found its table missing waits for the service to allow the table to + /// be created again. A deleted table is "being deleted" for a while — Azure documents "at least 40 + /// seconds"; measured at 62 s on a real account — and create answers 409 TableBeingDeleted until + /// then. (During that window the service still accepts reads and writes to the doomed table and + /// then discards them with it; nothing a client can detect, so the 404 that follows is the first + /// real signal.) Azurite has no such window. The wait honours the caller's cancellation token, so + /// a status-writer flush that times out simply requeues the write and the next flush resumes. + /// + private static readonly TimeSpan TableRecreateWait = TimeSpan.FromSeconds(120); + private static readonly TimeSpan TableRecreatePoll = TimeSpan.FromSeconds(3); - public Task UpsertAsync(string table, StoreRow row, CancellationToken ct = default) => - Client(table).UpsertEntityAsync(ToEntity(row), TableUpdateMode.Replace, ct); + public Task EnsureTableAsync(string table, CancellationToken ct = default) => CreateTableAsync(table, ct); + + /// + /// The service's answer for a table that does not exist: 404 with error code TableNotFound. A row + /// that does not exist is also a 404, but ResourceNotFound, so the code is what separates "row is + /// gone" (routine) from "the whole table is gone" (deleted out from under a live host). Transactions + /// report it the same way, through . + /// + private static bool IsTableNotFound(RequestFailedException ex) => + ex.Status == 404 && ( + string.Equals(ex.ErrorCode, "TableNotFound", StringComparison.OrdinalIgnoreCase) + || ex.Message.Contains("TableNotFound", StringComparison.OrdinalIgnoreCase)); + + private async Task CreateTableAsync(string table, CancellationToken ct) + { + var client = Client(table); + var deadline = DateTime.UtcNow + TableRecreateWait; + while (true) + { + try + { + await client.CreateIfNotExistsAsync(ct); + return; + } + catch (RequestFailedException ex) when (ex.Status == 409 + && string.Equals(ex.ErrorCode, "TableBeingDeleted", StringComparison.OrdinalIgnoreCase) + && DateTime.UtcNow < deadline) + { + await Task.Delay(TableRecreatePoll, ct); + } + } + } + + /// + /// An operation just learned its table no longer exists. Put the table back so the caller can run + /// the operation again. A table deleted through table maintenance, or by a reset that cleared the + /// orchestrator's state, must not take the host's writes down with it until the next restart — and + /// it did, silently: the batch path's per-entity fallback failed the same way and swallowed it. + /// If the table cannot be created (still being deleted past ), this + /// throws and so does the operation, which is the right outcome for a write that cannot land. + /// + private async Task RecreateTableAsync(string table, CancellationToken ct) + { + await CreateTableAsync(table, ct); + _logger?.LogWarning("[TableStore] Table {Table} was missing and has been recreated; retrying the operation that noticed", table); + } + + public async Task UpsertAsync(string table, StoreRow row, CancellationToken ct = default) + { + var client = Client(table); + var entity = ToEntity(row); + try + { + await client.UpsertEntityAsync(entity, TableUpdateMode.Replace, ct); + } + catch (RequestFailedException ex) when (IsTableNotFound(ex)) + { + await RecreateTableAsync(table, ct); + await client.UpsertEntityAsync(entity, TableUpdateMode.Replace, ct); + } + } public async Task UpsertBatchAsync(string table, string partitionKey, IReadOnlyList rows, CancellationToken ct = default) { @@ -102,7 +173,7 @@ public async Task UpsertBatchAsync(string table, string partitionKey, IReadOnlyL var rowChars = EstimateChars(row); if (batch.Count > 0 && (batch.Count >= MaxBatch || chars + rowChars > MaxBatchChars)) { - await SubmitAsync(client, batch, ct); + await SubmitAsync(table, client, batch, ct); batch.Clear(); chars = 0; } @@ -111,7 +182,7 @@ public async Task UpsertBatchAsync(string table, string partitionKey, IReadOnlyL } if (batch.Count > 0) - await SubmitAsync(client, batch, ct); + await SubmitAsync(table, client, batch, ct); } public async Task TryReplaceBatchAsync(string table, string partitionKey, IReadOnlyList rows, @@ -142,6 +213,13 @@ public async Task TryReplaceBatchAsync(string table, string partitionKey, await Client(table).SubmitTransactionAsync(actions, ct); return true; } + catch (RequestFailedException ex) when (IsTableNotFound(ex)) + { + // No table, so no rows, so nothing was claimed. Put the table back for the writes that + // follow; this claim itself is simply lost — there is nothing left to retry it against. + await RecreateTableAsync(table, ct); + return false; + } catch (RequestFailedException ex) when (ex.Status is 412 or 404 or 409) { // 412 precondition failed / 409 conflict — someone else changed or claimed a row. @@ -151,22 +229,42 @@ public async Task TryReplaceBatchAsync(string table, string partitionKey, } } - private static async Task SubmitAsync(TableClient client, List batch, CancellationToken ct) + private async Task SubmitAsync(string table, TableClient client, List batch, CancellationToken ct) { try { await client.SubmitTransactionAsync(batch, ct); + return; } - catch (Exception) + catch (RequestFailedException ex) when (IsTableNotFound(ex)) { - // A transaction is all-or-nothing; on failure fall back to individual upserts so one bad - // entity (or a transient 4xx) doesn't drop the whole batch. - foreach (var action in batch) + // The table is gone, not the batch. Put it back and submit the same transaction again; only + // if THAT fails does the per-entity fallback below get its turn. Without this the fallback + // ran against the missing table as well and swallowed every one of its failures, so a + // deleted table lost every status write until the next restart without a line in the log. + await RecreateTableAsync(table, ct); + try + { + await client.SubmitTransactionAsync(batch, ct); + return; + } + catch (Exception) { - try { await client.UpsertEntityAsync((TableEntity)action.Entity, TableUpdateMode.Replace, ct); } - catch { /* best-effort fallback */ } + // Fall through to the per-entity fallback. } } + catch (Exception) + { + // Fall through to the per-entity fallback. + } + + // A transaction is all-or-nothing; on failure fall back to individual upserts so one bad + // entity (or a transient 4xx) doesn't drop the whole batch. + foreach (var action in batch) + { + try { await client.UpsertEntityAsync((TableEntity)action.Entity, TableUpdateMode.Replace, ct); } + catch { /* best-effort fallback */ } + } } public async Task GetAsync(string table, string partitionKey, string rowKey, CancellationToken ct = default) @@ -176,24 +274,65 @@ private static async Task SubmitAsync(TableClient client, List(partitionKey, rowKey, cancellationToken: ct); return ToRow(response.Value); } + catch (RequestFailedException ex) when (IsTableNotFound(ex)) + { + // The row cannot exist, which is the same answer as below — but bring the table back now + // rather than leaving that to whichever write comes next. + await RecreateTableAsync(table, ct); + return null; + } catch (RequestFailedException ex) when (ex.Status == 404) { return null; } } + /// + /// Enumerates a query, recreating the table and starting over if the first page reports it missing. + /// The retry enumerates the fresh table — empty, but that is the query's answer, and it proves the + /// table is back. A table that vanishes mid-stream propagates; paging across a delete is not a case + /// worth hiding. + /// + private async IAsyncEnumerable EnumerateAsync(string table, Func> query, + [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default) + { + var recreated = false; + while (true) + { + await using var rows = query().GetAsyncEnumerator(ct); + bool moved; + try + { + moved = await rows.MoveNextAsync(); + } + catch (RequestFailedException ex) when (!recreated && IsTableNotFound(ex)) + { + recreated = true; + await RecreateTableAsync(table, ct); + continue; + } + + while (moved) + { + yield return rows.Current; + moved = await rows.MoveNextAsync(); + } + yield break; + } + } + public async IAsyncEnumerable QueryPartitionAsync(string table, string partitionKey, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default) { var filter = $"PartitionKey eq '{Escape(partitionKey)}'"; - await foreach (var entity in Client(table).QueryAsync(filter: filter, cancellationToken: ct)) + await foreach (var entity in EnumerateAsync(table, () => Client(table).QueryAsync(filter: filter, cancellationToken: ct), ct)) yield return ToRow(entity); } public async IAsyncEnumerable QueryTableAsync(string table, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default) { - await foreach (var entity in Client(table).QueryAsync(cancellationToken: ct)) + await foreach (var entity in EnumerateAsync(table, () => Client(table).QueryAsync(cancellationToken: ct), ct)) yield return ToRow(entity); } @@ -205,7 +344,7 @@ public async IAsyncEnumerable QueryTableAsync(string table, public async IAsyncEnumerable QueryTableAsync(string table, string? filter, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default) { - await foreach (var entity in Client(table).QueryAsync(filter: filter, cancellationToken: ct)) + await foreach (var entity in EnumerateAsync(table, () => Client(table).QueryAsync(filter: filter, cancellationToken: ct), ct)) yield return ToRow(entity); } @@ -217,7 +356,7 @@ public async IAsyncEnumerable QueryTableAsync(string table, string? fi public async IAsyncEnumerable QueryTableAsync(string table, string? filter, IReadOnlyList? properties, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default) { - await foreach (var entity in Client(table).QueryAsync(filter: filter, select: properties, cancellationToken: ct)) + await foreach (var entity in EnumerateAsync(table, () => Client(table).QueryAsync(filter: filter, select: properties, cancellationToken: ct), ct)) yield return ToRow(entity); } @@ -237,10 +376,8 @@ public async Task DeletePartitionAsync(string table, string partitionKey, Cancel { var client = Client(table); var keys = new List<(string pk, string rk)>(); - await foreach (var entity in client.QueryAsync( - filter: $"PartitionKey eq '{Escape(partitionKey)}'", - select: select, - cancellationToken: ct)) + var filter = $"PartitionKey eq '{Escape(partitionKey)}'"; + await foreach (var entity in EnumerateAsync(table, () => client.QueryAsync(filter: filter, select: select, cancellationToken: ct), ct)) { keys.Add((entity.PartitionKey, entity.RowKey)); } diff --git a/docs/configuration.md b/docs/configuration.md index 25f013a..ee195f7 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -422,6 +422,12 @@ Fan-out/fan-in task execution with crash recovery. All three function settings have sensible defaults provided by the `CraftRuntime/` scripts. Most apps only need to set `TablePrefix`. +A table that goes missing while the host is running — deleted through table maintenance, or by a reset +that cleared the orchestrator's state — is recreated by the next read or write that notices it, and that +operation is retried. If the table was only just dropped, the service refuses to recreate it for a while +(Azure documents "at least 40 seconds"; about a minute in practice) and the operation waits that out, up +to two minutes. A batch write into a missing table used to fail silently until the next restart. + **Queuing orchestrator runs from PowerShell:** Call `Start-CraftOrchestrator` (provided in `CraftRuntime/`) to queue a fan-out run: diff --git a/tests/Craft.Tests/AzureTableStoreRecreateTests.cs b/tests/Craft.Tests/AzureTableStoreRecreateTests.cs new file mode 100644 index 0000000..01c9da4 --- /dev/null +++ b/tests/Craft.Tests/AzureTableStoreRecreateTests.cs @@ -0,0 +1,202 @@ +using Azure; +using Azure.Data.Tables; +using Craft.Configuration; +using Craft.Storage; +using Microsoft.Extensions.Logging; + +namespace Craft.Tests; + +/// +/// A table that is missing when the host needs it — never created, or deleted through table maintenance +/// or a reset that cleared the orchestrator's state — must come back on the next operation that notices, +/// and that operation must still do its job. Only the real backend can prove this: the error code that +/// tells a missing table from a missing row is the service's, and the batch path's silent fallback is +/// exactly what a fake never reproduces. +/// +/// The fast tests use a table that was never created, which answers 404 TableNotFound at once on both +/// Azurite and the real service. A table that was just DELETED is different on the real service: for +/// about a minute it still accepts every operation and then discards them with the table, while create +/// answers 409 TableBeingDeleted — so the store can only react once the 404 finally arrives, and must +/// then wait the window out. That costs a minute per test against a real account, so it is covered by +/// one test gated on CRAFT_TEST_INCLUDE_SLOW=1. Azurite by default, a real account via +/// CRAFT_TEST_TABLE_CONNECTION; skipped, not failed, when neither is reachable — a skip looks like a pass. +/// +public class AzureTableStoreRecreateTests +{ + /// Captures the store's log lines; the recreate path announces itself with one. + private sealed class ListLogger(List sink) : ILogger + { + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + lock (sink) sink.Add(formatter(state, exception)); + } + } + + private sealed class Fixture : IAsyncDisposable + { + public required AzureTableStore Store { get; init; } + public required string Table { get; init; } + public required string Connection { get; init; } + public required List Logs { get; init; } + + public bool Recreated + { + get { lock (Logs) return Logs.Any(m => m.Contains("has been recreated", StringComparison.Ordinal)); } + } + + /// Whether the table exists before the test starts. + public static async Task TryConnectAsync(bool create) + { + var settings = new CraftSettings(); + var connection = Environment.GetEnvironmentVariable("CRAFT_TEST_TABLE_CONNECTION"); + if (!string.IsNullOrWhiteSpace(connection)) + settings.Auth.UserStorageConnection = connection; + else + { + settings.Storage.AllowDevelopmentStorage = true; + connection = "UseDevelopmentStorage=true"; + } + + var logs = new List(); + var store = new AzureTableStore(settings, new ListLogger(logs)); + try + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(3)); + await store.PingAsync(cts.Token); + } + catch + { + return null; + } + + var table = "azrc" + Guid.NewGuid().ToString("N")[..8]; + if (create) await store.EnsureTableAsync(table); + return new Fixture { Store = store, Table = table, Connection = connection, Logs = logs }; + } + + /// Delete the table behind the store's back, the way table maintenance would. + public async Task DropTableAsync() => await new TableServiceClient(Connection).DeleteTableAsync(Table); + + public async Task TableExistsAsync() + { + await foreach (var _ in new TableServiceClient(Connection).QueryAsync($"TableName eq '{Table}'")) + return true; + return false; + } + + public async ValueTask DisposeAsync() + { + try { await new TableServiceClient(Connection).DeleteTableAsync(Table); } + catch (RequestFailedException) { /* never created, or already gone */ } + } + } + + private static StoreRow Row(string rowKey) => new("p", rowKey) { Properties = { ["Value"] = rowKey } }; + + private static bool SlowTestsEnabled => + Environment.GetEnvironmentVariable("CRAFT_TEST_INCLUDE_SLOW") == "1"; + + [Fact] + public async Task Upsert_IntoAMissingTable_CreatesIt_AndTheRowLands() + { + await using var fx = await Fixture.TryConnectAsync(create: false); + if (fx == null) return; + + await fx.Store.UpsertAsync(fx.Table, Row("one")); + + Assert.True(fx.Recreated); + var back = await fx.Store.GetAsync(fx.Table, "p", "one"); + Assert.Equal("one", back?.GetString("Value")); + } + + [Fact] + public async Task UpsertBatch_IntoAMissingTable_CreatesIt_AndEveryRowLands() + { + // This path used to lose the batch without a trace: the transaction failed, the per-entity + // fallback failed the same way, and nothing was logged or thrown. + await using var fx = await Fixture.TryConnectAsync(create: false); + if (fx == null) return; + + await fx.Store.UpsertBatchAsync(fx.Table, "p", [Row("a"), Row("b"), Row("c")]); + + Assert.True(fx.Recreated); + var keys = new List(); + await foreach (var row in fx.Store.QueryPartitionAsync(fx.Table, "p")) keys.Add(row.RowKey); + Assert.Equal("a,b,c", string.Join(",", keys.OrderBy(k => k, StringComparer.Ordinal))); + } + + [Fact] + public async Task Query_OnAMissingTable_CreatesIt_AndYieldsNothing() + { + await using var fx = await Fixture.TryConnectAsync(create: false); + if (fx == null) return; + + var count = 0; + await foreach (var _ in fx.Store.QueryTableAsync(fx.Table)) count++; + + Assert.Equal(0, count); + Assert.True(fx.Recreated); + Assert.True(await fx.TableExistsAsync()); + } + + [Fact] + public async Task Get_OnAMissingTable_IsNull_AndCreatesIt() + { + await using var fx = await Fixture.TryConnectAsync(create: false); + if (fx == null) return; + + Assert.Null(await fx.Store.GetAsync(fx.Table, "p", "x")); + Assert.True(fx.Recreated); + Assert.True(await fx.TableExistsAsync()); + } + + [Fact] + public async Task Get_OnAMissingRow_IsStillJustNull() + { + // The routine 404 keeps its routine answer — only the table-level one triggers a create. + await using var fx = await Fixture.TryConnectAsync(create: true); + if (fx == null) return; + + Assert.Null(await fx.Store.GetAsync(fx.Table, "p", "never-written")); + Assert.False(fx.Recreated); + } + + [Fact] + public async Task DeletePartition_OnAMissingTable_DoesNotThrow() + { + await using var fx = await Fixture.TryConnectAsync(create: false); + if (fx == null) return; + + await fx.Store.DeletePartitionAsync(fx.Table, "p"); + + Assert.True(fx.Recreated); + Assert.True(await fx.TableExistsAsync()); + } + + [Fact] + public async Task Upsert_AfterTheTableWasJustDeleted_WaitsOutTheDeletionWindow_AndLands() + { + // Slow on the real service (about a minute): the doomed table keeps answering until the service + // finally reports it gone, and create is refused with TableBeingDeleted until the window ends. + // The store has to ride out both. Azurite has no window, so there the first write recreates. + if (!SlowTestsEnabled) return; + await using var fx = await Fixture.TryConnectAsync(create: true); + if (fx == null) return; + await fx.DropTableAsync(); + + // Writes the service accepts into the doomed table are lost with it — the service's behaviour, + // and why this keeps writing until the store reports that it saw the 404 and recreated the table. + var deadline = DateTime.UtcNow.AddMinutes(3); + while (!fx.Recreated && DateTime.UtcNow < deadline) + { + await fx.Store.UpsertAsync(fx.Table, Row("survivor")); + if (!fx.Recreated) await Task.Delay(TimeSpan.FromSeconds(2)); + } + + Assert.True(fx.Recreated, "The store never saw the table go missing within three minutes of its deletion."); + Assert.Equal("survivor", (await fx.Store.GetAsync(fx.Table, "p", "survivor"))?.GetString("Value")); + } +}