From 8dcc7e028bdf1d19eb0e72be17723c5d57a7790b Mon Sep 17 00:00:00 2001 From: Rick Hohler Date: Sat, 24 Oct 2026 19:43:07 -0500 Subject: [PATCH] Fix job cancellation, orchestrator errors, queue race, lock release fix-async-work-correctness (DesignAlgorithmsKit-Specs). - JobManager.cancel removes the pending job so it never runs - TaskOrchestrator wraps task errors in OrchestrationError.taskFailure - ProcessingQueue marks items .processing synchronously before spawning, bounding concurrency and starting each item once; drops finished handles - ResourceCoordinator releases locks before access returns, evicts idle per-key locks, adds withWriteLock(for:operation:) Two queue tests relied on add/retry leaving items pending until a later task ran; they now pause the queue first. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 9 + README.md | 1 - .../Behavioral/JobManager.swift | 6 +- .../Orchestration/TaskOrchestrator.swift | 22 ++- .../Behavioral/Queue.swift | 94 ++++------ .../ResourceCoordinator.swift | 55 ++++-- .../AsyncWorkCorrectnessTests.swift | 177 ++++++++++++++++++ .../Behavioral/QueueTests.swift | 4 +- 8 files changed, 282 insertions(+), 86 deletions(-) create mode 100644 Tests/DesignAlgorithmsKitTests/Behavioral/AsyncWorkCorrectnessTests.swift diff --git a/CHANGELOG.md b/CHANGELOG.md index 72a1b0d..7b8666c 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/README.md b/README.md index 43840dc..9a81b83 100644 --- a/README.md +++ b/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. diff --git a/Sources/DesignAlgorithmsKit/Behavioral/JobManager.swift b/Sources/DesignAlgorithmsKit/Behavioral/JobManager.swift index 5f42b12..81702d8 100644 --- a/Sources/DesignAlgorithmsKit/Behavioral/JobManager.swift +++ b/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 diff --git a/Sources/DesignAlgorithmsKit/Behavioral/Orchestration/TaskOrchestrator.swift b/Sources/DesignAlgorithmsKit/Behavioral/Orchestration/TaskOrchestrator.swift index 8e1a13f..ee2b27f 100644 --- a/Sources/DesignAlgorithmsKit/Behavioral/Orchestration/TaskOrchestrator.swift +++ b/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,10 +123,7 @@ 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) } } } } @@ -135,4 +131,14 @@ public actor TaskOrchestrator { } } } + + /// 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 + } } diff --git a/Sources/DesignAlgorithmsKit/Behavioral/Queue.swift b/Sources/DesignAlgorithmsKit/Behavioral/Queue.swift index 103a8a6..adad449 100644 --- a/Sources/DesignAlgorithmsKit/Behavioral/Queue.swift +++ b/Sources/DesignAlgorithmsKit/Behavioral/Queue.swift @@ -174,9 +174,7 @@ public actor ProcessingQueue where P // Start processing if not paused if !isPaused { - Task { - await processQueue() - } + processQueue() } } @@ -220,9 +218,7 @@ public actor ProcessingQueue 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 where P items[index].progress = 0.0 if !isPaused { - Task { - await processQueue() - } + processQueue() } } @@ -266,66 +260,48 @@ public actor ProcessingQueue 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() } } diff --git a/Sources/DesignAlgorithmsKit/Behavioral/ResourceCoordinator/ResourceCoordinator.swift b/Sources/DesignAlgorithmsKit/Behavioral/ResourceCoordinator/ResourceCoordinator.swift index 3e059fd..d3eaa7c 100644 --- a/Sources/DesignAlgorithmsKit/Behavioral/ResourceCoordinator/ResourceCoordinator.swift +++ b/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,26 +17,38 @@ public actor ResourceCoordinator { public func withReadLock(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(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(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 + 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 } @@ -42,6 +56,19 @@ public actor ResourceCoordinator { 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 diff --git a/Tests/DesignAlgorithmsKitTests/Behavioral/AsyncWorkCorrectnessTests.swift b/Tests/DesignAlgorithmsKitTests/Behavioral/AsyncWorkCorrectnessTests.swift new file mode 100644 index 0000000..08be01f --- /dev/null +++ b/Tests/DesignAlgorithmsKitTests/Behavioral/AsyncWorkCorrectnessTests.swift @@ -0,0 +1,177 @@ +// +// AsyncWorkCorrectnessTests.swift +// DesignAlgorithmsKitTests +// +// Cancellation, failure wrapping, queue concurrency and lock release (fix-async-work-correctness) +// + +import XCTest +@testable import DesignAlgorithmsKit + +@available(macOS 12.0, iOS 15.0, *) +final class AsyncWorkCorrectnessTests: XCTestCase { + + private actor Flag { + private(set) var isSet = false + func set() { isSet = true } + } + + private actor Log { + private(set) var entries: [String] = [] + func append(_ entry: String) { entries.append(entry) } + } + + private struct Boom: Error, Equatable {} + + /// Polls `condition` until it holds or `timeout` passes. + private func waitUntil(timeout: TimeInterval = 5, _ condition: () async -> Bool) async throws { + let deadline = Date().addingTimeInterval(timeout) + while await !condition() { + guard Date() < deadline else { + XCTFail("Timed out") + return + } + try await Task.sleep(nanoseconds: 10_000_000) + } + } + + // MARK: - JobManager + + func testCancelledPendingJobNeverRuns() async throws { + let manager = JobManager(maxConcurrentJobs: 1) + let ran = Flag() + let blocker = await manager.submit(description: "blocker") { + try await Task.sleep(nanoseconds: 100_000_000) + return "blocker" + } + let queued = await manager.submit(description: "queued") { + await ran.set() + return "ran" + } + await manager.cancel(id: queued) + + try await waitUntil { await manager.getJob(id: blocker)?.status == .completed } + try await Task.sleep(nanoseconds: 100_000_000) + + let didRun = await ran.isSet + XCTAssertFalse(didRun) + let job = await manager.getJob(id: queued) + XCTAssertEqual(job?.status, .failed) + XCTAssertEqual(job?.errorMessage, "Cancelled") + } + + // MARK: - TaskOrchestrator + + func testFailingTaskIsWrappedAndDependentsDoNotRun() async throws { + let orchestrator = TaskOrchestrator() + let log = Log() + await orchestrator.addTask(DAGTask(id: "a") { await log.append("a") }) + await orchestrator.addTask(DAGTask(id: "b", dependencies: ["a"]) { throw Boom() }) + await orchestrator.addTask(DAGTask(id: "c", dependencies: ["b"]) { await log.append("c") }) + + do { + try await orchestrator.execute() + XCTFail("Expected taskFailure") + } catch OrchestrationError.taskFailure(let taskId, let error) { + XCTAssertEqual(taskId, "b") + XCTAssertEqual(error as? Boom, Boom()) + } + let entries = await log.entries + XCTAssertEqual(entries, ["a"]) + } + + // MARK: - ProcessingQueue + + private struct Item: QueueItem { + let id: Int + var status: QueueItemStatus = .pending + var progress: Double = 0 + } + + private actor CountingProcessor: QueueProcessor { + private(set) var runs: [Int: Int] = [:] + private(set) var maxConcurrency = 0 + private var current = 0 + + func process(_ item: Item) async throws { + runs[item.id, default: 0] += 1 + current += 1 + maxConcurrency = max(maxConcurrency, current) + try await Task.sleep(nanoseconds: 2_000_000) + current -= 1 + } + } + + func testRapidAdditionsRunEachItemOnceWithinTheLimit() async throws { + let processor = CountingProcessor() + let queue = ProcessingQueue(processor: processor, maxConcurrent: 1) + for id in 0..<10 { + await queue.add([Item(id: id)]) + } + await queue.resume() + await queue.resume() + + try await waitUntil { await queue.completedItems.count == 10 } + + let runs = await processor.runs + let maxConcurrency = await processor.maxConcurrency + XCTAssertEqual(runs.count, 10) + XCTAssertTrue(runs.values.allSatisfy { $0 == 1 }, "\(runs)") + XCTAssertLessThanOrEqual(maxConcurrency, 1) + let handles = await queue.activeTaskCount + XCTAssertEqual(handles, 0) + } + + func testItemsAreProcessingAsSoonAsAdded() async { + let queue = ProcessingQueue(processor: CountingProcessor(), maxConcurrent: 2) + await queue.add([Item(id: 1), Item(id: 2), Item(id: 3)]) + let processing = await queue.processingItems.map(\.id) + let pending = await queue.pendingItems.map(\.id) + XCTAssertEqual(processing, [1, 2]) + XCTAssertEqual(pending, [3]) + } + + // MARK: - ResourceCoordinator + + func testReadThenWriteOnSameKeyAndLockStateIsDropped() async throws { + let coordinator = ResourceCoordinator() + let read = try await coordinator.withReadLock(for: "k") { "read" } + let write = try await coordinator.withWriteLock(for: "k") { "write" } + XCTAssertEqual([read, write], ["read", "write"]) + let tracked = await coordinator.trackedKeyCount + XCTAssertEqual(tracked, 0) + } + + func testThrowingWriteReleasesItsLock() async throws { + let coordinator = ResourceCoordinator() + do { + _ = try await coordinator.withWriteLock(for: "k") { () async throws -> Int in throw Boom() } + XCTFail("Expected Boom") + } catch is Boom {} + let value = try await coordinator.withWriteLock(for: "k") { 42 } + XCTAssertEqual(value, 42) + let tracked = await coordinator.trackedKeyCount + XCTAssertEqual(tracked, 0) + } + + func testWritesAreExclusive() async throws { + let coordinator = ResourceCoordinator() + let log = Log() + try await withThrowingTaskGroup(of: Void.self) { group in + for index in 0..<5 { + group.addTask { + try await coordinator.withWriteLock(for: "k") { + await log.append("start \(index)") + try await Task.sleep(nanoseconds: 2_000_000) + await log.append("end \(index)") + } + } + } + try await group.waitForAll() + } + let entries = await log.entries + for pair in stride(from: 0, to: entries.count, by: 2) { + XCTAssertEqual(entries[pair].dropFirst(6), entries[pair + 1].dropFirst(4), "\(entries)") + } + } +} diff --git a/Tests/DesignAlgorithmsKitTests/Behavioral/QueueTests.swift b/Tests/DesignAlgorithmsKitTests/Behavioral/QueueTests.swift index 688e299..062171a 100644 --- a/Tests/DesignAlgorithmsKitTests/Behavioral/QueueTests.swift +++ b/Tests/DesignAlgorithmsKitTests/Behavioral/QueueTests.swift @@ -279,6 +279,8 @@ final class QueueTests: XCTestCase { let queue = ProcessingQueue(processor: processor, maxConcurrent: 1) let item1 = TestItem(id: UUID(), data: "data1") + // Paused, so the retried item stays pending (add/retry start items immediately otherwise). + await queue.pause() await queue.add([item1]) await queue.updateItem(id: item1.id, status: .failed) @@ -442,8 +444,8 @@ final class QueueTests: XCTestCase { let queue = ProcessingQueue(processor: processor, maxConcurrent: 1) let item1 = TestItem(id: UUID(), data: "data1") - await queue.add([item1]) await queue.pause() + await queue.add([item1]) // When - wait a bit to see if processing happens try? await Task.sleep(nanoseconds: 100_000_000) // 0.1 seconds