Structured Concurrency with Scope

Fluent API for Task Coordination

"Structured concurrency is to concurrent programming what structured programming was to sequential programming." — Ron Pressler, Project Loom Lead

While Par combinators provide simple parallel execution, Scope offers a more flexible, fluent API for structured concurrent computations. It wraps Java 25's StructuredTaskScope with functional result handling.

┌──────────────────────────────────────────────────────────────────┐
│                          Scope API                               │
│                                                                  │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐   ┌──────────────┐   │
│  │  create  │ → │   fork   │ → │ configure│ → │     join     │   │
│  │  scope   │   │  tasks   │   │ timeout  │   │   results    │   │
│  └──────────┘   └──────────┘   └──────────┘   └──────────────┘   │
│                                                                  │
│  Factory methods: allSucceed, anySucceed, firstComplete,         │
│                   accumulating, withJoiner                       │
└──────────────────────────────────────────────────────────────────┘

Scope Usage Patterns

import org.higherkindedj.hkt.vtask.Scope;
import org.higherkindedj.hkt.vtask.VTask;

// All-succeed: wait for all tasks to complete successfully
VTask<List<String>> allResults = Scope.<String>allSucceed()
    .fork(VTask.of(() -> fetchUser(id)))
    .fork(VTask.of(() -> fetchProfile(id)))
    .fork(VTask.of(() -> fetchPreferences(id)))
    .join();

// Any-succeed: first successful result wins
VTask<String> fastest = Scope.<String>anySucceed()
    .fork(VTask.of(() -> fetchFromServerA()))
    .fork(VTask.of(() -> fetchFromServerB()))
    .fork(VTask.of(() -> fetchFromServerC()))
    .join();

// First-complete: whichever answers first decides, success or failure
VTask<Data> racing = Scope.<Data>firstComplete()
    .fork(riskyFastPath())
    .fork(reliableSlowPath())
    .join();

// With timeout
VTask<List<String>> withTimeout = Scope.<String>allSucceed()
    .timeout(Duration.ofSeconds(5))
    .fork(slowTask1())
    .fork(slowTask2())
    .join();

// Fork multiple tasks at once
List<VTask<Integer>> tasks = ids.stream()
    .map(id -> VTask.of(() -> processId(id)))
    .toList();

VTask<List<Integer>> all = Scope.<Integer>allSucceed()
    .forkAll(tasks)
    .join();

A timeout cancels the scope when it expires: subtasks still running are interrupted, and the run fails with a TimeoutException once they have stopped. A subtask that ignores interruption, such as one spinning in a loop, holds that failure back until it finishes. The scope is opened on the thread that runs the VTask, so its subtasks see that thread's ScopedValue bindings, such as the request context. named(...) gives the scope a name, which a thread dump in JSON format shows.

Scope Factory Methods

When to Use Each Joiner

allSucceed is the default choice when you need results from multiple independent operations:

// Fetch user data from three services - need all of them
VTask<List<UserData>> userData = Scope.<UserData>allSucceed()
    .fork(VTask.of(() -> authService.getPermissions(userId)))
    .fork(VTask.of(() -> profileService.getProfile(userId)))
    .fork(VTask.of(() -> prefsService.getPreferences(userId)))
    .join();

anySucceed is ideal for redundant requests where any response will do:

// Try multiple mirrors - first success wins
VTask<Package> download = Scope.<Package>anySucceed()
    .fork(VTask.of(() -> fetchFrom(mirror1)))
    .fork(VTask.of(() -> fetchFrom(mirror2)))
    .fork(VTask.of(() -> fetchFrom(mirror3)))
    .join();

firstComplete captures the first result regardless of success or failure, and cancels the tasks still running:

// Whichever path answers first decides, success or failure; the other is cancelled
VTask<Result> result = Scope.<Result>firstComplete()
    .fork(VTask.of(() -> fastButRisky()))
    .fork(VTask.of(() -> slowButSafe()))
    .join();

For a fallback that answers when the fast path fails, use anySucceed.

Running a Scope Again

A Scope is a description, like the VTask its join() returns. Each run of that VTask opens a StructuredTaskScope of its own, with a new Joiner. So you can run it again, retry it, or run it from several threads at once, and each run joins only the subtasks it forked.

