Pool de concorrência com Async Iterators e AbortSignal

Motivacao e trade-offs

Trabalhar com operações assíncronas em paralelo é rotina: chamadas HTTP, leitura de arquivos, processamento CPU-bound delegando para workers. Um pool de concorrência controla quantas tarefas rodam simultaneamente, evitando saturar recursos e melhorando previsibilidade. A implementação com Async Iterators e AbortSignal traz duas vantagens práticas: consumo incremental de tarefas e cancelamento cooperativo. Mas existem trade-offs que devem ser entendidos.

Por que usar um pool em vez de Promise.all

Promise.all é simples, mas dispara todas as promessas imediatamente. Quando o número de itens é grande ou as tarefas consomem recursos externos (DB, API, CPU), isso pode causar timeouts, quedas de conexão ou uso excessivo de memória. Um pool limita a simultaneidade e permite backpressure.

Compromissos: latência vs throughput

Limitar concorrência reduz pressão, mas pode aumentar latência média por item. A escolha do limite é um trade-off entre throughput (quantos completam por segundo) e latência por item. Também há complexidade adicional: propagação de erro, cancelamento e ordenação parcial. Use métricas e load testing para ajustar o tamanho do pool.

Lista rápida de considerações:

• Escolher limite baseado em recursos (CPU, conexões DB, quotas API).
• Propagar cancelamento quando relevante para liberar recursos.
• Manter semântica de erro clara: falha de uma tarefa não precisa abortar todas, a menos que seja desejado.

Fundamentos técnicos

Antes de implementar, revisite os fundamentos que usamos aqui: Async Iterators permitem produzir um fluxo assíncrono consumido com for await...of. AbortController/AbortSignal oferecem cancelamento cooperativo; APIs modernas (fetch, streams) aceitam sinal para interromper.

Async Iterators como fonte de tarefas

Async generators são úteis quando a lista de tarefas é gerada dinamicamente, p.ex.: leitura paginada de DB, eventos de fila ou produção por I/O assíncrono. Em vez de armazenar tudo em memória, você processa item a item.

AbortSignal para cancelamento cooperativo

AbortSignal não força interrupção; ele notifica. A tarefa deve observar o sinal e abortar suas operações (p.ex. fetch com signal, limpar timers, encerrar conexões). Isso evita leaks e possibilita cancelamento ordenado.

Arquitetura mínima que vamos usar:

1. Um produtor: Async Iterator que emite tarefas (valores ou funções retornando promessas).
2. Um pool: consome o iterator, mantém N workers rodando.
3. AbortController externo: permite cancelar todo o processamento.
4. Tratamento de erros: configura se um erro cancela todo o pool ou apenas a tarefa.

Tip: prefira propagar AbortSignal para as chamadas internas (fetch, db clients). Evite matar timers com hacks; trate o sinal e limpe explicitamente.

Exemplo 1: caso simples

Visão geral

Implementação compacta: um pool que consome um Async Iterator de funções assíncronas e respeita um limite de concorrência. O pool devolve uma promessa que resolve quando todas as tarefas forem processadas ou rejeita se o AbortSignal for acionado.

Código: pool básico com async iterator

A seguir, um exemplo funcional. Contexto: temos um async generator que produz funções que retornam promessas. O pool dispara até concurrency tarefas e espera que uma complete antes de puxar a próxima.

async function* taskProducer(items) {
  for (const item of items) {
    // cada yield é uma função assíncrona que aceita um AbortSignal
    yield async (signal) => {
      // simula trabalho assíncrono dependente de signal
      if (signal.aborted) throw new Error('cancelled');
      await new Promise((res, rej) => {
        const t = setTimeout(res, Math.random() * 200 + 50);
        signal.addEventListener('abort', () => {
          clearTimeout(t);
          rej(new Error('aborted'));
        });
      });
      return item * 2; // resultado
    };
  }
}

async function runPool(taskIterator, concurrency = 4, controller = new AbortController()) {
  const { signal } = controller;
  const running = new Set();

  for await (const task of taskIterator) {
    // espera até ter recurso para iniciar nova tarefa
    while (running.size >= concurrency) {
      await Promise.race(running);
    }

    const p = (async () => {
      try {
        return await task(signal);
      } finally {
        // removemos a promessa do conjunto quando concluída
        running.delete(p);
      }
    })();

    running.add(p);
  }

  // aguarda todos terminarem
  return Promise.all(running);
}

O que acontece: o produtor gera funções. O pool mantém até concurrency promessas em execução. Quando o sinal é abortado por fora, tarefas que observarem o signal devem rejeitar.

Exemplo de uso:

(async () => {
  const items = Array.from({ length: 10 }, (_, i) => i + 1);
  const producer = taskProducer(items);
  const controller = new AbortController();

  // cancela após 1 segundo (apenas demonstrativo)
  setTimeout(() => controller.abort(), 1000);

  try {
    const results = await runPool(producer, 3, controller);
    console.log('results', results);
  } catch (err) {
    console.error('pool failed', err.message);
  }
})();

Resultado prático: tarefas iniciadas até o limite concorrem; se o controller abortar, as tasks que observarem o signal vão rejeitar e o pool pode propagar erro.

Exemplo 2: caso avançado do mundo real

Contexto real: chamadas HTTP com timeouts e reuso de conexões

