Fix job cancellation, orchestrator errors, queue race, lock release

CHANGELOG.md

@@ -7,6 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added (`fix-async-work-correctness`)
+- `ResourceCoordinator.withWriteLock(for:operation:)`.
+
+### Fixed (`fix-async-work-correctness`)
+- `JobManager.cancel(id:)` removes a pending job from the queue, so it never runs and stays `.failed`/"Cancelled".
+- `TaskOrchestrator.execute()` throws `OrchestrationError.taskFailure(taskId:error:)` for a failing task; its dependents do not run. **Callers that caught the raw task error must unwrap `taskFailure`.**
+- `ProcessingQueue` marks items `.processing` before spawning their task, so it never starts an item twice or exceeds `maxConcurrent`, and it drops finished task handles. `add`, `resume` and `retry` now start items before returning (pause the queue first to keep items pending).
+- `ResourceCoordinator` releases a key's lock before `access` returns or throws and drops idle per-key lock state.
+
 ### Fixed (`fix-algorithm-edge-cases`)
 - `MerkleTree.generateProof(for:)` adds a path entry for an unpaired node (its own hash as the right sibling), so proofs verify for every leaf of trees with an odd node count. Roots are unchanged.
 - `BloomFilter` and `CountingBloomFilter` no longer trap on degenerate sizing: a capacity below 1 counts as 1 and the false-positive rate is clamped into (0, 1). `capacity` and `falsePositiveRate` report the clamped values.

README.md

@@ -98,7 +98,6 @@ swift test
 
 ## Known issues
 
-- `JobManager.cancel` does not stop a pending job; `TaskOrchestrator` never throws `taskFailure`; `ProcessingQueue` has a scheduling race; `ResourceCoordinator` releases locks late.
 - `ULID` text is not compatible with the ULID specification.
 - Platform claims are broader than tested, `WASMGuard.swift` is dead code, SHA code is duplicated, and the DocC catalog is not attached to the target.
 

Sources/DesignAlgorithmsKit/Behavioral/JobManager.swift

@@ -151,11 +151,11 @@ public actor JobManager {
         )
     }
 
-    /// Cancel a job (Not fully implemented in this basic pattern without Task storage)
+    /// Cancel a pending job: it is removed from the queue, never runs, and ends `.failed` with
+    /// "Cancelled". Running jobs are not interrupted.
     public func cancel(id: UUID) {
-        // In a full implementation, we would store the Task handle and cancel it.
-        // For this pattern demo, we simply mark as failed if pending.
         if var state = jobs[id], state.status == .pending {
+            pendingJobs.removeAll { $0.0 == id }
             state.status = .failed
             state.error = "Cancelled"
             jobs[id] = state

Sources/DesignAlgorithmsKit/Behavioral/Orchestration/TaskOrchestrator.swift

@@ -77,6 +77,8 @@ public actor TaskOrchestrator {
 
     /// Executes all tasks in the graph, respecting dependencies.
     /// Runs independent tasks in parallel where possible.
+    /// - Throws: `OrchestrationError.taskFailure(taskId:error:)` when a task throws; tasks that
+    ///   depend on it do not run.
     public func execute() async throws {
         try validate()
 
@@ -100,10 +102,7 @@ public actor TaskOrchestrator {
             // Initial batch
             for taskId in readyQueue {
                 guard let task = tasks[taskId] else { continue }
-                group.addTask {
-                    try await task.operation()
-                    return taskId
-                }
+                group.addTask { try await Self.run(task) }
             }
 
             // Loop as tasks finish
@@ -124,15 +123,22 @@ public actor TaskOrchestrator {
                          if inDegree[downstreamId] == 0 {
                              // Ready!
                              if let task = tasks[downstreamId] {
-                                 group.addTask {
-                                     try await task.operation()
-                                     return downstreamId
-                                 }
+                                 group.addTask { try await Self.run(task) }
                              }
                          }
                      }
                  }
             }
         }
     }
+
+    /// Runs one task, tagging its error with the task's ID.
+    private static func run(_ task: DAGTask) async throws -> String {
+        do {
+            try await task.operation()
+        } catch {
+            throw OrchestrationError.taskFailure(taskId: task.id, error: error)
+        }
+        return task.id
+    }
 }

Sources/DesignAlgorithmsKit/Behavioral/Queue.swift

@@ -174,9 +174,7 @@ public actor ProcessingQueue<Item: QueueItem, Processor: QueueProcessor> where P
 
         // Start processing if not paused
         if !isPaused {
-            Task {
-                await processQueue()
-            }
+            processQueue()
         }
     }
 
@@ -220,9 +218,7 @@ public actor ProcessingQueue<Item: QueueItem, Processor: QueueProcessor> where P
     /// Resume queue processing
     public func resume() {
         isPaused = false
-        Task {
-            await processQueue()
-        }
+        processQueue()
     }
 
     /// Retry a failed item
@@ -237,9 +233,7 @@ public actor ProcessingQueue<Item: QueueItem, Processor: QueueProcessor> where P
         items[index].progress = 0.0
 
         if !isPaused {
-            Task {
-                await processQueue()
-            }
+            processQueue()
         }
     }
 
@@ -266,66 +260,48 @@ public actor ProcessingQueue<Item: QueueItem, Processor: QueueProcessor> where P
         }
     }
 
