Checkpoint и восстановление

Checkpoint сохраняет прогресс workflow после каждого успешного или skipped node. После reload новый WorkflowStore с тем же workflowId продолжит с оставшихся nodes.

StorageWorkflowPersister

import { StorageWorkflowPersister, WorkflowStore } from "@dubium/query-layer/orchestration"

const persister = new StorageWorkflowPersister({
    // sessionStorage ограничивает checkpoint текущей вкладкой.
    storage: window.sessionStorage,

    // Итоговый key: portal-workflow:<encoded workflowId>.
    keyPrefix: "portal-workflow",

    onError: (error) => {
        console.error("Workflow checkpoint error", error)
    },
})

Workflow с checkpoint

const workflow = new WorkflowStore({
    // ID должен быть стабильным для одной бизнес-операции.
    workflowId: `checkout:${checkoutId}`,
    persister,
    nodes: checkoutNodes,
})

await workflow.execute()

Если передан persister, но нет workflowId, constructor бросит ошибку.

Что сохраняется

interface WorkflowSnapshot {
    workflowId: string
    version: 1
    completedNodeIds: readonly string[]
    results: readonly (readonly [string, unknown])[]
    updatedAt: number
}

Не сохраняются:

• функции execute/compensate;
• handlers и MobX stores;
• Axios config;
• AbortSignal;
• текущий running node.

После reload код снова создаёт nodes, stores и API adapters. Snapshot только говорит, какие nodes уже completed и какие serializable results они вернули.

Порядок восстановления

При workflow.execute():

1. persister читает snapshot по workflowId;
2. проверяется schema version 1;
3. проверяется, что все completedNodeIds существуют в текущем graph;
4. results восстанавливаются в Map;
5. completed nodes пропускаются;
6. оставшиеся nodes выполняются по dependencies;
7. после полного success snapshot удаляется.

Если snapshot содержит неизвестный node id, он удаляется и workflow начинается заново.

Когда сохраняется checkpoint

После каждого:

• successful node;
• skipped node.

Writes выстраиваются в последовательную chain, чтобы более старый checkpoint не перезаписал новый.

Что происходит после ошибки

Workflow компенсирует completed nodes и затем удаляет checkpoint. Это важно: после компенсации workflow не должен продолжить с уже отменённого результата.

Ошибки compensation попадают в WorkflowExecutionError.compensationErrors, но checkpoint всё равно удаляется.

Восстановленная compensation

Если snapshot содержит result completed node с compensate, этот node восстанавливается в compensation stack. Если следующий node после reload упадёт, ранее выполненное действие тоже будет компенсировано.

Поэтому сохранённый result должен содержать всё необходимое для rollback:

{
  id: "reserve",
  execute: async () => {
    const reservation = await reserveStore.reserve(input)

    // ID нужен compensation после reload.
    return {
      reservationId: reservation.id,
    }
  },
  compensate: async (result) => {
    const value = result as { reservationId: string }
    await releaseStore.release(value.reservationId)
  },
}

Serializability

StorageWorkflowPersister использует JSON.stringify. Node results должны быть plain JSON:

• null;
• string, number, boolean;
• arrays;
• plain objects.

Не возвращайте в persisted workflow:

• AxiosResponse целиком;
• Error instance;
• Date/Map/Set без явного mapper;
• MobX store;
• File/Blob;
• DOM element;
• function;
• token/credential.

Возвращайте минимальную domain data:

const response = await requestHandler.execute(input)

return {
    orderId: response?.data.id,
}

Options StorageWorkflowPersister

Option	Type	Default	Что делает
storage	Storage	—	Обязательный localStorage/sessionStorage adapter.
keyPrefix	string	QUERY_LAYER_WORKFLOW	Prefix storage key.
onError	(error) => void	—	Получает parse/serialize/storage errors.

Custom persister

Для IndexedDB или backend checkpoint реализуйте interface:

interface IWorkflowPersister {
    save(snapshot: IWorkflowSnapshot): Promise<void>
    restore(workflowId: string): Promise<IWorkflowSnapshot | null>
    remove(workflowId: string): Promise<void>
}

Remote API всё равно не должен отправлять checkpoint на произвольный URL. Backend persister оформляется как typed infrastructure adapter host/application.

Выбор workflowId

Плохо:

workflowId: "checkout"

Два заказа столкнутся.

Лучше:

workflowId: `checkout:${checkoutId}`

Storage key также должен быть разделён по user/tenant на уровне prefix или изолированного storage adapter. Не включайте raw email/token.

Изменение graph между deployments

Snapshot schema version пакета сейчас 1. Но изменение node ids может сделать старый snapshot несовместимым. Включайте version приложения в workflowId или keyPrefix:

keyPrefix: "portal-workflow:v2"

Либо удаляйте старые checkpoints при deployment migration.

Cleanup

StorageWorkflowPersister не имеет отдельного dispose: он не держит listeners. Успешный/скомпенсированный workflow сам вызывает remove(workflowId).

Для отменённого/не завершившегося браузером процесса product policy должна решить, сохранять ли checkpoint для продолжения или удалить его вручную через custom persister boundary.