Persistence extension for WorkflowForge enabling resumable workflows via a pluggable provider (bring your own storage). Zero-dependency core integration.
Get Started
$ dotnet add package WorkflowForge.Extensions.PersistenceReadme
WorkflowForge.Extensions.Persistence
Workflow state persistence extension for WorkflowForge with in-memory and SQLite providers for checkpointing and recovery.
Dependencies
InMemory Provider: ZERO external dependencies - Pure WorkflowForge
SQLite Provider: Uses Microsoft.Data.Sqlite as an external dependency (runtime unification)
Installation
dotnet add package WorkflowForge.Extensions.Persistence
Requires: .NET Standard 2.0 or later
Quick Start
In-Memory (Development/Testing)
using WorkflowForge.Extensions.Persistence.InMemory;
var provider = new InMemoryPersistenceProvider();
// Save workflow state
var state = new WorkflowState
{
ExecutionId = foundry.ExecutionId,
WorkflowName = workflow.Name,
Properties = foundry.Properties.ToDictionary(x => x.Key, x => x.Value),
CompletedOperations = new List<string> { "Op1", "Op2" }
};
await provider.SaveWorkflowStateAsync(foundry.ExecutionId, state);
// Load workflow state
var loadedState = await provider.LoadWorkflowStateAsync(foundry.ExecutionId);
SQLite (Production)
using WorkflowForge.Extensions.Persistence.SQLite;
var provider = new SQLitePersistenceProvider("workflows.db");
// Same API as InMemory
await provider.SaveWorkflowStateAsync(foundry.ExecutionId, state);
var loadedState = await provider.LoadWorkflowStateAsync(foundry.ExecutionId);
Key Features
- Two Providers: InMemory (testing) and SQLite (production)
- Checkpoint Support: Save workflow state at any point
- Resume Workflows: Continue from last checkpoint
- Pluggable Architecture:
IWorkflowPersistenceProviderinterface - Thread-Safe: Concurrent workflow support
- Transactional: SQLite provider uses transactions
Configuration
Via appsettings.json
{
"WorkflowForge": {
"Extensions": {
"Persistence": {
"Enabled": true,
"PersistOnOperationComplete": true,
"PersistOnWorkflowComplete": true,
"PersistOnFailure": true,
"MaxVersions": 10,
"InstanceId": "my-instance-id",
"WorkflowKey": "my-workflow-key"
}
}
}
}
Via Code
using WorkflowForge.Extensions.Persistence;
var options = new PersistenceOptions
{
Enabled = true,
PersistOnOperationComplete = true,
PersistOnWorkflowComplete = true,
PersistOnFailure = true,
MaxVersions = 10,
InstanceId = "my-instance-id",
WorkflowKey = "my-workflow-key"
};
foundry.UsePersistence(provider, options);
Via Dependency Injection
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using WorkflowForge.Extensions.Persistence;
services.AddPersistenceConfiguration(configuration);
var options = serviceProvider.GetRequiredService<IOptions<PersistenceOptions>>().Value;
See Configuration Guide for complete options.
Workflow State Structure
public class WorkflowState
{
public Guid ExecutionId { get; set; }
public string WorkflowName { get; set; }
public Dictionary<string, object> Properties { get; set; }
public List<string> CompletedOperations { get; set; }
public int NextOperationIndex { get; set; }
public DateTimeOffset CreatedAt { get; set; }
public DateTimeOffset? UpdatedAt { get; set; }
}
Custom Persistence Provider
public class CosmosDbPersistenceProvider : IWorkflowPersistenceProvider
{
private readonly CosmosClient _client;
private readonly Container _container;
public async Task SaveWorkflowStateAsync(
Guid executionId,
WorkflowState state,
CancellationToken cancellationToken = default)
{
await _container.UpsertItemAsync(state, new PartitionKey(executionId.ToString()), cancellationToken: cancellationToken);
}
public async Task<WorkflowState?> LoadWorkflowStateAsync(
Guid executionId,
CancellationToken cancellationToken = default)
{
try
{
var response = await _container.ReadItemAsync<WorkflowState>(
executionId.ToString(),
new PartitionKey(executionId.ToString()),
cancellationToken: cancellationToken);
return response.Resource;
}
catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
{
return null;
}
}
}
Usage Patterns
Save Checkpoint at Each Operation
foundry.OperationCompleted += async (s, e) =>
{
var state = CreateStateFromFoundry(foundry);
await provider.SaveWorkflowStateAsync(foundry.ExecutionId, state);
};
Resume from Checkpoint
var state = await provider.LoadWorkflowStateAsync(executionId);
if (state != null)
{
// Restore foundry state
foreach (var prop in state.Properties)
{
foundry.SetProperty(prop.Key, prop.Value);
}
// Skip completed operations
foundry.SetProperty("NextOperationIndex", state.NextOperationIndex);
}
Documentation
- Getting Started
- Configuration Guide
- Extensions Overview
- Sample 18: Persistence
- Recovery Extension - Recovery orchestration on top of persistence