-    /// Process the queue (internal method)
-    private func processQueue() async {
+    /// Start pending items while slots are free. Items are marked `.processing` here, before
+    /// their task is spawned, so back-to-back calls never start an item twice or exceed
+    /// `maxConcurrent`.
+    private func processQueue() {
         guard !isPaused else { return }
-
-        let pending = items.filter { $0.status == .pending }
-        let processing = items.filter { $0.status == .processing }
-        let availableSlots = maxConcurrent - processing.count
-
-        guard availableSlots > 0 else {
-            return
-        }
-
-        let toProcess = Array(pending.prefix(availableSlots))
-
-        for item in toProcess {
-            processItem(item)
+
+        var availableSlots = maxConcurrent - items.filter { $0.status == .processing }.count
+        var index = items.startIndex
+        while availableSlots > 0, index < items.endIndex {
+            if items[index].status == .pending {
+                items[index].status = .processing
+                items[index].progress = 0.0
+                processItem(items[index])
+                availableSlots -= 1
+            }
+            index += 1
         }
     }
-    
-    /// Process a single item (internal method)
+
+    /// Process a single item that is already marked `.processing` (internal method)
     private func processItem(_ item: Item) {
         let itemID = item.id
-        let task = Task { [weak self] in
+        let task = Task { [weak self, processor] in
             guard let self = self else { return }
             do {
-                // Update status to processing
-                // Note: await is required for actor isolation, even though updateItem is synchronous
-                // swiftlint:disable:next no_async_await_expression
-                await self.updateItem(id: itemID, status: .processing, progress: 0.0)
-
-                // Get current item (may have been updated)
-                // Note: await is required for actor isolation when accessing actor-isolated property
-                // swiftlint:disable:next no_async_await_expression
-                let currentItem = await self.items.first(where: { $0.id == itemID })
-                guard let itemToProcess = currentItem else {
-                    // Item was removed
-                    return
-                }
-
-                // Process the item
-                try await processor.process(itemToProcess)
-
-                // Mark as completed
-                // swiftlint:disable:next no_async_await_expression
-                await self.updateItem(id: itemID, status: .completed, progress: 1.0)
-
-                // Continue processing queue
-                await self.processQueue()
-
+                try await processor.process(item)
+                await self.finish(id: itemID, status: .completed, progress: 1.0)
             } catch {
-                // Mark as failed
-                // swiftlint:disable:next no_async_await_expression
-                await self.updateItem(id: itemID, status: .failed, progress: 0.0)
-
-                // Continue processing queue
-                await self.processQueue()
+                await self.finish(id: itemID, status: .failed, progress: 0.0)
             }
         }
-
-        processingTasks[item.id] = task
+        processingTasks[itemID] = task
+    }
+
+    /// Number of live task handles (for tests).
+    var activeTaskCount: Int { processingTasks.count }
+
+    /// Record an item's outcome, drop its task handle and start the next items.
+    private func finish(id: Item.ID, status: QueueItemStatus, progress: Double) {
+        processingTasks.removeValue(forKey: id)
+        updateItem(id: id, status: status, progress: progress)
+        processQueue()
     }
 }
 

Sources/DesignAlgorithmsKit/Behavioral/ResourceCoordinator/ResourceCoordinator.swift

@@ -7,6 +7,8 @@ import Foundation
 public actor ResourceCoordinator {
 
     private var locks: [String: PathLock] = [:]
+    /// Operations holding or awaiting each key's lock; the lock is dropped when this reaches 0.
+    private var users: [String: Int] = [:]
 
     public init() {}
 
@@ -15,33 +17,58 @@ public actor ResourceCoordinator {
     public func withReadLock<T>(for key: String, operation: () async throws -> T) async throws -> T {
         return try await access(path: key, type: .read, operation: operation)
     }
+
+    /// Executes the given block with an exclusive (write) lock on the specified resource.
+    public func withWriteLock<T>(for key: String, operation: () async throws -> T) async throws -> T {
+        return try await access(path: key, type: .write, operation: operation)
+    }
 
-    /// Helper to just acquire and return a token?
-    /// No, closures are safer.
-
+    /// Runs `operation` under a read or write lock on `path`. The lock is released before this
+    /// returns or throws.
     public func access<T>(path: String, type: AccessType, operation: () async throws -> T) async throws -> T {
-        let lock = getLock(for: path)
+        let lock = checkOut(path)
         switch type {
-        case .read:
-            await lock.lockRead()
-            // We use standard do/defer pattern here since we are inside an async function
-            defer { Task { await lock.unlockRead() } }
-            return try await operation()
-        case .write:
-            await lock.lockWrite()
-            defer { Task { await lock.unlockWrite() } }
-            return try await operation()
+        case .read: await lock.lockRead()
+        case .write: await lock.lockWrite()
         }
+
+        let result: Result<T, Error>
+        do {
+            result = .success(try await operation())
+        } catch {
+            result = .failure(error)
+        }
+
+        switch type {
+        case .read: await lock.unlockRead()
+        case .write: await lock.unlockWrite()
+        }
+        checkIn(path)
+        return try result.get()
     }
 
-    private func getLock(for key: String) -> PathLock {
+    private func checkOut(_ key: String) -> PathLock {
+        users[key, default: 0] += 1
         if let lock = locks[key] {
             return lock
         }
         let newLock = PathLock()
         locks[key] = newLock
         return newLock
     }
+
+    private func checkIn(_ key: String) {
+        let remaining = (users[key] ?? 1) - 1
+        if remaining > 0 {
+            users[key] = remaining
+        } else {
+            users.removeValue(forKey: key)
+            locks.removeValue(forKey: key)
+        }
+    }
+
+    /// Number of keys with lock state (for tests).
+    var trackedKeyCount: Int { locks.count }
 
     public enum AccessType {
         case read