Propagate cancellation causes through task trees

benchmark/cancel_cause.rb

@@ -0,0 +1,78 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Shopify Inc.
+# Copyright, 2026, by Samuel Williams.
+
+require "async"
+
+module CountCancelCauses
+	attr_accessor :for_calls
+
+	def for(*arguments, **options, &block)
+		self.for_calls ||= 0
+		self.for_calls += 1
+
+		super(*arguments, **options, &block)
+	end
+end
+
+Async::Cancel::Cause.singleton_class.prepend(CountCancelCauses)
+
+def positive_integer(name, default)
+	Integer(ENV.fetch(name, default)).tap do |value|
+		abort("#{name} must be positive!") unless value.positive?
+	end
+end
+
+def measure_cancel(total_tasks)
+	scheduler = Async::Scheduler.new
+	ready = Thread::Queue.new
+
+	Fiber.set_scheduler(scheduler)
+
+	begin
+		total_tasks.times do
+			scheduler.async do
+				ready.push(true)
+				sleep
+			end
+		end
+
+		total_tasks.times do
+			ready.pop
+		end
+
+		GC.start
+		Async::Cancel::Cause.for_calls = 0
+
+		# Measure the synchronous fan-out only; draining cancelled tasks happens in `ensure`.
+		allocated_before = GC.stat(:total_allocated_objects)
+		start_time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+		scheduler.cancel
+
+		duration = Process.clock_gettime(Process::CLOCK_MONOTONIC) - start_time
+		allocated_after = GC.stat(:total_allocated_objects)
+
+		return {
+			tasks: total_tasks,
+			cause_for_calls: Async::Cancel::Cause.for_calls,
+			allocated_objects: allocated_after - allocated_before,
+			duration_ms: duration * 1000.0
+		}
+	ensure
+		scheduler.run unless scheduler.closed?
+		Fiber.set_scheduler(nil)
+	end
+end
+
+tasks = positive_integer("TASKS", 1365)
+
+result = measure_cancel(tasks)
+
+puts "tasks=#{result.fetch(:tasks)}"
+puts "cause_for_calls=#{result.fetch(:cause_for_calls)}"
+puts "allocated_objects=#{result.fetch(:allocated_objects)}"
+puts "duration_ms=#{format('%.2f', result.fetch(:duration_ms))}"

lib/async/node.rb

@@ -283,9 +283,10 @@ def terminate
 		# Attempt to cancel the current node immediately, including all non-transient children. Invokes {#stop_children} to cancel all children.
 		#
 		# @parameter later [Boolean] Whether to defer cancelling until some point in the future.
-		def cancel(later = false)
+		# @parameter cause [Exception | Nil] The cause of the cancel operation.
+		def cancel(later = false, cause: $!)
 			# The implementation of this method may defer calling `stop_children`.
-			stop_children(later)
+			stop_children(later, cause: cause)
 		end
 
 		# Backward compatibility alias for {#cancel}.
@@ -295,9 +296,9 @@ def stop(...)
 		end
 
 		# Attempt to stop all non-transient children.
-		private def stop_children(later = false)
+		private def stop_children(later = false, cause:)
 			@children&.each do |child|
-				child.cancel(later) unless child.transient?
+				child.cancel(later, cause: cause) unless child.transient?
 			end
 		end
 

lib/async/promise.rb

@@ -98,7 +98,9 @@ def value
 			end
 
 			# Wait with deadline tracking:
-			until @resolved
+			loop do
+				break if @resolved
+
 				# Get remaining time for this wait iteration:
 				remaining = deadline.remaining
 

lib/async/scheduler.rb

@@ -514,15 +514,19 @@ def run_once(timeout = nil)
 		# Cancel all children, including transient children.
 		#
 		# @public Since *Async v1*.
-		def cancel
+		def cancel(later = false, cause: $!)
+			if @children and !cause
+				cause = Cancel::Cause.for("Cancelling task!")
+			end
+
 			@children&.each do |child|
-				child.cancel
+				child.cancel(later, cause: cause)
 			end
 		end
 
 		# Backward compatibility alias for cancel.
-		def stop
-			cancel
+		def stop(...)
+			cancel(...)
 		end
 
 		# Run the event loop, until the scheduler is interrupted or finished.

lib/async/task.rb

@@ -333,7 +333,7 @@ def cancel(later = false, cause: $!)
 
 			if self.cancelled?
 				# If the task is already cancelled, a `cancel` state transition re-enters the same state which is a no-op. However, we will also attempt to cancel any running children too. This can happen if the children did not cancel correctly the first time around. Doing this should probably be considered a bug, but it's better to be safe than sorry.
