Owner-bound asyncio workers: cancel siblings and await cleanup

Question

Prerequisites: familiarity with Python asyncio and, for the runnable analogy below, Node.js. Start with the lifetime of the operation: should a failure stop related tasks, or should independent work continue under a different owner? Python TaskGroup provides structured task ownership; ordinary gather aggregates outcomes without automatically cancelling siblings when one raises. Those Python rules are documented by Python, not established by the JavaScript fixture.

The runnable example below is a JavaScript conceptual analogy, executed with Node.js. It uses Promise.all, AbortController and Promise.allSettled to illustrate cooperative cancellation and waiting for cleanup. It is not Python code and its tests do not execute or verify asyncio.gather(), TaskGroup, CancelledError or TimeoutError. The Python recommendations in this article come from the linked Python documentation, not from those JavaScript test results.

The fixture exports runNaive and runSafe. runNaive returns Promise.all without signalling its sibling workers after a failure. runSafe creates an AbortController, signals its cooperative workers, waits with Promise.allSettled, and then reports the original error. This demonstrates an application-owned JavaScript cleanup pattern, not a reimplementation of TaskGroup. The tests cover ordered successful results, continued sibling progress in the naive case, and stopped progress after the safe runner returns.

For a Python service, separately test a successful group, a child exception while another child is waiting, and an enclosing cancellation or timeout with the real I/O library. The JavaScript checks cannot establish Python exception-group behavior, CancelledError handling or TimeoutError conversion. Swallowing cancellation and deliberately shielding work also change the native lifetime; inspect those choices before relying on cleanup.

Working example

export class AbortError extends Error {
  constructor(message = 'Aborted') {
    super(message);
    this.name = 'AbortError';
  }
}

function sleep(ms, signal) {
  return new Promise((resolve, reject) => {
    const timer = setTimeout(done, ms);

    function done() {
      cleanup();
      resolve();
    }

    function onAbort() {
      cleanup();
      reject(new AbortError());
    }

    function cleanup() {
      clearTimeout(timer);
      signal?.removeEventListener('abort', onAbort);
    }

    if (signal?.aborted) {
      cleanup();
      reject(new AbortError());
      return;
    }

    if (signal) signal.addEventListener('abort', onAbort, { once: true });
  });
}

export async function worker(id, {
  steps = 5,
  stepMs = 10,
  signal,
  failAtStep = null,
  progress = {},
} = {}) {
  for (let i = 1; i <= steps; i++) {
    if (signal?.aborted) throw new AbortError();
    await sleep(stepMs, signal);
    progress[id] = (progress[id] ?? 0) + 1;
    if (failAtStep === i) {
      throw new Error(`worker:${id} failed at step ${i}`);
    }
  }
  return `done:${id}`;
}

export async function runNaive(numWorkers, opts = {}) {
  const { failingWorker = null, progress = {}, steps = 5, stepMs = 10 } = opts;
  const tasks = Array.from({ length: numWorkers }, (_, id) => worker(id, {
    steps,
    stepMs,
    progress,
    failAtStep: id === failingWorker ? 2 : null,
  }));
  return Promise.all(tasks);
}

export async function runSafe(numWorkers, opts = {}) {
  const { failingWorker = null, progress = {}, steps = 5, stepMs = 10 } = opts;
  const controller = new AbortController();
  const tasks = Array.from({ length: numWorkers }, (_, id) => worker(id, {
    signal: controller.signal,
    steps,
    stepMs,
    progress,
    failAtStep: id === failingWorker ? 2 : null,
  }));

  try {
    return await Promise.all(tasks);
  } catch (error) {
    if (!(error instanceof AbortError)) controller.abort();
    await Promise.allSettled(tasks);
    throw error;
  }
}

Read the full article: https://chriseugenerodriguez.com/blog/asyncio-failing-task-does-not-leak-workers