Codalas

Español

English
Português
中文

Español

English
Português
中文

Tema Oscuro

Flujos asíncronos confiables de varias etapas en TypeScript ​

Introducción ​

Guardar un registro es fácil cuando solo requiere una petición.

Un usuario puede crear un borrador, subir varios archivos, sincronizar sus metadatos y finalmente publicar el registro. Cada paso puede tardar un tiempo distinto y fallar de una forma diferente. La interfaz todavía necesita mostrar progreso útil, la red necesita una estrategia de cancelación y soporte necesita saber hasta dónde llegó la operación.

Colocar todas las promesas dentro de un gran try/catch oculta esos detalles. Un mensaje genérico como "No se pudo guardar" no indica si no existe ningún dato remoto, si quedó un borrador sin archivos o si todos los archivos subieron pero falló la finalización.

En este artículo construiremos un pequeño ejecutor de etapas en TypeScript que le da a un flujo de varios pasos:

  • eventos de progreso con nombre
  • errores de timeout específicos para cada etapa
  • comportamiento abortable y no abortable explícito
  • un registro de las etapas completadas
  • inyección de dependencias para pruebas enfocadas
  • pruebas deterministas de timeout sin esperar al reloj real

El ejemplo usa un flujo genérico de registro y archivos. El mismo patrón sirve para checkout, onboarding, generación de reportes, importaciones, procesamiento de medios y cualquier operación que cruce varios límites asíncronos.

Modela el flujo antes de ejecutarlo ​

Comienza nombrando las etapas. Una unión de strings es simple, fácil de buscar y difícil de escribir mal:

ts
export type SaveStage =
  | 'create_record'
  | 'upload_files'
  | 'finalize_record';

export interface StageCallbacks<
  S extends string,
  P extends object = Record<string, never>,
> {
  onStageStart?: (stage: S) => void;
  onStageComplete?: (stage: S) => void;
  onStageError?: (
    stage: S,
    error: unknown,
    partialOutputs: Readonly<P>,
  ) => void;
}

Estos eventos son más útiles que un porcentaje numérico. La interfaz puede traducir upload_files como "Subiendo archivos adjuntos", analytics puede medir la duración por etapa y los logs pueden identificar el límite que falló.

Construye una primitiva de timeout con semántica honesta ​

Hay dos significados distintos detrás de "esta operación agotó su tiempo":

  1. Dejar de esperar: el código rechaza después de un límite, pero el trabajo subyacente puede continuar.
  2. Solicitar cancelación: el código rechaza y también indica a la operación subyacente que se detenga.

Las promesas de JavaScript no incluyen cancelación. Promise.race() puede hacer que el caller deje de esperar, pero no puede detener la promesa perdedora. AbortController solo ayuda cuando la API subyacente realmente observa su señal.

El siguiente helper admite ambos comportamientos y mantiene la diferencia explícita:

ts
export type ScheduleTimeout = (
  callback: () => void,
  timeoutMs: number,
) => () => void;

export const systemScheduleTimeout: ScheduleTimeout = (callback, timeoutMs) => {
  const timeoutId = setTimeout(callback, timeoutMs);
  return () => clearTimeout(timeoutId);
};

export class StageTimeoutError<S extends string> extends Error {
  constructor(
    public readonly stage: S,
    public readonly timeoutMs: number,
  ) {
    super(`Stage ${stage} timed out after ${timeoutMs}ms`);
    this.name = 'StageTimeoutError';
  }
}

export async function withStageTimeout<T, S extends string>(params: {
  stage: S;
  timeoutMs: number;
  abortable: boolean;
  run: (signal?: AbortSignal) => Promise<T>;
  scheduleTimeout?: ScheduleTimeout;
}): Promise<T> {
  const {
    stage,
    timeoutMs,
    abortable,
    run,
    scheduleTimeout = systemScheduleTimeout,
  } = params;

  const controller = abortable && typeof AbortController === 'function'
    ? new AbortController()
    : null;

  let rejectTimeout: (error: StageTimeoutError<S>) => void = () => {};
  const timeoutPromise = new Promise<never>((_resolve, reject) => {
    rejectTimeout = reject;
  });

  const cancelTimeout = scheduleTimeout(() => {
    // Rechaza primero para que el caller reciba el error de timeout aunque
    // abortar haga que la promesa de la operación rechace de inmediato.
    rejectTimeout(new StageTimeoutError(stage, timeoutMs));
    controller?.abort();
  }, timeoutMs);

  try {
    return await Promise.race([
      run(controller?.signal),
      timeoutPromise,
    ]);
  } finally {
    cancelTimeout();
  }
}

Para una etapa abortable, pasa la señal hasta una API como fetch:

ts
run: (signal) => fetch('/records', {
  method: 'POST',
  body: JSON.stringify(input),
  signal,
})

No marques una etapa como abortable solo porque el ejecutor puede crear una señal. Si la operación ignora esa señal, el timeout únicamente deja de esperar. El trabajo puede terminar y producir un efecto secundario más tarde.

Un timeout no es un rollback

Un timeout indica que el caller dejó de esperar. No demuestra que una escritura remota falló. Usa claves de idempotencia o una lectura de reconciliación antes de repetir una etapa que pudo haber llegado al servidor.

Envuelve cada etapa y conserva la finalización parcial ​

El helper de timeout controla una promesa. El ejecutor de etapas agrega eventos de ciclo de vida y contexto de fallo:

ts
export class StagedWorkflowError<
  S extends string,
  P extends object,
> extends Error {
  constructor(
    public readonly failedStage: S,
    public readonly completedStages: readonly S[],
    public readonly partialOutputs: Readonly<P>,
    public readonly cause: unknown,
  ) {
    super(`Workflow failed during ${failedStage}`);
    this.name = 'StagedWorkflowError';
  }
}

export async function runStage<
  T,
  S extends string,
  P extends object,
>(params: {
  stage: S;
  completedStages: S[];
  partialOutputs: Readonly<P>;
  timeoutMs: number;
  abortable?: boolean;
  run: (signal?: AbortSignal) => Promise<T>;
  callbacks?: StageCallbacks<S, P>;
  scheduleTimeout?: ScheduleTimeout;
}): Promise<T> {
  const {
    stage,
    completedStages,
    partialOutputs,
    timeoutMs,
    abortable = false,
    run,
    callbacks = {},
    scheduleTimeout,
  } = params;

  callbacks.onStageStart?.(stage);

  try {
    const result = await withStageTimeout({
      stage,
      timeoutMs,
      abortable,
      run,
      scheduleTimeout,
    });

    completedStages.push(stage);
    callbacks.onStageComplete?.(stage);
    return result;
  } catch (error) {
    const outputSnapshot = Object.freeze({ ...partialOutputs });
    callbacks.onStageError?.(stage, error, outputSnapshot);
    throw new StagedWorkflowError(
      stage,
      [...completedStages],
      outputSnapshot,
      error,
    );
  }
}

Una etapa se registra solo después de resolverse. Si upload_files falla, el error todavía puede mostrar que create_record terminó y exponer el borrador creado en esa etapa. Copiar los arreglos y el objeto de resultados parciales evita que mutaciones posteriores de nivel superior cambien el estado histórico del fallo.

Los callbacks del ciclo de vida deben ser livianos. Si un callback de progreso puede lanzar una excepción, aísla ese comportamiento dentro del callback para que un problema de renderizado o analytics no convierta una operación remota exitosa en un guardado fallido.

Compón el flujo real de guardado ​

La inyección de dependencias mantiene el orquestador independiente de un cliente HTTP, una base de datos o un SDK de almacenamiento específico:

ts
interface UploadedFile {
  key: string;
  url: string;
}

interface FileToUpload {
  name: string;
  bytes: Uint8Array;
}

export interface SavePartialOutputs {
  readonly record?: { readonly id: string };
  readonly uploadedFiles?: readonly UploadedFile[];
}

interface SaveDependencies {
  createDraft: (
    input: { title: string },
    signal?: AbortSignal,
  ) => Promise<{ id: string }>;
  uploadFiles: (
    recordId: string,
    files: FileToUpload[],
  ) => Promise<UploadedFile[]>;
  finalizeRecord: (
    recordId: string,
    files: UploadedFile[],
    signal?: AbortSignal,
  ) => Promise<void>;
}

export async function saveRecordWithFiles(params: {
  title: string;
  files: FileToUpload[];
  dependencies: SaveDependencies;
  callbacks?: StageCallbacks<SaveStage, SavePartialOutputs>;
  timeoutMs?: number;
}) {
  const {
    title,
    files,
    dependencies,
    callbacks,
    timeoutMs = 20_000,
  } = params;

  const completedStages: SaveStage[] = [];
  const initialOutputs: SavePartialOutputs = {};

  const record = await runStage({
    stage: 'create_record',
    completedStages,
    partialOutputs: initialOutputs,
    timeoutMs,
    abortable: true,
    callbacks,
    run: (signal) => dependencies.createDraft({ title }, signal),
  });

  const createdOutputs: SavePartialOutputs = { record };
  const uploadedFiles = files.length > 0
    ? await runStage({
        stage: 'upload_files',
        completedStages,
        partialOutputs: createdOutputs,
        timeoutMs,
        callbacks,
        run: () => dependencies.uploadFiles(record.id, files),
      })
    : [];

  const uploadedOutputs: SavePartialOutputs = {
    record,
    uploadedFiles: [...uploadedFiles],
  };

  await runStage({
    stage: 'finalize_record',
    completedStages,
    partialOutputs: uploadedOutputs,
    timeoutMs,
    abortable: true,
    callbacks,
    run: (signal) => dependencies.finalizeRecord(
      record.id,
      uploadedFiles,
      signal,
    ),
  });

  return { record, uploadedFiles, completedStages };
}