-				return cancelled!
+				return cancelled!(cause: cause)
 			end
 
 			# If the fiber is alive, we need to cancel it:
@@ -369,7 +369,7 @@ def cancel(later = false, cause: $!)
 				end
 			else
 				# We are not running, but children might be, so transition directly into cancelled state:
-				cancel!
+				cancel!(cause: cause)
 			end
 		end
 
@@ -479,7 +479,7 @@ def failed!(exception = false)
 			@promise.reject(exception)
 		end
 
-		def cancelled!
+		def cancelled!(cause: $!)
 			# Console.info(self, status:) {"Task #{self} was cancelled with #{@children&.size.inspect} children!"}
 
 			# Cancel the promise, specify nil here so that no exception is raised when waiting on the promise:
@@ -489,24 +489,25 @@ def cancelled!
 
 			begin
 				# We are not running, but children might be so we should stop them:
-				stop_children(true)
-			rescue Cancel
+				stop_children(true, cause: cause)
+			rescue Cancel => exception
 				cancelled = true
+				cause ||= exception.cause || exception
 				# If we are cancelling children, and one of them tries to cancel the current task, we should ignore it. We will be cancelled later.
 				retry
 			end
 
 			if cancelled
-				raise Cancel, "Cancelling current task!"
+				raise Cancel, "Cancelling current task!", cause: cause
 			end
 		end
 
 		def stopped!
 			cancelled!
 		end
 
-		def cancel!
-			cancelled!
+		def cancel!(cause: $!)
+			cancelled!(cause: cause)
 
 			finish!
 		end
@@ -519,8 +520,8 @@ def schedule(&block)
 			@fiber = Fiber.new(annotation: self.annotation) do
 				begin
 					completed!(yield)
-				rescue Cancel
-					cancelled!
+				rescue Cancel => exception
+					cancelled!(cause: exception.cause || exception)
 				rescue StandardError => error
 					failed!(error)
 				rescue Exception => exception

test/async/limited_queue.rb

@@ -4,16 +4,39 @@
 # Copyright, 2025, by Samuel Williams.
 # Copyright, 2025, by Shopify Inc.
 
-require "async/limited_queue"
+require "async/queue"
 
 require "sus/fixtures/async"
 require "async/a_queue"
 
+limited_queue_new = Async::LimitedQueue.method(:new)
+
+require "async/limited_queue"
+
 describe Async::LimitedQueue do
 	include Sus::Fixtures::Async::ReactorContext
 
 	it_behaves_like Async::AQueue
 
+	it "warns when constructed through async/queue" do
+		verbose = $VERBOSE
+		$VERBOSE = true
+
+		warning = nil
+		Async::LimitedQueue.define_singleton_method(:warn) do |*arguments, **options|
+			warning = arguments.first
+		end
+
+		queue = limited_queue_new.call(1)
+
+		expect(queue.limit).to be == 1
+		expect(warning).to be =~ /require 'async\/limited_queue'/
+	ensure
+		singleton_class = class << Async::LimitedQueue; self; end
+		singleton_class.remove_method(:warn) if singleton_class.instance_methods(false).include?(:warn)
+		$VERBOSE = verbose
+	end
+
 	let(:queue) {subject.new}
 
 	with "#limit"	do
@@ -108,4 +131,3 @@ def before
 		end
 	end
 end
-

test/async/list.rb

@@ -67,6 +67,12 @@ def initialize(value)
 			expect(list.each.map(&:value)).to be(:empty?)
 		end
 
+		it "returns nil when removing an item that is not in the list" do
+			item = Item.new(1)
+
+			expect(list.remove?(item)).to be_nil
+		end
+
 		it "can't remove an item twice" do
 			item = Item.new(1)
 

test/async/node.rb

@@ -288,6 +288,17 @@
 
 			node.cancel
 		end
+
+		it "forwards the cause to child tasks" do
+			cause = RuntimeError.new("boom")
+
+			middle = Async::Node.new(node, annotation: "middle")
+			child = Async::Node.new(middle, annotation: "child")
+
+			expect(child).to receive(:cancel).with(false, cause: cause)
+
+			middle.cancel(cause: cause)
+		end
 	end
 
 	with "#terminate" do

test/async/promise.rb

@@ -52,6 +52,17 @@
 		end
 	end
 
+	with "#wait" do
+		it "waits until a pending promise resolves before timeout" do
+			reactor.async do
+				sleep(0.001)
+				promise.resolve(:done)
+			end
+
+			expect(promise.wait(timeout: 1)).to be == :done
+		end
+	end
+
 	with "#reject" do
 		it "can be rejected with an exception" do
 			error = StandardError.new("test error")