Workflow Execution Persistence#

Dify uses a pluggable repository pattern to persist workflow execution state to the database. Two repository protocols — WorkflowExecutionRepository and WorkflowNodeExecutionRepository — are instantiated at runtime by DifyCoreRepositoryFactory, which reads a dotted module path from config and uses import_string to load the implementation class. This lets operators swap backends without code changes.

Pluggable Backends#

Two concrete implementations exist for each protocol.

SQLAlchemy (default) — SQLAlchemyWorkflowExecutionRepository writes synchronously inside a per-call session using session.merge() for upsert 1. An in-memory _execution_cache avoids redundant round-trips for hot records 2.

Celery-based (async) — CeleryWorkflowExecutionRepository serializes the WorkflowExecution domain model via model_dump() and calls save_workflow_execution_task.delay() — fire-and-forget 3. The background task runs on the workflow_storage Celery queue with max_retries=3 and exponential backoff (countdown = 60 * 2**retries) 4.

Config keys (set as dotted module path strings in env):

To switch to the async backend, set CORE_WORKFLOW_EXECUTION_REPOSITORY=core.repositories.celery_workflow_execution_repository.CeleryWorkflowExecutionRepository.

Celery Async Task: save_workflow_execution_task#

save_workflow_execution_task (in api/tasks/workflow_execution_tasks.py) is a @shared_task on the workflow_storage queue. It deserializes the payload, then either creates a new WorkflowRun DB record or updates an existing one — patching status, outputs, error, tokens, steps, and finished_at on updates 7. Failures retry up to 3 times with doubling delay.

HITL Pause State: Synchronous Path#

Pause persistence is intentionally not offloaded to Celery. When a GraphRunPausedEvent fires, PauseStatePersistenceLayer.on_event() runs synchronously in the execution thread to guarantee the pause record exists before the human-input form is surfaced to the user.

What gets packed into WorkflowResumptionContext:

1. serialized_graph_runtime_state — full GraphRuntimeState.dumps() output 8
2. serialized_response_stream_filter_state — ResponseStreamFilter.dumps(); must be the exact same instance WorkflowEntry is using — a different instance silently persists the wrong (empty) state 9
3. generate_entity — type-discriminated union of WorkflowAppGenerateEntity | AdvancedChatAppGenerateEntity 10

The serialized JSON is written to object storage (not a DB column); the WorkflowPause DB record stores only the state_object_key 11. WorkflowRun.status is atomically set to PAUSED in the same transaction 12.

PauseStatePersistenceLayer is only injected when a PauseStateLayerConfig is passed to _generate() — it is absent in debug/single-iteration runs 13.

Resume Flow#

On resume, _resume_app_execution() in workflow_execute_task.py:

1. Fetches WorkflowPause → loads state bytes from object storage via pause_entity.get_state()
2. Deserializes via WorkflowResumptionContext.loads()
3. Restores GraphRuntimeState and ResponseStreamFilter
4. Dispatches to _resume_workflow() or _resume_advanced_chat(); stream=True is forced on the resumed generate entity

repo.resume_workflow_pause() atomically sets WorkflowPause.resumed_at and transitions WorkflowRun.status back to RUNNING 14.

For iteration/loop nodes containing Human Input nodes, CeleryWorkflowNodeExecutionRepository backfills its in-memory cache from the database on first access to a workflow_execution_id, so resumed workflows can see node executions saved in prior iterations 15.

API-Layer Repository (Service Reads)#

A separate repository (API_WORKFLOW_RUN_REPOSITORY, default DifyAPISQLAlchemyWorkflowRunRepository) handles service-layer reads: pagination, statistics, and pause CRUD. This is distinct from the core execution repository used during graph execution 16 17.

Key Source Files#