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/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/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 01d28b9..d3c78ce 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" @@ -1345,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/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/Services/Storage/OrchestratorTableStore.cs b/Services/Storage/OrchestratorTableStore.cs index 8a083bc..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"), @@ -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/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); + } +} 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] 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); + } +}