Imagine fetch de vários endpoints (ou downloads) em que é preciso limitar concorrência, aplicar timeout por requisição, e parar tudo se uma condição crítica ocorrer (p.ex. erro de autenticação). Usaremos AbortSignal para timeout e cancelamento global, e um padrão para re-tentativa limitada.

Código: pool com timeout, retries e propagação de abort

Descrição: o código abaixo consome um async iterator de URLs. Cada worker faz fetch com timeout e retries. AbortController global cancela todos. O retry respeita o mesmo AbortSignal.

async function* urlProducer(urls) {
  for (const url of urls) yield url;
}

function fetchWithTimeout(url, { signal, timeoutMs }) {
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), timeoutMs);

  // Encadeia sinais: se o outer signal abortar, propaga
  const chained = new AbortController();
  const onOuter = () => chained.abort();
  signal?.addEventListener('abort', onOuter);
  signal = chained.signal;

  return fetch(url, { signal: controller.signal })
    .finally(() => {
      clearTimeout(timeout);
      signal?.removeEventListener?.('abort', onOuter);
    });
}

async function runPoolUrls(urlIterator, concurrency = 5, controller = new AbortController()) {
  const { signal } = controller;
  const running = new Set();
  const results = [];

  for await (const url of urlIterator) {
    while (running.size >= concurrency) await Promise.race(running);

    const p = (async () => {
      const maxRetries = 2;
      for (let attempt = 0; attempt <= maxRetries; attempt++) {
        if (signal.aborted) throw new Error('global abort');
        try {
          const res = await fetchWithTimeout(url, { signal, timeoutMs: 5000 });
          if (!res.ok) throw new Error(`status ${res.status}`);
          const text = await res.text();
          results.push({ url, ok: true, length: text.length });
          return;
        } catch (err) {
          if (attempt === maxRetries) throw err;
          // backoff simples
          await new Promise(r => setTimeout(r, 100 * (attempt + 1)));
        }
      }
    })();

    running.add(p.finally(() => running.delete(p)));
  }

  await Promise.all(running);
  return results;
}

// Uso
(async () => {
  const urls = ['https://example.com', 'https://example.org', 'https://httpstat.us/500'];
  const producer = urlProducer(urls);
  const controller = new AbortController();

  // Exemplo: aborta em 8s
  setTimeout(() => controller.abort(), 8000);

  try {
    const r = await runPoolUrls(producer, 3, controller);
    console.log('fetched', r);
  } catch (err) {
    console.error('error', err.message);
  }
})();

Explicação: cada fetch tem timeout independente e re-tenta. O controlador global aborta a cadeia. Atenção: a implementação do fetchWithTimeout faz encadeamento simples de sinais — em código de produção prefira utilitários testados para encadear AbortSignals corretamente.

Debugging e troubleshooting

Quando algo falha em pools assíncronos, as causas mais comuns são: tarefas que não respeitam AbortSignal, promessas não resolvidas (leaks), e concorrência incorretamente controlada. Vamos listar erros comuns e como detectá-los.

Erros comuns e armadilhas

1) Promessas que nunca terminam: timers ou sockets não limpos. Ferramenta: inspecione heap/handles ativos (node --trace-warnings, ferramentas do runtime). 2) Ignorar AbortSignal: se uma tarefa não observa o sinal, o pool pode parecer "travado" após um abort. 3) Race conditions: remover promessas do conjunto de running no lugar errado pode causar Promise.race que nunca resolve.

Checklist de investigação:

• Verifique se every promise has a finally that limpa recursos.
• Logue quando cada task começa e termina com timestamps.
• Use timeouts de watchdog para detectar tasks presas.

Ferramentas práticas

Use logging estruturado (request id) e métricas: histrograma de latências, contadores de tarefas activas, e total de erros. No Node, o inspector e tracer ajudam a identificar handles pendentes. No browser, DevTools -> Performance e Network mostram fetch pendentes.

// Exemplo de wrapper para instrumentacao simples
function instrumentTask(fn, name) {
  return async (signal) => {
    const start = Date.now();
    try {
      console.log(`${name} start`);
      const r = await fn(signal);
      console.log(`${name} done in ${Date.now() - start}ms`);
      return r;
    } catch (e) {
      console.log(`${name} error after ${Date.now() - start}ms: ${e.message}`);
      throw e;
    }
  };
}

Contexto: esse wrapper ajuda a identificar tarefas problemáticas em um pool grande. Enrole antes de enviar ao pool e filtre logs por request id.

Conclusao e guia prático

Resumo prático: use Async Iterators para fontes dinâmicas de tarefas, implemente um pool simples controlando um conjunto de promessas e use AbortSignal para cancelamento cooperativo. Priorize limpeza (finally), instrumente latências e trate a política de erros (fail-fast vs best-effort).

Recomendações rápidas:

• Escolha concurrency baseado em testes reais, não chute.
• Propague AbortSignal para todo I/O: fetch, streams, clients.
• Implemente timeouts e retries com backoff onde faz sentido.
• Instrumente e monitore: sem métricas, ajustar pool é adivinhação.

Por fim, implemente um contrato claro: o que acontece em caso de erro ou cancelamento? Documente e escreva testes que simulem abortos e tarefas presas. Um pool bem projetado reduz bugs de produção e torna o sistema previsível sob carga.

Implementações mais avançadas podem adicionar prioridades, fairness, filas persistentes e métricas por tarefa. Mas os padrões mostrados aqui já resolvem a maior parte dos casos pragmáticos com clareza e desempenho.