Aquí las etapas de red son abortables porque sus dependencias aceptan una señal. El adaptador de carga está marcado intencionalmente como no abortable. Su timeout termina la espera de la interfaz, pero el adaptador puede continuar. Si tu SDK de almacenamiento admite cancelación, cambia el contrato de la dependencia para aceptar una señal y habilítala.

Este flujo es secuencial porque las etapas posteriores necesitan resultados anteriores. Las cargas independientes todavía pueden ejecutarse en paralelo dentro de uploadFiles, donde puedes controlar límites de concurrencia y errores por archivo sin dificultar la lectura del flujo principal.

Diseña el estado de fallo, no solo el de éxito ​

La lista de etapas completadas es estado operativo, no un mecanismo de rollback.

Supón que la finalización falla después de crear el borrador y subir los archivos. El error y el callback reciben un snapshot tipado de partialOutputs, así que la aplicación puede conservar el ID del registro y el contexto de los archivos subidos para un reintento específico.

Los resultados parciales pueden contener URLs firmadas u otros datos sensibles. No serialices el error completo directamente. Un log de soporte sanitizado puede contener:

json
{
  "failedStage": "finalize_record",
  "completedStages": ["create_record", "upload_files"],
  "recordId": "record-1",
  "uploadedFileKeys": ["report.txt"]
}

Eso permite actuar mucho mejor que un stack trace por sí solo.

Para flujos de producción, combina las etapas con protecciones del servidor:

  • usa una clave de idempotencia al crear un registro
  • crea registros con estado draft o processing
  • haz que la finalización se pueda repetir de forma segura
  • conserva claves estables para los archivos en lugar de generar nuevas en cada reintento
  • agrega limpieza o reconciliación para borradores abandonados y cargas huérfanas
  • reintenta solo etapas cuyos efectos secundarios entiendas

Los reintentos automáticos son peligrosos para operaciones no idempotentes. Si ocurre un timeout después de que el servidor confirmó una escritura, repetir la petición a ciegas puede crear un duplicado. Primero consulta por la clave de idempotencia o por el ID estable del registro.

Convierte los eventos de etapa en estado útil para la interfaz ​

Los callbacks crean un límite claro entre orquestación y presentación. Una pantalla no necesita comprender las promesas del flujo; solo necesita la etapa activa y el resultado final:

ts
let activeStage: SaveStage | null = null;
let lastPartialOutputs: Readonly<SavePartialOutputs> = {};

const callbacks: StageCallbacks<SaveStage, SavePartialOutputs> = {
  onStageStart: (stage) => {
    activeStage = stage;
    renderProgress(stage);
  },
  onStageComplete: () => {
    activeStage = null;
  },
  onStageError: (stage, error, partialOutputs) => {
    activeStage = null;
    lastPartialOutputs = partialOutputs;
    reportSaveFailure({
      stage,
      error,
      recordId: partialOutputs.record?.id,
      uploadedFileKeys: partialOutputs.uploadedFiles?.map(
        (file) => file.key,
      ),
    });
  },
};

Si la finalización falla, lastPartialOutputs.record?.id y lastPartialOutputs.uploadedFiles quedan disponibles para la interfaz de reintento. En un timeout de carga no abortable, el snapshot contiene intencionalmente solo el registro: la carga que sigue ejecutándose todavía no produjo un resultado confirmado.

Usa los nombres de las etapas como estado, no como texto visible directamente. Una capa de presentación puede mapear create_record a "Preparando tu registro" y upload_files a "Subiendo archivos adjuntos", y luego localizar esas etiquetas normalmente.

Evita prometer un porcentaje exacto si no puedes medir el trabajo de cada etapa. Un spinner con una etiqueta honesta suele ser mejor que un progreso que salta de 10 % a 90 % o queda detenido cerca del final. También bloquea envíos duplicados mientras el flujo está activo, pero muestra un botón para cancelar solo cuando cancelar tenga una semántica definida. Cerrar un modal no equivale a detener el trabajo remoto.

Prueba el orden y la finalización parcial ​

Las dependencias inyectadas convierten el camino exitoso en una prueba pura de orquestación:

ts
import assert from 'node:assert/strict';
import test from 'node:test';

