Chains

A chain continues a job with its result. Here the app loads a number as text, parses it and saves it. then returns a new job that starts once its source has succeeded, children and cleanup included, and gives its callback the value and a context of its own:

final loaded = Job<String>((ctx) => ctx.join(loadText));
final parsed = loaded.then<int>((ctx, text) => int.parse(text));
final saved = parsed.then<void>(
  (ctx, number) => ctx.join(() => saveNumber(number)),
);

await saved.value;

The output blocks of this page hold:

• loaded:, parsed:, saved:, shown: the outcome of that job;
• parent:, child: the outcomes after tail.cancel() in A continuation of a child.

Cancelling a chain

The text is still loading when the user leaves the screen, and the screen cancels the last link:

// The user left the screen while the text was loading.
await saved.cancel();
print('loaded: ${loaded.outcome}');
print('parsed: ${parsed.outcome}');
print('saved: ${saved.outcome}');

loaded: Cancelled(chain)
parsed: Cancelled(chain)
saved: Cancelled(manual)

A cancellation travels along the chain both ways. It asks the unfinished links before the cancelled one to cancel, and it cancels the links after it. Each link it reaches gets ChainCancelReason, with the Cancelled of the link it came from in cause:

You cancel	Asked to cancel, if unfinished	Cancelled once the link you cancel accepts
loaded	none	parsed and saved
parsed	loaded	saved
saved	loaded and parsed	none
parsed, made on a handle of loaded.shared()	the handle, which ends at once; loaded is not asked	saved

A link that refuses stops the cancellation there: cancel a loaded made with cancellable: false, and parsed and saved go on. Finished links keep their outcomes. The handle of the last row is the subject of Two continuations of one source.

The links after a source hear of its cancellation through whenCancelled: a continuation and a handle of shared register there when they are made. A listener registered on the source before them may find them not yet cancelled.

A continuation starts only when its source has finished, so a running link has nothing unfinished behind it. While parsed runs, loaded is over, and loaded.cancel() does nothing: parsed goes on, and saved after it. While saved runs, cancelling loaded or parsed leaves it alone. To stop a running link, cancel that link or one after it.

await saved.cancel() returns once the unfinished links before saved have ended, with their children and cleanup, and once saved has ended the work and cleanup it started. A source keeps its own rules: cancellable: false refuses the request, and uncancellable holds it back. A cancelled continuation still waits for such a source and then ends without calling its callback. Meanwhile it is isCancelled but not isFinished. Its outcome has started: false if it was cancelled while it waited for its source.

Two continuations of one source

The screen shows the loaded text and, beside it, the number parsed from it. The number leaves the screen while the text is still loading, and the text stays.

The first attempt

// The first attempt: the second continuation throws.
final loaded = Job<String>((ctx) => ctx.join(loadText));
final parsed = loaded.then<int>((ctx, text) => int.parse(text));
final shown = loaded.then<void>((ctx, text) => showText(text));

Bad state: Job() already has a continuation: a job is continued once; call shared() for each branch

A job is continued once, and the second then throws before anything is made. A continuation asks its unfinished source to cancel, as any link asks the links before it, and a source does not know how many continuations wait for it. With two of them, the number leaving would cancel loaded, and the text would never reach the screen that stayed.

A handle for each branch

final loaded = Job<String>((ctx) => ctx.join(loadText));
final parsed = loaded.shared().then<int>((ctx, text) => int.parse(text));
final shown = loaded.shared().then<void>((ctx, text) => showText(text));

// The number left the screen while the text was loading.
await parsed.cancel();
await shown.done;
print('parsed: ${parsed.outcome}');
print('loaded: ${loaded.outcome}');
print('shown: ${shown.outcome}');

parsed: Cancelled(manual)
loaded: Done(21)
shown: Done(null)

shared() makes a handle that ends with the outcome of loaded and cannot reach it. parsed asks its source to cancel, and its source is the handle: the handle ends at once and leaves loaded be. Each call makes another handle, so each branch has a source of its own. A cancellation that loaded accepts still reaches every handle, and each ends Cancelled(chain) without waiting for the cleanup of loaded.

A failure of loaded reaches every handle still there, and it is answered for once however many of them carry it. If parsed and shown were both cancelled before loaded failed, no handle is left to carry the failure. loaded then answers for it itself, unless something observes it, as the source does in the second row of When a link fails. It answers in onUnanswered of its observer if the observer mixes in JobAnswerer, otherwise in the zone loaded was made in, not the one then was called in.

Jobs that wait for the source