Here the quote service is busy on its first call, and Retry runs the same prices task again:

    AtomicInteger quoteCalls = new AtomicInteger();
    VTask<String> quote =
        VTask.of(
            () -> {
              if (quoteCalls.incrementAndGet() == 1) {
                throw new IllegalStateException("quote service busy");
              }
              return "quote";
            });
    VTask<List<String>> prices =
        Scope.<String>allSucceed().fork(quote).fork(VTask.succeed("rates")).join();

    List<String> retried =
        Retry.retryTask(prices, RetryPolicy.fixed(3, Duration.ofMillis(100))).run();
    // [quote, rates]
    List<String> runAgain = prices.run();
    // [quote, rates]

The first attempt fails. The second forks both subtasks again and joins only those, so the first attempt's failure cannot fail it, and it returns [quote, rates]. Running prices once more gives the same list. A retry runs every subtask again, rates included, so retry a Scope whose subtasks are safe to repeat.

Safe Result Handling

Scope provides functional result wrappers that capture failures instead of throwing:

// Get result as Try
VTask<Try<List<String>>> trySafe = Scope.<String>allSucceed()
    .fork(task1())
    .fork(task2())
    .joinSafe();

// Get result as Either
VTask<Either<Throwable, List<String>>> eitherResult = Scope.<String>allSucceed()
    .fork(task1())
    .fork(task2())
    .joinEither();

// Get result as Maybe (Nothing on failure)
VTask<Maybe<List<String>>> maybeResult = Scope.<String>allSucceed()
    .fork(task1())
    .fork(task2())
    .joinMaybe();

These wrappers are useful when you want to handle failures within your VTask pipeline rather than at the execution boundary.

Error Accumulation with ScopeJoiner

The accumulating joiner is particularly powerful for validation scenarios where you want to report all errors at once rather than failing on the first error.

import org.higherkindedj.hkt.vtask.Scope;
import org.higherkindedj.hkt.validated.Validated;

// Define how to convert exceptions to your error type
record ValidationError(String field, String message) {
    static ValidationError from(Throwable t) {
        return new ValidationError("unknown", t.getMessage());
    }
}

// Validate multiple fields in parallel, collecting all errors
VTask<Validated<List<ValidationError>, List<String>>> validation =
    Scope.<ValidationError, String>accumulating(ValidationError::from)
        .fork(VTask.of(() -> validateUsername(input.username())))
        .fork(VTask.of(() -> validateEmail(input.email())))
        .fork(VTask.of(() -> validatePassword(input.password())))
        .join();

// Handle the result
Validated<List<ValidationError>, List<String>> result = validation.run();

result.fold(
    errors -> {
        // All errors collected, not just the first
        errors.forEach(err ->
            System.out.println(err.field() + ": " + err.message()));
        return null;
    },
    successes -> {
        System.out.println("All validations passed: " + successes);
        return null;
    }
);

Why Error Accumulation Matters

Traditional fail-fast behaviour returns only the first error encountered. This frustrates users who must fix one error, resubmit, discover another error, fix it, resubmit again, and so on. Error accumulation runs all validations in parallel and collects every failure, enabling you to report all problems at once:

// User submits a form with multiple invalid fields
// Instead of: "Username is too short" (fix, resubmit)
//             "Email is invalid" (fix, resubmit)
//             "Password needs a number" (fix, resubmit)
//
// You can show all three errors immediately

ScopeJoiner Interface

ScopeJoiner is a joining policy. Scope.withJoiner takes one, and its joiner() builds a Java 25 StructuredTaskScope.Joiner when you open a scope yourself:

import org.higherkindedj.hkt.vtask.ScopeJoiner;

// Access joiner factories
ScopeJoiner<String, List<String>> allSucceed = ScopeJoiner.allSucceed();
ScopeJoiner<String, String> anySucceed = ScopeJoiner.anySucceed();
ScopeJoiner<String, String> firstComplete = ScopeJoiner.firstComplete();
ScopeJoiner<String, Validated<List<ValidationError>, List<String>>> accum =
    ScopeJoiner.accumulating(ValidationError::from);

// Use with a Scope
VTask<List<String>> result = Scope.withJoiner(allSucceed)
    .fork(task1())
    .fork(task2())
    .join();

// Open a StructuredTaskScope yourself: a new Joiner for each scope
try (var scope = StructuredTaskScope.open(allSucceed.joiner())) {
    scope.fork(() -> "first");
    scope.fork(() -> "second");
    List<String> both = scope.join();
}

A ScopeJoiner holds its configuration, not a scope's state. Each call to joiner() returns a new Joiner, because a StructuredTaskScope needs one that no other scope has used. So one ScopeJoiner can serve any number of scopes.

ScopeJoiner Methods

Previous: VTask Next: Resource Management