test('runs stages in order and reports completed work', async () => {
  const calls: string[] = [];

  const result = await saveRecordWithFiles({
    title: 'Quarterly report',
    files: [{
      name: 'report.txt',
      bytes: new TextEncoder().encode('data'),
    }],
    dependencies: {
      createDraft: async () => {
        calls.push('create');
        return { id: 'record-1' };
      },
      uploadFiles: async () => {
        calls.push('upload');
        return [{ key: 'report.txt', url: '/files/report.txt' }];
      },
      finalizeRecord: async () => {
        calls.push('finalize');
      },
    },
  });

  assert.deepEqual(calls, ['create', 'upload', 'finalize']);
  assert.deepEqual(result.completedStages, [
    'create_record',
    'upload_files',
    'finalize_record',
  ]);
});

Prueba el contrato de recuperación directamente haciendo fallar la finalización después de que las etapas anteriores terminen:

ts
test('exposes partial outputs when finalization fails', async () => {
  const expectedOutputs: SavePartialOutputs = {
    record: { id: 'record-1' },
    uploadedFiles: [{
      key: 'report.txt',
      url: '/files/report.txt',
    }],
  };
  let callbackOutputs: Readonly<SavePartialOutputs> | undefined;

  const promise = saveRecordWithFiles({
    title: 'Quarterly report',
    files: [{
      name: 'report.txt',
      bytes: new TextEncoder().encode('data'),
    }],
    callbacks: {
      onStageError: (_stage, _error, partialOutputs) => {
        callbackOutputs = partialOutputs;
      },
    },
    dependencies: {
      createDraft: async () => ({ id: 'record-1' }),
      uploadFiles: async () => [{
        key: 'report.txt',
        url: '/files/report.txt',
      }],
      finalizeRecord: async () => {
        throw new Error('finalization failed');
      },
    },
  });

  await assert.rejects(promise, (error: unknown) => {
    assert.ok(error instanceof StagedWorkflowError);
    assert.equal(error.failedStage, 'finalize_record');
    assert.deepEqual(error.completedStages, [
      'create_record',
      'upload_files',
    ]);
    assert.deepEqual(error.partialOutputs, expectedOutputs);
    return true;
  });

  assert.deepEqual(callbackOutputs, expectedOutputs);
});

Agrega una prueba complementaria donde uploadFiles rechace. Verifica que failedStage sea upload_files, que completedStages contenga solo create_record, que partialOutputs incluya el registro pero no archivos subidos y que finalizeRecord nunca se ejecute.

Prueba los timeouts sin hacer sleep ​

Las pruebas que esperan 10 o 50 milisegundos pueden ser inestables en un runner de CI ocupado. Como el scheduler del timeout es inyectable, la prueba puede dispararlo manualmente:

ts
test('aborts an abortable stage and keeps timeout context', async () => {
  let fireTimeout: () => void = () => {};
  let wasAborted = false;

  const scheduleTimeout: ScheduleTimeout = (callback) => {
    fireTimeout = callback;
    return () => {};
  };

  const promise = runStage({
    stage: 'create_record' as SaveStage,
    completedStages: [],
    partialOutputs: {},
    timeoutMs: 20_000,
    abortable: true,
    scheduleTimeout,
    run: (signal) => new Promise((_resolve, reject) => {
      signal?.addEventListener('abort', () => {
        wasAborted = true;
        reject(new Error('request aborted'));
      });
    }),
  });

  fireTimeout();

  await assert.rejects(promise, (error: unknown) => {
    assert.ok(error instanceof StagedWorkflowError);
    assert.equal(error.failedStage, 'create_record');
    assert.deepEqual(error.completedStages, []);
    assert.ok(error.cause instanceof StageTimeoutError);
    return true;
  });

  assert.equal(wasAborted, true);
});

El reloj real no avanza, así que la prueba es rápida y determinista. Una prueba complementaria para una etapa no abortable debe confirmar que el caller recibe un timeout y documentar que el trabajo pendiente no se cancela.

Conclusión ​

Un flujo confiable de varias etapas hace visibles sus límites. Las etapas con nombre explican el progreso, los errores de timeout identifican la operación detenida, AbortSignal expresa cancelación cuando la dependencia la admite y el historial de etapas completadas junto con resultados parciales tipados permite actuar ante un fallo parcial.

El ejecutor es pequeño a propósito. El trabajo de diseño importante está alrededor: semántica honesta de cancelación, operaciones idempotentes en el servidor, reglas deliberadas de reintento y pruebas deterministas. Con esas piezas, un guardado complicado se convierte en una secuencia que puedes observar, explicar y recuperar.

Todos los derechos reservados © 2018-2026 Esteban Quezada