[async] Add `StructuredTaskScope`
diff --git a/pkgs/async/lib/async.dart b/pkgs/async/lib/async.dart index 1b489aa..0b784f4 100644 --- a/pkgs/async/lib/async.dart +++ b/pkgs/async/lib/async.dart
@@ -20,6 +20,7 @@ export 'src/delegate/stream_consumer.dart'; export 'src/delegate/stream_sink.dart'; export 'src/delegate/stream_subscription.dart'; +export 'src/disposable.dart'; export 'src/future_group.dart'; export 'src/lazy_stream.dart'; export 'src/null_stream_sink.dart'; @@ -39,5 +40,6 @@ export 'src/stream_splitter.dart'; export 'src/stream_subscription_transformer.dart'; export 'src/stream_zip.dart'; +export 'src/structured_task_scope.dart'; export 'src/subscription_stream.dart'; export 'src/typed_stream_transformer.dart';
diff --git a/pkgs/async/lib/src/disposable.dart b/pkgs/async/lib/src/disposable.dart new file mode 100644 index 0000000..97a6cf3 --- /dev/null +++ b/pkgs/async/lib/src/disposable.dart
@@ -0,0 +1,84 @@ +// Copyright (c) 2026, the Dart project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. + +import 'dart:async'; + +// TODO: Disposable and SyncDisposable should be moved to the SDK +// (e.g. dart:core or dart:async). + +/// A resource or component that participates in cooperative shutdown. +/// +/// A disposable must be registered after creation before any async yield point. +/// An isolate shutdown request (such as during Hot Restart) can arrive during +/// any asynchronous yield point. The disposable must be either registered +/// to the isolate or to an owner disposable already registered to an isolate. +/// +/// If the [Disposable] is the outermost resource of an isolate (or the root +/// application service), register it upon creation: +/// ```dart +/// class AppService implements Disposable { +/// AppService() { +/// Isolate.registerShutdownDisposable(this); +/// } +/// } +/// ``` +/// +/// If the [Disposable] is created inside an enclosing parent that is already +/// registered, register it with that parent before any asynchronous yield +/// point occurs: +/// ```dart +/// Future<void> fetchData() async { +/// final client = IsolateHttpClient(); +/// _children.add(client); // Register with outer parent before `await` +/// try { +/// await client.get(...); // Protected yield point +/// } finally { +/// await client.dispose(); +/// _children.remove(client); +/// } +/// } +/// ``` +/// +/// If a short-lived [Disposable] is used inside a local [try]/[finally] block +/// and has no enclosing parent, register and unregister directly via `Isolate`: +/// ```dart +/// Future<void> runStandaloneTask() async { +/// final client = IsolateHttpClient(); +/// Isolate.registerShutdownDisposable(client); +/// try { +/// await client.get(...); // Protected yield point +/// } finally { +/// await client.dispose(); +/// Isolate.unregisterShutdownDisposable(client); +/// } +/// } +/// ``` +abstract interface class Disposable { + /// Disposes this resource. + /// + /// Implementations must be idempotent and safe to call more than once. + /// Calling [dispose] on an already disposed object should return immediately + /// (or return the pending [Future] if disposal is in progress). + /// + /// If disposal requires asynchronous work, this method returns a [Future] + /// that completes when disposal is finished. If disposal is synchronous, + /// this method returns `null` (or implements [SyncDisposable]). + /// + /// Disposables may throw errors or complete the returned future with an + /// error. If this happens during cooperative isolate shutdown, it is an + /// unrecoverable error. + FutureOr<void> dispose(); +} + +/// A [Disposable] that can be disposed synchronously. +abstract interface class SyncDisposable implements Disposable { + /// Disposes this resource synchronously. + /// + /// Implementations must be idempotent and safe to call more than once. + /// + /// Disposables may throw errors. If this happens during cooperative isolate + /// shutdown, it is an unrecoverable error. + @override + void dispose(); +}
diff --git a/pkgs/async/lib/src/structured_task_scope.dart b/pkgs/async/lib/src/structured_task_scope.dart new file mode 100644 index 0000000..bd718c5 --- /dev/null +++ b/pkgs/async/lib/src/structured_task_scope.dart
@@ -0,0 +1,258 @@ +// Copyright (c) 2026, the Dart project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. + +import 'dart:async'; + +import 'cancelable_operation.dart'; +import 'disposable.dart'; + +/// A scope for structured concurrency that manages concurrent tasks and +/// resources. +/// +/// A [StructuredTaskScope] coordinates the lifecycles of child operations +/// ([CancelableOperation]) and resources ([Disposable]). +/// +/// Tasks are added to the scope via [fork] or [forkOperation]. Resources can +/// be attached via [attachDisposable]. +/// +/// If any task fails and [shutdownOnFailure] is `true` (the default), the scope +/// automatically cancels all remaining active tasks and disposes attached +/// resources. +/// +/// Calling [dispose] (or using [StructuredTaskScope.run]) guarantees that all +/// child tasks and attached resources are cleanly shut down and awaited before +/// the scope finishes. +class StructuredTaskScope implements Disposable { + /// Whether an error in a child task automatically triggers scope shutdown. + final bool shutdownOnFailure; + + final Set<CancelableOperation<dynamic>> _operations = {}; + final Set<Disposable> _disposables = {}; + + bool _closed = false; + bool _disposed = false; + bool _isShuttingDown = false; + + Object? _firstError; + StackTrace? _firstStackTrace; + + /// Creates a new [StructuredTaskScope]. + /// + /// If [shutdownOnFailure] is `true` (the default), an error in any child + /// operation will trigger automatic cancellation of all sibling operations + /// and attached disposables. + StructuredTaskScope({this.shutdownOnFailure = true}); + + /// Whether the scope is closed for adding new tasks. + bool get isClosed => _closed; + + /// Whether the scope has been disposed. + bool get isDisposed => _disposed; + + /// Whether the scope is currently shutting down due to an error or + /// cancellation. + bool get isShuttingDown => _isShuttingDown; + + /// The first error that triggered scope shutdown, if any. + Object? get error => _firstError; + + /// The stack trace for [error], if any. + StackTrace? get stackTrace => _firstStackTrace; + + /// Runs [computation] within a new [StructuredTaskScope]. + /// + /// Automatically disposes the scope when [computation] completes or throws, + /// guaranteeing that all spawned operations are cancelled and awaited. + static Future<R> run<R>( + FutureOr<R> Function(StructuredTaskScope scope) computation, { + bool shutdownOnFailure = true, + }) async { + final scope = StructuredTaskScope(shutdownOnFailure: shutdownOnFailure); + try { + final result = await computation(scope); + await scope.join(); + return result; + } catch (error, stackTrace) { + await scope._shutdown(error, stackTrace); + scope.throwIfFailed(); + rethrow; + } finally { + await scope.dispose(); + } + } + + /// Forks a new task in this scope. + /// + /// The task executes [computation]. If [onCancel] is provided, it will be + /// invoked if the operation is cancelled before it completes. + /// + /// Throws [StateError] if the scope is closed, disposed, or shutting down. + CancelableOperation<T> fork<T>( + FutureOr<T> Function() computation, { + FutureOr<void> Function()? onCancel, + }) { + if (_closed || _disposed || _isShuttingDown) { + throw StateError('Cannot fork tasks in a closed, disposed, ' + 'or shutting down StructuredTaskScope.'); + } + + final completer = CancelableCompleter<T>(onCancel: onCancel); + final operation = forkOperation(completer.operation); + + Future.sync(computation).then( + (value) { + if (!completer.isCanceled && !completer.isCompleted) { + completer.complete(value); + } + }, + onError: (Object error, StackTrace stackTrace) { + if (!completer.isCanceled && !completer.isCompleted) { + completer.completeError(error, stackTrace); + } + }, + ); + + return operation; + } + + /// Forks an existing [CancelableOperation] into this scope. + /// + /// If [shutdownOnFailure] is `true` and the operation completes with an + /// error, the scope will shut down and cancel all sibling tasks. + CancelableOperation<T> forkOperation<T>(CancelableOperation<T> operation) { + if (_closed || _disposed || _isShuttingDown) { + throw StateError('Cannot fork tasks in a closed, disposed, ' + 'or shutting down StructuredTaskScope.'); + } + + _operations.add(operation); + + operation.valueOrCancellation().then( + (_) { + _operations.remove(operation); + }, + onError: (Object error, StackTrace stackTrace) { + _operations.remove(operation); + if (shutdownOnFailure && !operation.isCanceled) { + _shutdown(error, stackTrace); + } + }, + ); + + return operation; + } + + /// Attaches a [Disposable] resource to this scope. + /// + /// When the scope is shut down or disposed, [disposable.dispose()] will be + /// called and awaited. + /// + /// Throws [StateError] if the scope is closed, disposed, or shutting down. + void attachDisposable(Disposable disposable) { + if (_closed || _disposed || _isShuttingDown) { + throw StateError('Cannot attach disposables to a closed, disposed, ' + 'or shutting down StructuredTaskScope.'); + } + _disposables.add(disposable); + } + + /// Detaches a previously attached [Disposable] resource from this scope. + bool detachDisposable(Disposable disposable) { + return _disposables.remove(disposable); + } + + /// Closes the scope so that no new tasks or disposables can be added. + void close() { + _closed = true; + } + + /// Waits for all active tasks in the scope to settle. + /// + /// If [shutdownOnFailure] is `true` and any task failed, this will throw the + /// error after active tasks have settled. + Future<void> join() async { + _closed = true; + while (_operations.isNotEmpty) { + final futures = _operations + .map((op) => op.valueOrCancellation().catchError((_, __) => null)) + .toList(); + await Future.wait(futures); + } + throwIfFailed(); + } + + /// Throws the first error that occurred in a child task if the scope failed. + void throwIfFailed() { + final err = _firstError; + if (err != null) { + Error.throwWithStackTrace(err, _firstStackTrace ?? StackTrace.current); + } + } + + /// Initiates shutdown of the scope due to an error or cancellation signal. + Future<void> _shutdown([Object? error, StackTrace? stackTrace]) async { + if (_firstError == null && error != null) { + _firstError = error; + _firstStackTrace = stackTrace; + } + + if (_isShuttingDown) return; + _isShuttingDown = true; + _closed = true; + + // Phase 1: Cancel all active operations and await their cancellation hooks. + final opCancelFutures = <Future<void>>[]; + for (final op in List.of(_operations)) { + if (!op.isCompleted) { + final cancelResult = op.cancel(); + opCancelFutures + .add(cancelResult.catchError((Object error, StackTrace stackTrace) { + if (_firstError == null) { + _firstError = error; + _firstStackTrace = stackTrace; + } + })); + } + } + if (opCancelFutures.isNotEmpty) { + await Future.wait(opCancelFutures); + } + + // Phase 2: Now that child tasks are stopped, dispose attached resources. + final disposeFutures = <Future<void>>[]; + for (final disposable in List.of(_disposables)) { + try { + final disposeResult = disposable.dispose(); + if (disposeResult is Future<void>) { + disposeFutures.add( + disposeResult.catchError((Object error, StackTrace stackTrace) { + if (_firstError == null) { + _firstError = error; + _firstStackTrace = stackTrace; + } + })); + } + } catch (error, stackTrace) { + if (_firstError == null) { + _firstError = error; + _firstStackTrace = stackTrace; + } + } + } + if (disposeFutures.isNotEmpty) { + await Future.wait(disposeFutures); + } + } + + /// Disposes this scope, cancelling all active tasks and disposing all + /// attached resources. + @override + FutureOr<void> dispose() async { + if (_disposed) return; + await _shutdown(); + _disposed = true; + _operations.clear(); + _disposables.clear(); + } +}
diff --git a/pkgs/async/test/structured_task_scope_test.dart b/pkgs/async/test/structured_task_scope_test.dart new file mode 100644 index 0000000..bbf45bd --- /dev/null +++ b/pkgs/async/test/structured_task_scope_test.dart
@@ -0,0 +1,216 @@ +// Copyright (c) 2026, the Dart project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. + +import 'dart:async'; + +import 'package:async/async.dart'; +import 'package:test/test.dart'; + +class _TestSyncDisposable implements SyncDisposable { + bool isDisposed = false; + + @override + void dispose() { + isDisposed = true; + } +} + +class _TestAsyncDisposable implements Disposable { + bool isDisposed = false; + final Completer<void> completer = Completer<void>(); + + @override + Future<void> dispose() async { + isDisposed = true; + await completer.future; + } +} + +void main() { + group('StructuredTaskScope', () { + test('runs computation and returns result', () async { + final result = await StructuredTaskScope.run((scope) async { + final task1 = scope.fork(() async => 21); + final task2 = scope.fork(() async => 21); + return (await task1.value) + (await task2.value); + }); + expect(result, equals(42)); + }); + + test('cancels sibling tasks when a task fails with shutdownOnFailure', + () async { + var task2Cancelled = false; + + expect( + StructuredTaskScope.run((scope) async { + scope.fork(() async { + await Future.delayed(const Duration(milliseconds: 10)); + throw StateError('Task 1 failed'); + }); + + scope.fork( + () async { + await Future.delayed(const Duration(milliseconds: 200)); + return 42; + }, + onCancel: () { + task2Cancelled = true; + }, + ); + + await scope.join(); + }), + throwsA(isA<StateError>() + .having((e) => e.message, 'message', 'Task 1 failed')), + ); + + await Future.delayed(const Duration(milliseconds: 50)); + expect(task2Cancelled, isTrue); + }); + + test('attaches and disposes attached Disposable resources', () async { + final syncDisposable = _TestSyncDisposable(); + final asyncDisposable = _TestAsyncDisposable(); + + final scope = StructuredTaskScope(); + scope.attachDisposable(syncDisposable); + scope.attachDisposable(asyncDisposable); + + expect(syncDisposable.isDisposed, isFalse); + expect(asyncDisposable.isDisposed, isFalse); + + asyncDisposable.completer.complete(); + await scope.dispose(); + + expect(syncDisposable.isDisposed, isTrue); + expect(asyncDisposable.isDisposed, isTrue); + expect(scope.isDisposed, isTrue); + }); + + test('throws StateError if fork or attachDisposable called when closed', + () async { + final scope = StructuredTaskScope(); + scope.close(); + + expect(() => scope.fork(() => 42), throwsStateError); + expect(() => scope.attachDisposable(_TestSyncDisposable()), + throwsStateError); + }); + + test('detaches Disposable', () async { + final syncDisposable = _TestSyncDisposable(); + final scope = StructuredTaskScope(); + scope.attachDisposable(syncDisposable); + expect(scope.detachDisposable(syncDisposable), isTrue); + + await scope.dispose(); + expect(syncDisposable.isDisposed, isFalse); + }); + + test('awaits operation cancellation before disposing disposables', + () async { + final events = <String>[]; + final scope = StructuredTaskScope(); + + final disposable = _TrackingDisposable(events); + scope.attachDisposable(disposable); + + scope.fork( + () async { + await Completer<void>().future; + }, + onCancel: () async { + events.add('task_cancelling'); + await Future.delayed(const Duration(milliseconds: 10)); + events.add('task_cancelled'); + }, + ); + + await scope.dispose(); + + expect(events, + equals(['task_cancelling', 'task_cancelled', 'disposable_disposed'])); + }); + + test( + 'awaits ongoing cancellation for externally cancelled operation before ' + 'disposing disposables', () async { + final events = <String>[]; + final scope = StructuredTaskScope(); + + final disposable = _TrackingDisposable(events); + scope.attachDisposable(disposable); + + final completer = CancelableCompleter<void>( + onCancel: () async { + events.add('task_cancelling'); + await Future.delayed(const Duration(milliseconds: 50)); + events.add('task_cancelled'); + }, + ); + + scope.forkOperation(completer.operation); + + // Cancel operation externally so operation.isCanceled becomes true + completer.operation.cancel(); + + // Immediately dispose scope + await scope.dispose(); + + expect( + events, + equals(['task_cancelling', 'task_cancelled', 'disposable_disposed']), + ); + }); + + test( + 'captures error thrown during cancellation callback during dispose ' + 'shutdown', () async { + final scope = StructuredTaskScope(); + + final completer = CancelableCompleter<void>( + onCancel: () async { + throw StateError('Cancellation failed'); + }, + ); + + scope.forkOperation(completer.operation); + + await scope.dispose(); + + expect(scope.error, isA<StateError>()); + }); + + test('handles forkOperation with already completed operation', () async { + final scope = StructuredTaskScope(); + final op = CancelableOperation.fromValue(42); + + final forked = scope.forkOperation(op); + expect(await forked.value, equals(42)); + + await scope.dispose(); + }); + + test('handles forkOperation with already cancelled operation', () async { + final scope = StructuredTaskScope(); + final completer = CancelableCompleter<int>(); + completer.operation.cancel(); + + final forked = scope.forkOperation(completer.operation); + expect(forked.isCanceled, isTrue); + + await scope.dispose(); + }); + }); +} + +class _TrackingDisposable implements Disposable { + final List<String> events; + _TrackingDisposable(this.events); + + @override + void dispose() { + events.add('disposable_disposed'); + } +}