From b54577dd1096e0a167e01637d2d6bf718deecd3c Mon Sep 17 00:00:00 2001 From: Zacgoose <107489668+Zacgoose@users.noreply.github.com> Date: Tue, 18 Aug 2026 19:30:32 +0800 Subject: [PATCH 1/3] fix(powershell): dispose worker runspaces on recycle and shutdown PowerShell.Create(iss) assigns the runspace rather than creating it lazily, and an assigned runspace is caller-owned: PowerShell.Dispose() does not close it. Every recycled worker therefore left its runspace open, and with ReuseRunspaceThread each open runspace keeps a dedicated pipeline thread alive, rooting the entire session state (every SSFE-injected function of every module) through any GC, however aggressive. Measured in production: ~20 MB retained per recycled worker at RecycleAfterInvocations=1000, i.e. ~20 KB of apparently-leaked heap per invocation, ~2 GB after 95 recycles over 3.5 days, growing until the platform kills the container. Dispose now captures the runspace, disposes the PowerShell object, then disposes the runspace, which closes it and reclaims the pipeline thread. The recycle path, pool shutdown, and the throwaway base worker all funnel through Worker.Dispose(), so one change covers all three. --- Services/PowerShellHost/PowerShellWorker.cs | 12 ++++- .../WorkerRunspaceDisposalTests.cs | 47 +++++++++++++++++++ 2 files changed, 58 insertions(+), 1 deletion(-) create mode 100644 tests/Craft.Tests/WorkerRunspaceDisposalTests.cs diff --git a/Services/PowerShellHost/PowerShellWorker.cs b/Services/PowerShellHost/PowerShellWorker.cs index 4f68087..22b8b57 100644 --- a/Services/PowerShellHost/PowerShellWorker.cs +++ b/Services/PowerShellHost/PowerShellWorker.cs @@ -49,6 +49,9 @@ public PowerShellWorker(int id, InitialSessionState iss, ILogger logger) _pwsh.Runspace.Name = $"Worker{id}"; } + /// This worker's runspace. Test-only access — production code goes through _pwsh. + internal Runspace Runspace => _pwsh.Runspace; + public void Initialize(ScriptRepository repo, string apiBasePath, CraftSettings settings) { if (_initialized) return; @@ -582,8 +585,15 @@ public ExportedModuleState ExportModuleState() public void Dispose() { + // PowerShell.Create(iss) ASSIGNS the runspace rather than creating it lazily, and an assigned + // runspace is caller-owned — _pwsh.Dispose() does not close it. Left open, the runspace keeps + // its ReuseThread pipeline thread alive, and a live thread roots the entire session state + // (every SSFE-injected function of every module) through any GC, however aggressive: measured + // at ~20 MB retained per recycled worker, for the process lifetime. + var runspace = _pwsh.Runspace; _pwsh.Dispose(); - // Nothing here owns unmanaged resources directly, but suppressing finalization keeps a + runspace?.Dispose(); + // Nothing else here owns unmanaged resources directly, but suppressing finalization keeps a // derived type that adds a finalizer from having to re-implement IDisposable to do it. GC.SuppressFinalize(this); } diff --git a/tests/Craft.Tests/WorkerRunspaceDisposalTests.cs b/tests/Craft.Tests/WorkerRunspaceDisposalTests.cs new file mode 100644 index 0000000..ba3bc08 --- /dev/null +++ b/tests/Craft.Tests/WorkerRunspaceDisposalTests.cs @@ -0,0 +1,47 @@ +using System.Management.Automation.Runspaces; +using Craft.PowerShellHost; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Craft.Tests; + +/// +/// Disposing a worker must dispose its runspace. PowerShell.Create(iss) ASSIGNS the runspace +/// (caller-owned) rather than creating it lazily, so PowerShell.Dispose() deliberately leaves it +/// open — and an open runspace with ReuseThread keeps a dedicated pipeline thread alive, which +/// roots the entire session state (every SSFE-injected function of every module) through any GC. +/// Measured live: ~20 MB retained per recycled worker, ~2 GB after 95 recycles, indistinguishable from +/// a managed-heap leak because that is exactly what it is. +/// +/// This is the only place the invariant is checked. Nothing functional breaks when the runspace +/// outlives the worker — the replacement worker works fine — so a refactor of Dispose can silently +/// reintroduce the leak without failing anything else. +/// +/// +public class WorkerRunspaceDisposalTests +{ + [Fact] + public void DisposeClosesTheRunspace() + { + var worker = new PowerShellWorker(1, InitialSessionState.CreateDefault2(), NullLogger.Instance); + var runspace = worker.Runspace; + if (runspace.RunspaceStateInfo.State == RunspaceState.BeforeOpen) + runspace.Open(); + + worker.Dispose(); + + Assert.Equal(RunspaceState.Closed, runspace.RunspaceStateInfo.State); + } + + [Fact] + public void DisposeOfANeverOpenedWorkerStillTearsTheRunspaceDown() + { + // The base worker used for ISS cloning is created and disposed without ever running a + // pipeline; its runspace must not survive either. + var worker = new PowerShellWorker(2, InitialSessionState.CreateDefault2(), NullLogger.Instance); + var runspace = worker.Runspace; + + worker.Dispose(); + + Assert.NotEqual(RunspaceState.Opened, runspace.RunspaceStateInfo.State); + } +} From a8fd2bae1ac3f3b4a1540495975b3f1511604ecc Mon Sep 17 00:00:00 2001 From: Zacgoose <107489668+Zacgoose@users.noreply.github.com> Date: Tue, 18 Aug 2026 19:30:46 +0800 Subject: [PATCH 2/3] fix(orchestration): unstick runs that could defer, re-drive, or leak forever Four ways a run could stay in _activeRuns (pinning its whole task graph in memory) for the process lifetime, plus the residue those paths leave behind: - A lost decrement was permanent. DecrementRemainingAsync exhausting its optimistic retries returned null, the batch writer ignored it, and nothing ever reconciled Remaining against the task rows, so a fully-terminal run deferred finalize on every 60s tick forever ("complete in memory but storage shows N outstanding"). The store can now recount the partition (ReconcileRemainingAsync), the batch writer invokes it whenever a decrement is lost, and CheckRunCompletion recounts after 3 consecutive deferrals of a fully-terminal run and finalizes when the recount says done. - The re-drive loop was infinite for a task storage keeps rejecting. Each pass reset the deferral counter and never incremented AttemptCount, so the fail-after-3 rule could never fire. Consecutive re-queue failures are now counted (cleared on success) and the task fails terminally after 5; an exhausted deferral cycle now counts as one attempt and the task fails terminally after 3 cycles. Terminal failure flows through the status writer, so the counter decrements and the run can finish without the poison task. - StartOrResumeRun fell through after finalizing a resumed run and re-created it under the same name, which the finalize''s own post-execution then deleted via CleanupRunAsync, wiping the new run''s rows mid-flight. Finalize now returns; the next scheduler tick starts the fresh outing cleanly. - The status-timer ContainsKey/TryAdd race leaked the losing Timer undisposed. An active periodic Timer is rooted by the runtime timer queue, so it fired every 60s and pinned the run graph through its closure for the process lifetime. The loser is now disposed. Run names are also sanitized through TableKeys at dispatch (bridge and StartFromBatch). Names built from user-typed task names ("Alert on Entra ID P1/P2 license over-utilization") put table-illegal characters into PartitionKeys, which 400s every write for the run identically forever: the run could neither start nor be re-driven, and the scheduled task behind it silently never ran. Finalize now also sweeps the per-task deferral and re-queue tracking entries, which otherwise outlive their run. --- Services/Bridges/OrchestratorBridge.cs | 6 + Services/Orchestration/OrchestratorService.cs | 120 +++++++++++++++++- Services/Storage/OrchestratorTableStore.cs | 47 ++++++- tests/Craft.Tests/RunRemainingCounterTests.cs | 69 ++++++++++ 4 files changed, 236 insertions(+), 6 deletions(-) diff --git a/Services/Bridges/OrchestratorBridge.cs b/Services/Bridges/OrchestratorBridge.cs index cd41cfe..322faa2 100644 --- a/Services/Bridges/OrchestratorBridge.cs +++ b/Services/Bridges/OrchestratorBridge.cs @@ -1,6 +1,7 @@ using System.Collections.Concurrent; using Craft.Hosting; using Craft.Orchestration; +using Craft.Storage; // NAMESPACE PINNED — do not change. // Downstream PowerShell reaches these types by fully-qualified name, e.g. @@ -26,6 +27,10 @@ public static void QueueOrchestration(string name, string batchJson, int priorit string? postExecFunctionName = null, string? postExecParametersJson = null, string? reference = null) { + // Sanitized here as well as at run creation so the child-run registration in DrainPending + // records the SAME name the service ends up creating — a raw name with a table-illegal + // character would register a child link no live run ever matches. + name = TableKeys.Sanitize(name); var parentRunName = OperationContext.Current?.RunName; s_pending.Enqueue(new PendingOrchestration(name, batchJson, priority, postExecFunctionName, postExecParametersJson, parentRunName, reference)); @@ -46,6 +51,7 @@ public static void QueueOrchestrationFromFile(string name, string batchFilePath, string? postExecFunctionName = null, string? postExecParametersJson = null, string? reference = null) { + name = TableKeys.Sanitize(name); var parentRunName = OperationContext.Current?.RunName; s_pending.Enqueue(new PendingOrchestration(name, string.Empty, priority, postExecFunctionName, postExecParametersJson, parentRunName, reference, batchFilePath)); diff --git a/Services/Orchestration/OrchestratorService.cs b/Services/Orchestration/OrchestratorService.cs index 01d28b9..c8c4a8b 100644 --- a/Services/Orchestration/OrchestratorService.cs +++ b/Services/Orchestration/OrchestratorService.cs @@ -287,8 +287,13 @@ public async Task StartOrResumeRun(string name, string plannerPath, string taskP return; } - // All tasks finished — finalize + // All tasks finished — finalize, and STOP. Finalize dispatches post-execution + // asynchronously, and its success path deletes the run's partitions by name + // (CleanupRunAsync). Falling through to start a fresh same-named outing here raced + // that delete and lost the new run's rows mid-flight; the next scheduler tick starts + // the fresh outing cleanly instead. await FinalizeRunAsync(run); + return; } // Start a new run @@ -569,6 +574,14 @@ private async Task StartFromBatchCoreAsync(string name, string batchJson, int pr string? postExecFunctionName, string? postExecParametersJson, string? parentRunName, string? reference, string? batchFilePath, CancellationToken ct) { + // Run names become PartitionKeys verbatim, and batch names carry user-typed task names + // ("Alert on Entra ID P1/P2 …"). An illegal key character 400s every write for the run — + // run row, task rows, counter, queue rows — identically forever, so the run can neither + // start nor be re-driven. Sanitize at the boundary, like task ids at mint. + name = TableKeys.Sanitize(name); + if (!string.IsNullOrEmpty(parentRunName)) + parentRunName = TableKeys.Sanitize(parentRunName); + // A run cannot be its own parent. The ambient RunName rides along when a run is re-queued // from inside its own context; persisting it would feed the reattach loop a self-link on the // next start and show circular lineage in the status APIs. @@ -696,7 +709,14 @@ private async Task DispatchPendingTasksAsync(OrchestratorRun run, string taskPat lock (_lock) { CheckRunCompletion(run); } }, null, TimeSpan.FromSeconds(60), TimeSpan.FromSeconds(60)); - _runStatusTimers.TryAdd(run.Name, timer); + if (!_runStatusTimers.TryAdd(run.Name, timer)) + { + // Lost the ContainsKey→TryAdd race (concurrent dispatch of the same run — startup + // resume vs a scheduler tick). An active periodic Timer is rooted by the runtime's + // timer queue, so an undisposed loser would fire — and pin this run graph through its + // closure — for the process lifetime. + timer.Dispose(); + } } var pending = run.Tasks.Where(t => t.Status == "Pending").ToList(); @@ -766,15 +786,55 @@ private void RequeueToTable(OrchestratorRun run, OrchestratorTaskItem task) var priority = task.Priority ?? run.Priority; _ = Task.Run(async () => { - try { await _queue.EnqueueAsync(run.Name, task.Id, priority, DateTime.UtcNow); } + var key = DeferralKey(run.Name, task.Id); + try + { + await _queue.EnqueueAsync(run.Name, task.Id, priority, DateTime.UtcNow); + _requeueFailures.TryRemove(key, out _); + } catch (Exception ex) { - _logger.LogWarning(ex, "[Scheduler] Could not re-queue {Task} in {Run} — the re-drive will retry", - task.Id, run.Name); + // A row storage rejects is rejected identically forever (illegal key remnant, + // oversized property), and the re-drive resets the deferral counter on every pass — + // without this cap the retry loop is infinite and the run it belongs to can never + // finalize. Consecutive failures only: a success above clears the count. + var failures = _requeueFailures.AddOrUpdate(key, 1, (_, c) => c + 1); + if (failures >= MaxRequeueFailures) + { + _requeueFailures.TryRemove(key, out _); + FailTaskTerminally(run, task, + $"Could not re-queue after {failures} consecutive attempts: {ex.Message}"); + return; + } + _logger.LogWarning(ex, + "[Scheduler] Could not re-queue {Task} in {Run} (attempt {Count}/{Max}) — the re-drive will retry", + task.Id, run.Name, failures, MaxRequeueFailures); } }); } + /// + /// Move a task that can never run to Failed and let its run finish without it. The terminal write + /// flows through the status writer like any other completion, so the remaining counter decrements + /// and finalize proceeds — the alternative is a Pending task retried for the process lifetime, + /// pinning the whole run graph with it. + /// + private void FailTaskTerminally(OrchestratorRun run, OrchestratorTaskItem task, string reason) + { + lock (_lock) + { + if (task.Status is "Completed" or "Failed" or "Cancelled") return; + task.Status = "Failed"; + task.LastError = reason; + task.CompletedUtc = DateTime.UtcNow; + task.Parameters = null!; + CheckRunCompletion(run); + } + PersistTaskAndRunAsync(run, task); + _logger.LogError("[Scheduler] Task {TaskId} in {Run} permanently failed: {Reason}", + task.Id, run.Name, reason); + } + /// /// Rebuild the work for a queued task. Registered on the JobManager at startup. /// @@ -1013,6 +1073,18 @@ private sealed record DeferralState(int Count, DateTime LastUtc); /// Cap on in-process retries before a task is left for the next recovery pass to pick up. private const int MaxDeferrals = 3; + /// Consecutive finalize checks where storage still reported outstanding work for a run whose + /// in-memory tasks are all terminal. At the counter is recounted + /// from the task rows — a lost decrement otherwise defers finalize forever. + private readonly ConcurrentDictionary _finalizeDeferrals = new(); + private const int ReconcileAfterDeferrals = 3; + + /// Consecutive re-queue failures per task. Storage rejecting the same entity is not + /// transient — the write fails identically forever (see ) — so past + /// the task is failed terminally instead of re-driven again. + private readonly ConcurrentDictionary _requeueFailures = new(); + private const int MaxRequeueFailures = 5; + /// /// Re-queue a task whose durable marker could not be written, so it retries once storage recovers /// instead of waiting for a restart. Bounded: after the task is simply @@ -1029,6 +1101,18 @@ private void DeferTask(OrchestratorRun run, OrchestratorTaskItem task, Exception if (count > MaxDeferrals) { + // One exhausted deferral cycle counts as one attempt on the task, mirroring startup + // recovery's 3-attempts rule. The re-drive resets the deferral counter when it re-queues, + // so without this the marker-fail → re-queue → marker-fail cycle repeats for the process + // lifetime and the run never finalizes. Exactly-once per cycle: only the call that + // crosses the cap increments (a duplicate queue row can push count past it again). + if (count == MaxDeferrals + 1 && ++task.AttemptCount >= 3) + { + FailTaskTerminally(run, task, + $"Durable Running marker rejected across {task.AttemptCount} deferral cycles: {cause.Message}"); + return; + } + // Left Pending on purpose — storage already says Pending, so nothing is lost. It is no longer // terminal though: RedrivePendingTasks picks it up once it has aged, so recovery is not // gated on a restart the way it used to be. @@ -1179,12 +1263,28 @@ private void CheckRunCompletion(OrchestratorRun run) var remaining = await _store.GetRemainingAsync(run.Name); if (remaining is > 0) { + // A counter that keeps contradicting a fully-terminal graph is drifted, not + // busy — a decrement that exhausted its retries is never re-applied, and + // without a recount this deferral repeats on every 60s tick for the process + // lifetime, pinning the run graph with it. Give in-flight terminal writes a + // few checks to land, then recount the partition the counter summarizes. + var misses = _finalizeDeferrals.AddOrUpdate(run.Name, 1, (_, c) => c + 1); + if (misses >= ReconcileAfterDeferrals) + { + _finalizeDeferrals.TryRemove(run.Name, out _); + if (await _store.ReconcileRemainingAsync(run.Name) is 0) + { + await FinalizeRunAsync(run); + return; + } + } _logger.LogInformation( "[Scheduler] Run {Name} complete in memory but storage shows {Remaining} outstanding - deferring finalize", run.Name, remaining); return; } + _finalizeDeferrals.TryRemove(run.Name, out _); await FinalizeRunAsync(run); } catch (Exception ex) { _logger.LogError(ex, "[Scheduler] FinalizeRun failed for {Name}", run.Name); } @@ -1284,6 +1384,16 @@ private async Task FinalizeRunCoreAsync(OrchestratorRun run) _taskScriptPaths.TryRemove(run.Name, out _); _runStatusTimers.TryRemove(run.Name, out var timer); timer?.Dispose(); + _finalizeDeferrals.TryRemove(run.Name, out _); + // Deferral and re-queue tracking is keyed per task and nothing else removes entries for tasks + // that ended without passing through their happy-path cleanup — without this sweep the residue + // of every run that ever deferred outlives the run. + foreach (var t in run.Tasks) + { + var key = DeferralKey(run.Name, t.Id); + _deferrals.TryRemove(key, out _); + _requeueFailures.TryRemove(key, out _); + } var wallDisplay = wallClock.TotalSeconds < 60 ? $"{wallClock.TotalSeconds:F1}s" diff --git a/Services/Storage/OrchestratorTableStore.cs b/Services/Storage/OrchestratorTableStore.cs index 8a083bc..2a368ea 100644 --- a/Services/Storage/OrchestratorTableStore.cs +++ b/Services/Storage/OrchestratorTableStore.cs @@ -289,6 +289,45 @@ public Task InitRemainingAsync(string runName, int total, CancellationToken ct = return null; } + /// + /// Recount Remaining from the task rows the counter summarizes, and repair the counter row + /// when they disagree. + /// + /// A decrement that exhausts its retries is never re-applied — the terminal task rows landed but + /// the counter kept its old value, and from then on it permanently overstates the outstanding work + /// and finalize defers forever. The scan is the whole-partition read the counter exists to avoid, + /// which is why this runs only when a caller has evidence of drift (a lost decrement, a finalize + /// deferred repeatedly), never on the hot path. + /// + /// The reconciled outstanding count, or null if the run has no counter row or a concurrent + /// writer moved the counter mid-recount — the caller's next pass re-reads either way. + public async Task ReconcileRemainingAsync(string runName, CancellationToken ct = default) + { + var counter = await _store.GetAsync(_tasksTable, runName, CounterRowKey, ct); + if (counter == null) return null; + + var outstanding = 0; + await foreach (var row in _store.QueryPartitionAsync(_tasksTable, runName, ct)) + { + if (row.RowKey == CounterRowKey) continue; + if (!IsTerminal(row.GetString("Status"))) outstanding++; + } + + var stored = counter.GetInt32("Remaining") ?? 0; + if (stored == outstanding) return outstanding; + + // ETag-guarded: a decrement landing between the read above and this write rejects the + // replace, so a recount can never overwrite fresher progress with a stale count. + counter["Remaining"] = outstanding; + if (!await _store.TryReplaceBatchAsync(_tasksTable, runName, [counter], ct)) + return null; + + _logger.LogWarning( + "[OrchestratorStore] Reconciled remaining for {Run}: counter said {Stored}, task rows say {Actual}", + runName, stored, outstanding); + return outstanding; + } + /// /// Mark one task terminal and decrement its run's outstanding count, atomically. /// @@ -425,7 +464,13 @@ public async Task> WriteTaskStatusBatchAsync(IReadOnlyList // that knows a terminal write actually applied, and the writer never re-sends a group // that did, which is what keeps the count honest. var terminal = group.Count(w => IsTerminal(w.Status)); - if (terminal > 0) await DecrementRemainingAsync(group.Key, terminal, ct); + if (terminal > 0 && await DecrementRemainingAsync(group.Key, terminal, ct) == null) + { + // Retry exhaustion here loses the decrement for good — the terminal rows above + // landed, so the writer will never re-send this group. Recount now rather than + // letting the counter overstate the run's outstanding work forever. + await ReconcileRemainingAsync(group.Key, ct); + } } catch (Exception ex) { diff --git a/tests/Craft.Tests/RunRemainingCounterTests.cs b/tests/Craft.Tests/RunRemainingCounterTests.cs index 9b1cc7f..c2c7763 100644 --- a/tests/Craft.Tests/RunRemainingCounterTests.cs +++ b/tests/Craft.Tests/RunRemainingCounterTests.cs @@ -287,6 +287,75 @@ public async Task MissingCounterReportsNullRatherThanGuessing() Assert.Null(await store.CompleteTaskAsync("never-seeded", Task_("task-0"))); } + // ─── Reconciliation (the lost-decrement repair) ─── + + /// + /// A decrement that exhausts its retries is never re-sent — the terminal rows landed, the counter + /// didn't move, and from then on it overstates the run's outstanding work forever. Production + /// symptom: "complete in memory but storage shows N outstanding - deferring finalize" on every 60s + /// tick for the life of the process. Reconcile recounts the rows and repairs the counter. + /// + [Fact] + public async Task ReconcileRepairsALostDecrement() + { + var backing = new ConditionalStore(); + var store = await SeededAsync(backing, 3); + + // Terminal rows written WITHOUT the counter moving — exactly what a lost decrement leaves. + await store.UpsertTaskAsync(Run, Task_("task-0")); + await store.UpsertTaskAsync(Run, Task_("task-1", "Failed")); + Assert.Equal(3, await store.GetRemainingAsync(Run)); + + Assert.Equal(1, await store.ReconcileRemainingAsync(Run)); + Assert.Equal(1, await store.GetRemainingAsync(Run)); + } + + [Fact] + public async Task ReconcileWithoutDriftChangesNothing() + { + var backing = new ConditionalStore(); + var store = await SeededAsync(backing, 2); + + var writesBefore = backing.ConditionalWrites; + + Assert.Equal(2, await store.ReconcileRemainingAsync(Run)); + Assert.Equal(writesBefore, backing.ConditionalWrites); + } + + [Fact] + public async Task ReconcileWithoutACounterRowIsANoOp() + { + var (store, _) = NewStore(); + await store.InitializeAsync(); + + Assert.Null(await store.ReconcileRemainingAsync("never-seeded")); + } + + /// + /// A decrement landing between reconcile's recount and its write must win. The recount is stale the + /// moment a competitor moves the counter, so the ETag guard has to reject the repair — reporting + /// null sends the caller back around rather than letting an old count overwrite fresh progress. + /// + [Fact] + public async Task ReconcileLosingARaceDoesNotClobberTheCompetitor() + { + var backing = new ConditionalStore(); + var store = await SeededAsync(backing, 3); + + // Manufacture drift so reconcile attempts a write at all. + await store.UpsertTaskAsync(Run, Task_("task-0")); + + backing.OnBeforeConditionalWrite = () => + { + backing.OnBeforeConditionalWrite = null; + store.CompleteTaskAsync(Run, Task_("task-1")).GetAwaiter().GetResult(); + }; + + Assert.Null(await store.ReconcileRemainingAsync(Run)); + // The competitor's decrement survived: 3 seeded − 1 completed-by-competitor. + Assert.Equal(2, await store.GetRemainingAsync(Run)); + } + // ─── Status-guarded cancel (the cancel-a-run write) ─── [Fact] From d3f57338ab943961b2f9678725c0aa3849177af3 Mon Sep 17 00:00:00 2001 From: Zacgoose <107489668+Zacgoose@users.noreply.github.com> Date: Tue, 18 Aug 2026 19:51:07 +0800 Subject: [PATCH 3/3] fix(orchestration): inherit nested job priority Propagate ambient run priority through OperationContext so nested orchestrator starts and post-exec jobs keep the parent run's priority instead of falling back to the default band. This also adds an explicit-priority QueueBridge overload while preserving the legacy P5 default, updates persisted orchestrator defaults to P4 to match live enqueue behavior, and adds coverage for descriptor jobs, closure jobs, reprioritization, and QueueBridge compatibility. --- .../CraftRuntime/Start-CraftOrchestrator.ps1 | 26 +++- Services/Bridges/QueueBridge.cs | 17 +- Services/Hosting/OperationContext.cs | 9 ++ Services/Orchestration/JobManager.cs | 27 +++- Services/Orchestration/OrchestratorRun.cs | 4 +- Services/Orchestration/OrchestratorService.cs | 3 + .../PowerShellHost/PowerShellRunnerService.cs | 8 +- Services/Storage/OrchestratorTableStore.cs | 2 +- .../OperationContextPriorityTests.cs | 147 ++++++++++++++++++ 9 files changed, 224 insertions(+), 19 deletions(-) create mode 100644 tests/Craft.Tests/OperationContextPriorityTests.cs diff --git a/Runtime/CraftRuntime/Start-CraftOrchestrator.ps1 b/Runtime/CraftRuntime/Start-CraftOrchestrator.ps1 index 93ee3d2..f6310cb 100644 --- a/Runtime/CraftRuntime/Start-CraftOrchestrator.ps1 +++ b/Runtime/CraftRuntime/Start-CraftOrchestrator.ps1 @@ -6,14 +6,17 @@ function Start-CraftOrchestrator { Generic bridge function that serializes a batch of tasks and queues them for fan-out execution via the C# OrchestratorService. - This is the framework-provided default. Application-specific wrappers - (e.g. Start-CIPPOrchestrator) can override via Orchestrator.BridgeFunction - in appsettings.json if they need additional logic (queue routing, dual-boot, etc.) + This is the framework-provided default. Applications with additional logic + (queue routing, dual-boot, etc.) ship their own wrapper (e.g. Start-CIPPOrchestrator) + and simply call that instead of this function. .PARAMETER InputObject Orchestrator input with the following structure: - OrchestratorName (string) — unique run identifier - Batch (array) — task objects, each with at least FunctionName + - Priority (int) — optional; queue priority bucket for the run (lower = sooner). + Defaults to the parent run's priority when queued from inside + an orchestrator run, else 4. - QueueFunction (object) — optional; called first to generate the batch dynamically - FunctionName (string) — Push-{FunctionName} is called - Parameters (object) — passed as -Item to the queue function @@ -97,11 +100,24 @@ function Start-CraftOrchestrator { throw } - Write-Information "Craft: Queuing orchestrator '$OrchestratorName' ($TaskCount tasks$(if ($PostExecFunctionName) { ", PostExec: $PostExecFunctionName" }))" + # Priority resolution: explicit on the InputObject wins; otherwise inherit the enclosing run's + # priority (ambient, set by JobManager for orchestrator activities and post-exec jobs); otherwise + # the default band. Guard the explicit value — the store clamps into 0-99 buckets, so a stray + # negative would silently land in the critical P00 bucket. + $Priority = $InputObject.Priority + if ($null -ne $Priority) { + $Priority = [int]$Priority + if ($Priority -lt 0 -or $Priority -gt 99) { $Priority = $null } + } + if ($null -eq $Priority) { + $Priority = [Craft.Hosting.OperationContext]::Current.Priority ?? 4 + } + + Write-Information "Craft: Queuing orchestrator '$OrchestratorName' ($TaskCount tasks, P$Priority$(if ($PostExecFunctionName) { ", PostExec: $PostExecFunctionName" }))" [Craft.Services.OrchestratorBridge]::QueueOrchestrationFromFile( $OrchestratorName, $BatchPath, - 4, + $Priority, $PostExecFunctionName, $PostExecParametersJson, $InputObject.Reference diff --git a/Services/Bridges/QueueBridge.cs b/Services/Bridges/QueueBridge.cs index 65f3729..0a05db1 100644 --- a/Services/Bridges/QueueBridge.cs +++ b/Services/Bridges/QueueBridge.cs @@ -28,11 +28,22 @@ public static void Initialize(PowerShellRunnerService runner, JobManager jobMana s_queueTaskFunction = queueTaskFunction; } + /// Default-priority enqueue. Kept for compatibility with callers that predate priorities. public static void Enqueue(string cmdlet, string parametersJson) + => Enqueue(cmdlet, parametersJson, DefaultPriority); + + /// + /// Enqueue with an explicit job priority. User-initiated starters (run-now scheduled tasks) pass a + /// high band here so they are not claimed behind the background fan-out backlog — the queue claims + /// strictly by priority bucket, so a starter below the backlog's band cannot run until it drains. + /// + public static void Enqueue(string cmdlet, string parametersJson, int priority) { - s_pending.Enqueue(new PendingQueueCommand(cmdlet, parametersJson)); + s_pending.Enqueue(new PendingQueueCommand(cmdlet, parametersJson, priority)); } + private const int DefaultPriority = 5; + public static void DrainPending() { if (string.IsNullOrEmpty(s_queueTaskFunction)) return; @@ -46,7 +57,7 @@ public static void DrainPending() var captured = cmd; s_jobManager.Enqueue( name: $"Queue-{captured.Cmdlet}", - priority: 5, + priority: captured.Priority, runName: $"Queue-{captured.Cmdlet}-{Guid.NewGuid():N}", id: $"Queue-{Guid.NewGuid():N}", work: async (ct) => @@ -65,5 +76,5 @@ public static void DrainPending() } } - public record PendingQueueCommand(string Cmdlet, string ParametersJson); + public record PendingQueueCommand(string Cmdlet, string ParametersJson, int Priority = DefaultPriority); } diff --git a/Services/Hosting/OperationContext.cs b/Services/Hosting/OperationContext.cs index c88d2b4..4336f60 100644 --- a/Services/Hosting/OperationContext.cs +++ b/Services/Hosting/OperationContext.cs @@ -52,6 +52,15 @@ public sealed class Invocation /// Parent orchestrator run name (null for HTTP requests). public string? RunName { get; init; } + /// + /// Queue priority of the enclosing run, exposed so nested enqueues can inherit it + /// (Start-CIPPOrchestrator reads this to default a child run to its parent's priority). + /// Set only for orchestrator activity jobs and post-execution jobs — plain closure jobs + /// (scheduler starters, queue starters) deliberately leave it null, because their own job + /// priority orders the starter script, not the work it goes on to enqueue. + /// + public int? Priority { get; init; } + /// Category: "HTTP", "Job", "Planner". public string Category { get; init; } = "Job"; diff --git a/Services/Orchestration/JobManager.cs b/Services/Orchestration/JobManager.cs index 1f96cfb..2c19503 100644 --- a/Services/Orchestration/JobManager.cs +++ b/Services/Orchestration/JobManager.cs @@ -16,10 +16,11 @@ namespace Craft.Orchestration; /// - Old completed jobs are cleaned up every 5 minutes /// /// Priority levels (lower = higher priority, callers can use any int): -/// 0-1 = Critical (system cleanup, user tasks) -/// 2-3 = High (audit logs, webhooks) -/// 4-5 = Normal (standards, drift, cache) -/// 6+ = Low (alerts, DB cache, tests, extensions) +/// 0-1 = Critical (reserved: system cleanup, emergencies) +/// 2 = User-initiated (HTTP-triggered fan-outs, user scheduled tasks, run-now starters) +/// 3 = Elevated background (baseline runs) +/// 4-5 = Normal (background fan-out default; non-HTTP queue starters) +/// 6+ = Low (alerts, tests, extensions) /// /// How priority dispatch works: /// The dispatch loop waits for both an item AND a concurrency slot. @@ -109,8 +110,12 @@ public JobManager(ILogger logger, CraftSettings settings, Background /// Async work function. Receives a CancellationToken for shutdown. /// Optional run group name (e.g. "CIPPDBCacheRun") for grouping in status APIs. /// Optional explicit job ID. Auto-generated if null. + /// Priority that work THIS JOB ENQUEUES should inherit, exposed to the + /// job via . Distinct from : + /// a starter script's own queue priority orders the starter, not its fan-out. Only post-execution jobs + /// pass this (the run's priority); leave null everywhere else. public string Enqueue(string name, int priority, Func work, - string? runName = null, string? id = null) + string? runName = null, string? id = null, int? inheritPriority = null) { var jobId = id ?? $"{name}_{Guid.NewGuid():N}"; var record = new JobRecord @@ -128,7 +133,7 @@ public string Enqueue(string name, int priority, Func w lock (_queueLock) { - _pendingQueue.Enqueue(new QueuedJob(record, work, null), priority); + _pendingQueue.Enqueue(new QueuedJob(record, work, null) { InheritPriority = inheritPriority }, priority); } _itemAvailable.Release(); @@ -338,10 +343,15 @@ private async Task RunJobAsync(QueuedJob job, CancellationToken ct) try { - // Set operation context for traceability — ExecuteScript reads RunName from this + // Set operation context for traceability — ExecuteScript reads RunName from this. + // Priority is the value nested enqueues should inherit: for descriptor jobs (orchestrator + // activities) that is the task's own queue priority; for closures it is only set when the + // enqueuer said so (post-exec passes the run's priority). Plain starters expose none — + // their job priority orders the starter script, not the work it spawns. var parentInvocation = new OperationContext.Invocation(job.Record.Name) { RunName = job.Record.RunName, + Priority = job.Descriptor != null ? job.Record.Priority : job.InheritPriority, Category = "Job" }; opScope = OperationContext.Set(parentInvocation); @@ -702,5 +712,8 @@ private sealed record QueuedJob(JobRecord Record, Func? { /// Bumped by ; entries below the live epoch are superseded. public int Epoch { get; init; } + + /// Priority nested enqueues should inherit (closure jobs only — see Enqueue). + public int? InheritPriority { get; init; } } } diff --git a/Services/Orchestration/OrchestratorRun.cs b/Services/Orchestration/OrchestratorRun.cs index 2750fb5..851ab92 100644 --- a/Services/Orchestration/OrchestratorRun.cs +++ b/Services/Orchestration/OrchestratorRun.cs @@ -5,7 +5,9 @@ public class OrchestratorRun public string Name { get; set; } = string.Empty; public string? Reference { get; set; } public string Status { get; set; } = "Pending"; - public int Priority { get; set; } = 2; + // 4 matches what every live enqueue path actually passes when a caller sets nothing — a run row + // rehydrated without a stored priority must not come back HIGHER than it originally ran. + public int Priority { get; set; } = 4; public DateTime StartedUtc { get; set; } public DateTime? CompletedUtc { get; set; } public List Tasks { get; set; } = []; diff --git a/Services/Orchestration/OrchestratorService.cs b/Services/Orchestration/OrchestratorService.cs index c8c4a8b..d3c78ce 100644 --- a/Services/Orchestration/OrchestratorService.cs +++ b/Services/Orchestration/OrchestratorService.cs @@ -1455,6 +1455,9 @@ private void DispatchPostExecution(OrchestratorRun run) _jobManager.Enqueue( name: $"{run.Name}-PostExec", priority: run.Priority, + // Post-exec commonly starts follow-up runs (baseline → cache refresh); they should land + // at this run's priority, not the enqueue default. + inheritPriority: run.Priority, runName: run.Name, work: async (jobCt) => { diff --git a/Services/PowerShellHost/PowerShellRunnerService.cs b/Services/PowerShellHost/PowerShellRunnerService.cs index 94bdd0d..1c9b872 100644 --- a/Services/PowerShellHost/PowerShellRunnerService.cs +++ b/Services/PowerShellHost/PowerShellRunnerService.cs @@ -458,13 +458,15 @@ public async Task ExecuteScript(string functionName, Dictionary? var worker = _pool.CheckoutBackground(CancellationToken.None); if (prof) checkoutTicks = Stopwatch.GetTimestamp() - checkoutStart; - // Set invocation context — inherits RunName from parent OperationContext if set by JobManager + // Set invocation context — inherits RunName and Priority from parent OperationContext if set + // by JobManager (Priority is what nested Start-CIPPOrchestrator calls inherit) var parentRun = OperationContext.Current?.RunName; var parentFunction = OperationContext.Current?.Function; var invocation = new OperationContext.Invocation(functionName) { WorkerId = $"W{worker.Id}", RunName = parentRun, + Priority = OperationContext.Current?.Priority, Category = "Job" }; using var opScope = OperationContext.Set(invocation); @@ -588,13 +590,15 @@ public async Task ExecuteScriptWithOutput(string functionName, Dictionar var worker = _pool.CheckoutBackground(CancellationToken.None); if (prof) checkoutTicks = Stopwatch.GetTimestamp() - checkoutStart; - // Set invocation context — inherits RunName from parent OperationContext if set by JobManager + // Set invocation context — inherits RunName and Priority from parent OperationContext if set + // by JobManager (Priority is what nested Start-CIPPOrchestrator calls inherit) var parentRun = OperationContext.Current?.RunName; var parentFunction = OperationContext.Current?.Function; var invocation = new OperationContext.Invocation(functionName) { WorkerId = $"W{worker.Id}", RunName = parentRun, + Priority = OperationContext.Current?.Priority, Category = "Planner" }; using var opScope = OperationContext.Set(invocation); diff --git a/Services/Storage/OrchestratorTableStore.cs b/Services/Storage/OrchestratorTableStore.cs index 2a368ea..7880d7d 100644 --- a/Services/Storage/OrchestratorTableStore.cs +++ b/Services/Storage/OrchestratorTableStore.cs @@ -132,7 +132,7 @@ public async Task> WriteRunStatusBatchAsync(IReadOnlyList< { Name = name, Status = runRow.GetString("Status") ?? "Pending", - Priority = runRow.GetInt32("Priority") ?? 2, + Priority = runRow.GetInt32("Priority") ?? 4, StartedUtc = runRow.GetDateTimeOffset("StartedUtc")?.UtcDateTime ?? DateTime.UtcNow, CompletedUtc = runRow.GetDateTimeOffset("CompletedUtc")?.UtcDateTime, TaskScriptName = runRow.GetString("TaskScriptName"), diff --git a/tests/Craft.Tests/OperationContextPriorityTests.cs b/tests/Craft.Tests/OperationContextPriorityTests.cs new file mode 100644 index 0000000..b1117d8 --- /dev/null +++ b/tests/Craft.Tests/OperationContextPriorityTests.cs @@ -0,0 +1,147 @@ +using Craft.Configuration; +using Craft.Hosting; +using Craft.Orchestration; +using Craft.PowerShellHost; +using Craft.Services; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Craft.Tests; + +/// +/// Ambient priority: nested enqueues (Start-CIPPOrchestrator called from inside a running job) read +/// to inherit the enclosing run's priority. +/// +/// Who exposes it is deliberate, not incidental: +/// - Descriptor jobs (orchestrator activities) expose their own queue priority — a child run +/// started from an activity belongs to the parent run's band. +/// - Closure jobs expose it ONLY when the enqueuer passed inheritPriority (post-exec does, with the +/// run's priority). A plain starter script must NOT donate its job priority: scheduler starters +/// run at CIPPTimers priorities (0-30) that order the starters themselves, and letting them leak +/// would silently reprioritize every fan-out they spawn. +/// +public class OperationContextPriorityTests +{ + private static JobManager NewManager(int concurrency = 2) + { + var settings = new CraftSettings(); + settings.Worker.BgPoolSize = concurrency; + var config = new ConfigurationBuilder().AddInMemoryCollection([]).Build(); + var repo = new ScriptRepository(NullLogger.Instance, settings); + var pool = new PowerShellWorkerPool(repo, NullLogger.Instance, config, settings); + var limiter = new BackgroundTaskLimiter(NullLogger.Instance, config, settings, pool); + return new JobManager(NullLogger.Instance, settings, limiter); + } + + private static Task Start(JobManager jobs) => Task.Run(() => jobs.StartAsync(CancellationToken.None)); + + private static async Task WaitUntilAsync(Func condition, string because) + { + for (var i = 0; i < 200 && !condition(); i++) await Task.Delay(25); + Assert.True(condition(), because); + } + + [Fact] + public async Task DescriptorJob_ExposesItsQueuePriorityAmbiently() + { + var jobs = NewManager(); + int? observed = int.MinValue; + var done = 0; + + jobs.SetWorkResolver((descriptor, _) => Task.FromResult?>( + _ => + { + observed = OperationContext.Current?.Priority; + done = 1; + return Task.CompletedTask; + })); + + _ = Start(jobs); + jobs.Enqueue(new JobDescriptor("run", "task-0", 3), "run-task-0"); + + await WaitUntilAsync(() => Volatile.Read(ref done) == 1, "descriptor job never ran"); + Assert.Equal(3, observed); + await jobs.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task ClosureJob_WithoutInheritPriority_ExposesNone() + { + var jobs = NewManager(); + int? observed = int.MinValue; + var done = 0; + + _ = Start(jobs); + // A starter script's own priority (here 1, like a CIPPTimers starter) must not leak to the + // work it enqueues. + jobs.Enqueue("starter", priority: 1, work: _ => + { + observed = OperationContext.Current?.Priority; + done = 1; + return Task.CompletedTask; + }); + + await WaitUntilAsync(() => Volatile.Read(ref done) == 1, "closure job never ran"); + Assert.Null(observed); + await jobs.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task ClosureJob_WithInheritPriority_ExposesIt() + { + var jobs = NewManager(); + int? observed = int.MinValue; + var done = 0; + + _ = Start(jobs); + // Post-exec shape: the job itself runs at the run's priority AND donates it to nested enqueues. + jobs.Enqueue("run-PostExec", priority: 3, work: _ => + { + observed = OperationContext.Current?.Priority; + done = 1; + return Task.CompletedTask; + }, runName: "run", inheritPriority: 3); + + await WaitUntilAsync(() => Volatile.Read(ref done) == 1, "post-exec job never ran"); + Assert.Equal(3, observed); + await jobs.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task ChangePriority_PreservesInheritPriority() + { + var jobs = NewManager(concurrency: 1); + int? observed = int.MinValue; + var done = 0; + var release = new SemaphoreSlim(0); + + _ = Start(jobs); + + // Hold the only slot so the target stays queued long enough to reprioritize. + jobs.Enqueue("blocker", priority: 0, work: async ct => + await release.WaitAsync(TimeSpan.FromSeconds(10), CancellationToken.None)); + + var id = jobs.Enqueue("run-PostExec", priority: 3, work: _ => + { + observed = OperationContext.Current?.Priority; + done = 1; + return Task.CompletedTask; + }, runName: "run", inheritPriority: 3); + + // The reprioritized entry is a `with`-copy of the original — inheritPriority must ride along. + Assert.True(jobs.ChangePriority(id, 1), "target was not queued when reprioritized"); + release.Release(); + + await WaitUntilAsync(() => Volatile.Read(ref done) == 1, "reprioritized job never ran"); + Assert.Equal(3, observed); + await jobs.StopAsync(CancellationToken.None); + } + + [Fact] + public void QueueBridge_TwoArgEnqueue_KeepsTheHistoricalDefault() + { + // Callers that predate priorities must keep landing at P5 — the 3-arg overload exists so + // user-initiated starters can opt INTO a higher band, not to move everyone else. + Assert.Equal(5, new QueueBridge.PendingQueueCommand("Start-Thing", "{}").Priority); + } +}