fix(async): preserve cancellation and add revision-aware job gate

packages/async/src/task.zigmodified+54-20

Changes to packages/async/src/task.zig

		⋯15 linesShow the 15 lines above this hunk@@ -16,18 +16,8 @@ pub const TaskId = struct {
16	16	};
17	17
18	18	/// Task state
19		-pub const TaskState = enum(u8) {
20		- /// Task is ready to run
21		- Pending,
22		- /// Task is currently running
23		- Running,
24		- /// Task completed successfully
25		- Completed,
26		- /// Task failed with an error
27		- Failed,
28		- /// Task was cancelled
29		- Cancelled,
30		-};
	19	+const job_state = ;
	20	+pub const TaskState = job_state.State;
31	21
32	22	/// A spawned task that can be awaited
33	23	///
		⋯24 linesShow the 24 lines above this hunk@@ -58,6 +48,8 @@ pub fn Task(comptime T: type) type {
58	48	return task;
59	49	}
60	50
	51	+ /// Call only after the poll owner/executor has quiesced. Cancellation
	52	+ /// marks an outcome; it does not stop a currently executing poll.
61	53	pub fn deinit(self: *Self) void {
62	54	self.future.deinit();
63	55	self.allocator.destroy(self);
		⋯24 linesShow the 24 lines above this hunk@@ -88,19 +80,17 @@ pub fn Task(comptime T: type) type {
88	80	switch (result) {
89	81	.Ready => |value| {
90	82	self.result = value;
91		- self.state.store(.Completed, .release);
92		- return true;
	83	+ return job_state.finishPoll(&self.state, true);
93	84	},
94	85	.Pending => {
95		- self.state.store(.Pending, .release);
96		- return false;
	86	+ return job_state.finishPoll(&self.state, false);
97	87	},
98	88	}
99	89	}
100	90
101	91	/// Cancel the task
102	92	pub fn cancel(self: *Self) void {
103		- self.state.store(.Cancelled, .release);
	93	+ _ = job_state.cancel(&self.state);
104	94	}
105	95
106	96	/// Check if task is completed
		⋯30 linesShow the 30 lines above this hunk@@ -137,9 +127,9 @@ pub fn JoinHandle(comptime T: type) type {
137	127	/// Only use this from non-async code or the runtime's block_on.
138	128	pub fn await(self: Self) !T {
139	129	while (!self.task.isCompleted()) {
140		- if (self.task.isCancelled()) {
141		- return error.TaskCancelled;
142		- }
	130	+ const state = self.task.state.load(.acquire);
	131	+ if (state == .Cancelled) return error.TaskCancelled;
	132	+ if (state == .Failed) return error.TaskFailed;
143	133
144	134	// Spin or yield
145	135	std.Thread.yield() catch {};
		⋯235 linesShow the 235 lines above this hunk@@ -381,3 +371,47 @@ test "RawTask - type erasure" {
381	371	try testing.expect(completed);
382	372	try testing.expectEqual(@as(i32, 77), task.result.?);
383	373	}
	374	+
	375	+fn regressionContext() Context {
	376	+ const Noop = struct {
	377	+ fn wake(_: *anyopaque) void {}
	378	+ fn clone(ptr: *anyopaque) *anyopaque {
	379	+ return ptr;
	380	+ }
	381	+ };
	382	+ return Context.init(.{ .data = undefined, .vtable = &.{ .wake = Noop.wake, .wake_by_ref = Noop.wake, .clone = Noop.clone, .drop = Noop.wake } });
	383	+}
	384	+test "cancel during a future poll survives both Ready and Pending returns" {
	385	+ const Poller = struct {
	386	+ task: *Task(u32) = undefined,
	387	+ ready: bool,
	388	+ fn poll(ptr: *anyopaque, _: *Context) future_mod.PollResult(u32) {
	389	+ const self: *@This() = @ptrCast(@alignCast(ptr));
	390	+ self.task.cancel();
	391	+ return if (self.ready) .{ .Ready = 42 } else .Pending;
	392	+ }
	393	+ };
	394	+ for ([_]bool{ false, true }) |ready| {
	395	+ var poller: Poller = .{ .ready = ready };
	396	+ const task = try Task(u32).init(std.testing.allocator, .{ .state = &poller, .poll_fn = Poller.poll });
	397	+ defer task.deinit();
	398	+ poller.task = task;
	399	+ var ctx = regressionContext();
	400	+ try std.testing.expect(task.poll(&ctx));
	401	+ try std.testing.expect(task.isCancelled());
	402	+ try std.testing.expect(!task.isCompleted());
	403	+ const handle: JoinHandle(u32) = .{ .task = task };
	404	+ try std.testing.expectError(error.TaskCancelled, handle.await());
	405	+ }
	406	+}
	407	+test "completed tasks resist late cancellation and failed joins return immediately" {
	408	+ const task = try Task(u32).init(std.testing.allocator, try future_mod.ready(u32, 42, std.testing.allocator));
	409	+ defer task.deinit();
	410	+ var ctx = regressionContext();
	411	+ try std.testing.expect(task.poll(&ctx));
	412	+ task.cancel();
	413	+ const handle: JoinHandle(u32) = .{ .task = task };
	414	+ try std.testing.expectEqual(@as(u32, 42), try handle.await());
	415	+ task.state.store(.Failed, .release);
	416	+ try std.testing.expectError(error.TaskFailed, handle.await());
	417	+}