final loaded = Job<String>((ctx) => ctx.join(loadText));
final parsed = Job<int>(
  (ctx) async => int.parse(await ctx.abandonable(() => loaded.value)),
);
final shown = Job<void>(
  (ctx) async => showText(await ctx.abandonable(() => loaded.value)),
);

// The number left the screen while the text was loading.
await parsed.cancel();
await shown.done;
print('parsed: ${parsed.outcome}');
print('loaded: ${loaded.outcome}');
print('shown: ${shown.outcome}');

parsed: Cancelled(manual)
loaded: Done(21)
shown: Done(null)

parsed and shown are jobs of their own that wait for loaded, not its continuations, so a cancellation of one of them stops its own wait and goes no further. loaded belongs to whoever made it, and that code cancels it when nobody needs the text. A failure of loaded that comes after parsed has left reaches parsed as the late error of an action ctx.abandonable stopped waiting for: see Where each error goes.

When a link fails

A failed link passes its error and stack on without calling the next callback. The failure comes down the chain and ends every link after it, so whoever observes the last link through value, done or ignoreFailure handles a failure from anywhere up the chain. The JobObserver of the last link hears it as that link’s own failure.

Where an error goes depends on what failed:

What failed	Where the error goes
The body of a link or the callback of a continuation	Down the chain, to whoever observes the last link. Unobserved, the last link answers for it: in onUnanswered of its observer if the observer mixes in JobAnswerer, otherwise in the zone the last then was called in
A source that refused the cancellation of its continuation and failed later	Answered for by the source, unless something observes it: in onUnanswered of its observer if the observer mixes in JobAnswerer, otherwise in the zone the source was made in. The cancelled continuation does not forward the error, so it does not observe it either
Work of ctx.unattended or a disposer, which no outcome carries	The observer of the link where it happened, then, unless the observer catches it in onUnanswered, the zone the link was made in. For a continuation, that is the zone then was called in

What the callback returns

The callback returns a value or a future, and the continuation waits for the future. Here a job of its own saves the number.

The first attempt

// The first attempt: the continuation does not wait for a job.
final stored = parsed.then<void>((ctx, number) {
  final saving = Job<void>((ctx) => ctx.join(() => saveNumber(number)));
  return saving;
});

The code compiles. The void_checks lint, which package:lints turns on, marks this callback, but not the arrow form (ctx, number) => saving, and the compiler marks neither. stored does not wait for saving. This is what happens:

1. then waits for a future, and a Job is not a future. The continuation takes saving for its value and ends Done at once.
2. saving starts by itself and goes on alone, and nobody waits for it.
3. A saving made with Job.deferred would never start, because nothing here starts it. For a void continuation the core checks this, and stored fails with an ArgumentError. A continuation of a wider type takes the job for its value: a job may be the value it means to hand on.

A job the continuation runs as its child

final stored = parsed.then<void>((ctx, number) {
  final saving = Job.deferred<void>(
    (ctx) => ctx.join(() => saveNumber(number)),
  );
  return ctx.run(saving);
});

ctx.run starts saving and returns a future of its value, and the continuation waits for it.

A continuation is not a child

Each continuation is a root job of the core, wherever then is called. Made in the body of another job, it is not that job’s child, and that job does not wait for it. It has an optional observer argument and inherits neither the source’s observer nor anything an engine attaches to the source.

	A child	A continuation
Made by	Job.deferred	then
Starts	when its parent runs it with ctx.run	by itself, once its source has succeeded
Waited for by	its parent	nobody: it is a root job
ctx.run takes it	yes, it is a ChildJob	no, it is a plain Job

then does not start its source. A chain hung off a Job.deferred stays unstarted, every link of it, until the source is started: by start(), or by ctx.run(source) in the body that adopts it.

What the source hands out

opened opens the rows and hands them out, and archived writes them to an archive:

final opened = Job<Source>(
  (ctx) => ctx.abandonable(openRows, discard: (rows) => rows.close()),
);
final archived = opened.then<void>((ctx, rows) async {
  ctx.onDispose(rows.close);
  await ctx.join(() => writeArchive([rows]));
});

Cleanup the source registered has run by the time the continuation receives its value: a resource closed by the source’s onDispose is already closed then. discard is the other way round: it runs only when its job does not end Done. opened ended Done, so its discard did not run and never will. The receiver is the continuation: rows comes as its argument, and it registers the closing in its own body, on the first line.

One case has no receiver at all. archived.cancel() while opened is still opening ends archived without calling its callback, so there is no body to register anything in. The cancellation goes on to opened, and who closes the rows depends on what opened does with it:

How the chain goes	Who closes the rows
archived receives the rows	archived, through ctx.onDispose
archived is cancelled while opened opens, and opened takes the cancellation	opened, through its discard, when the rows arrive
The same, and opened refuses it: it was made with cancellable: false	The caller, through opened: (await opened.value).close()
The same, and opened has finished by the time the request arrives	The caller, through opened, the same way

The caller closes the rows through opened the way the body of a group closes the value of a branch that refuses the group’s stop.

A source continued in several branches through shared() hands every branch the same rows. Each handle is cancelled on its own, but the value is one: a branch that registers the closing closes the rows under the other, and the second branch to do it closes them twice. With branches, the rows belong to the code that waits for all of them:

final opened = Job<Source>(
  (ctx) => ctx.abandonable(openRows, discard: (rows) => rows.close()),
);
final archived = opened.shared().then<void>(
  (ctx, rows) => ctx.join(() => writeArchive([rows])),
);
final warmed = opened.shared().then<void>(
  (ctx, rows) => ctx.join(() => cache.warm(rows)),
);
await [archived.done, warmed.done].wait;
if (await opened.done case Done(:final value)) value.close();

A handle does not cancel opened, so the closing waits for opened itself as well as for the branches. If opened ends otherwise than Done, the rows are closed by its discard or were never opened.

A continuation of a child

The parent needs the rows; the report on them may finish after the parent has:

final child = Job.deferred<int>((ctx) => ctx.abandonable(load));
final tail = child.then<void>((ctx, rows) => report(rows));

final parent = Job<void>((ctx) async {
  // The source is a child: the parent starts it and waits for it.
  ctx.log(await ctx.run(child));
});

await parent.value; // Done, whatever tail is doing.
await tail.value; // tail is yours to observe.

The parent waits for its children, not for what hangs off them. Once ctx.run(child) has returned and the body ends, the parent finishes Done while a slow continuation is still running. What stops tail depends on whether it still waits for the child:

What	While tail waits for the child	While tail runs
Cancelling the parent	Stops it: the parent cancels the child, and the child’s cancellation travels forward as ChainCancelReason	Does not reach it
tail.cancel()	Stops it	Stops it
ctx.onCancel(() => tail.cancel().ignore()) in a parent body that still runs	Stops it	Stops it

The other way round, the continuation does reach its source while the source runs. tail.cancel() while the child still loads travels back along the chain, as Cancelling a chain describes, and asks the child to cancel. ctx.run(child) throws the child’s cancellation into the parent’s body, and the parent ends Cancelled too, with BodyCancelReason:

parent: Cancelled(body: Job(): Cancelled(chain)), child: Cancelled(chain)

A failure in the continuation is nobody’s business but its own. Observe it through value, done or ignoreFailure; an unobserved one is answered for by the continuation, in onUnanswered of its observer or in the zone then was called in, after the parent has already finished. Its outcome comes back to you, not to the parent.

Steps in a row

A job loads the rows and then reports them, and it must not end before the report is done.

The first attempt

then reads like the word for “and then”, so the report becomes a continuation, and the body adopts both:

// The first attempt: ctx.run does not take a continuation.
final child = Job.deferred<int>((ctx) => ctx.abandonable(load));
final tail = child.then<void>((ctx, rows) => report(rows));

final parent = Job<void>((ctx) async {
  ctx.log(await ctx.run(child));
  await ctx.run(tail);
});

The second ctx.run does not compile:

The argument type 'Job<void>' can't be assigned to the parameter type 'ChildJob<dynamic>'.

ctx.run takes a ChildJob, a job its parent starts, and a continuation starts itself once its source finishes: then returns a plain Job. The length of the chain changes nothing: child.then(...).then(...) hangs one continuation off another, and no link of it is a child.

Two children in a row

final parent = Job<void>((ctx) async {
  final rows = await ctx.run(
    Job.deferred<int>((ctx) => ctx.abandonable(load)),
  );
  await ctx.run(Job.deferred<void>((ctx) => ctx.join(() => report(rows))));
});

A sequence that belongs to the operation is children. The second ctx.run starts once the first has returned its value, the parent waits for both, and its cancellation reaches whichever of them is running. A chain is for the opposite case, a step that outlives the operation or runs outside it: A continuation of a child.

A source that waits for its continuation

A continuation starts once its source has finished, cleanup included, so a source that waits for its own continuation waits forever:

late final Job<int> parsed;
final loaded = Job<String>((ctx) async {
  final text = await ctx.join(loadText);
  await parsed.value;
  return text;
});
parsed = loaded.then<int>((ctx, text) => int.parse(text));

parsed starts once loaded has finished, and loaded is waiting for parsed. Neither ends, and loaded.cancel() does not return either. parsed.done in place of parsed.value changes nothing, and neither does moving the await into the cleanup of loaded.