Узлы и зависимости

Workflow nodes образуют DAG — направленный граф без циклов.

Поля node

interface IWorkflowNode<TValue = unknown> {
    id: string
    dependencies?: readonly string[]
    condition?: (context: IWorkflowNodeContext) => boolean | Promise<boolean>
    execute(context: IWorkflowNodeContext): Promise<TValue>
    compensate?(value: TValue, context: IWorkflowNodeContext): Promise<void>
}

Поле	Обязательное	Что означает
id	да	Уникальное имя node и key результата.
dependencies	нет	Nodes, которые должны завершиться или быть skipped раньше.
condition	нет	false пропускает node.
execute	да	Выполняет шаг и возвращает result.
compensate	нет	Отменяет эффект успешного node после последующей ошибки.

Context node

interface IWorkflowNodeContext {
    results: ReadonlyMap<string, unknown>
    signal: AbortSignal
}

• results содержит значения уже завершённых nodes;
• signal сообщает, что весь workflow отменён.

Execution levels

const nodes = [
    { id: "profile", execute: loadProfile },
    { id: "permissions", execute: loadPermissions },
    {
        id: "settings",
        dependencies: ["permissions"],
        execute: loadSettings,
    },
    {
        id: "dashboard",
        dependencies: ["profile", "settings"],
        execute: buildDashboard,
    },
]

Граф создаёт уровни:

Level	Nodes	Почему
1	profile, permissions	Нет dependencies.
2	settings	Ждёт permissions.
3	dashboard	Ждёт profile и settings.

concurrency ограничивает параллелизм внутри одного level. Он не запускает node до dependencies.

Использование результатов

{
  id: "dashboard",
  dependencies: ["profile", "settings"],
  execute: async ({ results }) => {
    // Map имеет unknown values, поэтому domain code проверяет или уточняет type.
    const profile = results.get("profile") as Profile
    const settings = results.get("settings") as DashboardSettings

    return {
      title: `Кабинет ${profile.name}`,
      widgets: settings.widgets,
    }
  },
}

results — Map, а не объект. Node id может содержать точки/дефисы и не становится property name.

Для строгой type safety создайте project helpers/guards:

const requireResult = <T>(results: ReadonlyMap<string, unknown>, nodeId: string): T => {
    if (!results.has(nodeId)) {
        throw new Error(`Missing workflow result: ${nodeId}`)
    }

    return results.get(nodeId) as T
}

Этот helper не входит в npm-пакет.

Condition

{
  id: "admin-settings",
  dependencies: ["permissions"],
  condition: ({ results }) => {
    const permissions = requireResult<string[]>(results, "permissions")
    return permissions.includes("admin")
  },
  execute: async () => adminSettingsStore.fetch(),
}

Если condition вернула false:

• node получает progress status skipped;
• считается завершённой для dependencies;
• result в Map не добавляется;
• checkpoint запоминает node как completed.

Зависимый node должен учитывать отсутствие result skipped dependency.

Ошибка condition

Если condition бросила ошибку, это ошибка node. Workflow остановит выдачу новых nodes и начнёт compensation уже завершённых шагов.

Cancellation signal

workflow.cancel() вызывает AbortController workflow. Он прекращает запуск новых nodes, но уже начавшийся domain request должен сам отреагировать на signal.

Надёжный adapter для store с методом cancel():

const executeCancelableStore = async <T>(
    signal: AbortSignal,
    execute: () => Promise<T>,
    cancel: () => void,
): Promise<T> => {
    // Если workflow уже отменён, не запускаем store.
    signal.throwIfAborted()

    // При abort вызываем cancellation конкретного query-layer store.
    const onAbort = (): void => cancel()
    signal.addEventListener("abort", onAbort, { once: true })

    try {
        return await execute()
    } finally {
        signal.removeEventListener("abort", onAbort)
    }
}

Helper проектный и не входит в пакет.

{
  id: "profile",
  execute: ({ signal }) => {
    return executeCancelableStore(
      signal,
      () => profileStore.fetch(),
      () => profileStore.cancel(),
    )
  },
}

Не передавайте signal в Axios из workflow node напрямую, обходя store.

Валидация графа

При workflow.execute() проверяются:

• duplicate id;
• неизвестная dependency;
• dependency на самого себя;
• cycle A → B → A.

Пример цикла:

;[
    { id: "a", dependencies: ["b"], execute: async () => null },
    { id: "b", dependencies: ["a"], execute: async () => null },
]

Workflow завершится ошибкой до выполнения HTTP nodes.

Ограничение concurrency

const workflow = new WorkflowStore({
    concurrency: 3,
    nodes,
})

Значение округляется вниз и не может быть меньше 1. Без option параллелизм уровня не ограничен. Для backend rate limits задавайте явное значение.

Следующая страница: ошибки и компенсация.