Orchestration Persistence

tramai-orchestration keeps workflow persistence storage-agnostic. The workflow runtime owns step ordering and checkpoint timing — your application chooses how state is serialized and where checkpoints live.

What This Covers

WorkflowStateCodec — encode/decode typed workflow state
WorkflowCheckpointStore — store revisioned checkpoints
WorkflowPersistence — wire codec and store together
Reference stores: InMemory, File, Markdown, JDBC
Multi-node ownership with WorkflowLeaseStore
File-backed and JDBC lease guidance

When to Use Persistence

Add persistence when:

workflows need to survive process restarts
workflows run longer than the JVM's expected lifetime
you need checkpoint-based resume capability
multiple workers share the same workflow queue

Core Pieces

WorkflowStateCodec

The codec is the right place to choose JSON, YAML, protobuf-over-base64, or a domain-specific format.

data class ReviewState(
    val requestId: String,
    val draft: String? = null,
    val approved: Boolean = false,
)

object ReviewStateCodec : WorkflowStateCodec<ReviewState> {
    override fun encode(state: ReviewState): String = listOf(
        state.requestId, state.draft.orEmpty(), state.approved.toString(),
    ).joinToString("|")

    override fun decode(payload: String): ReviewState {
        val parts = payload.split("|", limit = 3)
        return ReviewState(
            requestId = parts[0],
            draft = parts.getOrNull(1).orEmpty().ifBlank { null },
            approved = parts.getOrNull(2)?.toBooleanStrictOrNull() ?: false,
        )
    }
}

For richer state, JSON is usually the most practical choice.

WorkflowCheckpointStore

Implement this SPI when you need a custom storage backend:

class MyCheckpointStore(
    private val client: CheckpointClient,
) : WorkflowCheckpointStore {
    override suspend fun load(
        workflowName: String,
        workflowId: String,
    ): WorkflowCheckpoint? = client.read(workflowName, workflowId)

    override suspend fun save(
        checkpoint: WorkflowCheckpoint,
        expectedRevision: Long?,
    ): WorkflowCheckpoint {
        return client.write(checkpoint, expectedRevision)
    }

    override suspend fun delete(
        workflowName: String,
        workflowId: String,
        expectedRevision: Long?,
    ) { client.delete(workflowName, workflowId, expectedRevision) }
}

Revision contract:

first save: expectedRevision = null
later saves: use the current revision
stale writers: fail with WorkflowCheckpointConflictException

This gives database, object-store, and filesystem implementations the same optimistic-concurrency model.

Wiring a Workflow

val persistence = WorkflowPersistence(
    checkpointStore = MyCheckpointStore(client),
    stateCodec = ReviewStateCodec,
)

// Start
val result = workflow.run(
    initialState = ReviewState(requestId = "invoice-123", approved = true),
    persistence = persistence,
)

// Resume
val result = workflow.resume(
    context = WorkflowContext(workflowId = "invoice-123"),
    persistence = persistence,
)

Reference Stores

The repository includes these reference implementations:

Store	File-based	JDBC	In-memory
Checkpoint	`FileWorkflowCheckpointStore`	`JdbcWorkflowCheckpointStore`	`InMemoryWorkflowCheckpointStore`
Alternative	`MarkdownWorkflowCheckpointStore`	—	—
Lease	`FileWorkflowLeaseStore`	`JdbcWorkflowLeaseStore`	`InMemoryWorkflowLeaseStore`

These are conveniences, not the architectural boundary. You can replace them for Postgres, S3, GCS, Redis, or a domain-specific store.

Multi-Node Ownership

Revision checks prevent stale writes but do not guarantee single active execution. For that, use WorkflowLeaseStore and WorkflowLeasePolicy.

val persistence = WorkflowPersistence(
    checkpointStore = MyCheckpointStore(client),
    stateCodec = ReviewStateCodec,
    leaseStore = MyLeaseStore(client),
    leasePolicy = WorkflowLeasePolicy(
        ownerId = "worker-7",
        leaseDurationMillis = 30_000,
    ),
)

Lease semantics

run(...) claims the workflow before checkpointing starts
resume(...) claims the workflow before resumed execution starts
each successful checkpoint write renews the lease with the latest revision
success or failure releases the lease

Lease Store Contract

class MyLeaseStore(
    private val client: LeaseClient,
) : WorkflowLeaseStore {
    override suspend fun currentLease(
        workflowName: String, workflowId: String,
    ): WorkflowLease? = client.current(workflowName, workflowId)

    override suspend fun claim(
        workflowName: String, workflowId: String,
        ownerId: String, checkpointRevision: Long?,
        leaseDurationMillis: Long,
    ): WorkflowLease = client.claim(workflowName, workflowId, ownerId, checkpointRevision, leaseDurationMillis)

    override suspend fun renew(
        lease: WorkflowLease, checkpointRevision: Long?,
        leaseDurationMillis: Long,
    ): WorkflowLease = client.renew(lease, checkpointRevision, leaseDurationMillis)

    override suspend fun release(lease: WorkflowLease) {
        client.release(lease)
    }
}

If another executor already owns the workflow, the store should fail with WorkflowLeaseConflictException.

File-Backed Leases

For local tooling or single-filesystem deployments:

val persistence = WorkflowPersistence(
    checkpointStore = FileWorkflowCheckpointStore(Path.of(".tramai/workflows")),
    stateCodec = ReviewStateCodec,
    leaseStore = FileWorkflowLeaseStore(Path.of(".tramai/workflows")),
    leasePolicy = WorkflowLeasePolicy(
        ownerId = "worker-local-1",
        leaseDurationMillis = 30_000,
    ),
)

Good for: one host or shared filesystem, inspectable artifacts, simple deployments. Not for: distributed deployments with weak filesystem semantics.

JDBC Lease Guidance

For multi-node backends centered on a relational database:

CREATE TABLE tramai_workflow_lease (
    workflow_name VARCHAR(255) NOT NULL,
    workflow_id VARCHAR(255) NOT NULL,
    lease_id VARCHAR(255) NOT NULL,
    owner_id VARCHAR(255) NOT NULL,
    checkpoint_revision BIGINT NULL,
    acquired_at_epoch_millis BIGINT NOT NULL,
    expires_at_epoch_millis BIGINT NOT NULL,
    PRIMARY KEY (workflow_name, workflow_id)
);

Key operational rule: lease ownership is time-bound. An active row with expires_at_epoch_millis > now blocks another owner. An expired row may be replaced.

Reference SQL patterns:

Claim — insert with PK. If PK exists and not expired, fail with conflict.

Renew — update where lease_id = ? AND owner_id = ? AND expires_at_epoch_millis > ?. Zero rows means conflict.

Release — delete where workflow_name, workflow_id, lease_id, owner_id.

Implementation notes:

index (owner_id) for operator visibility
use the same clock source for claim and renewal decisions
keep lease duration larger than the expected time between checkpoint writes
include checkpoint_revision for operator correlation

Choosing a Backend

Use case	Recommended
Transactional app with relational DB	JDBC stores
Local tooling, dev environments	File stores
Shared filesystem ownership	File lease store
Audit trail in human-readable form	Markdown checkpoint store
Large payloads, cheap snapshots	Custom object-store backend

Limitations

checkpoints happen only at top-level workflow step boundaries
nested branch internals are not resumed mid-step
in-flight parallel work is not resumed mid-branch
partially emitted provider streams are not resumed token-by-token

Next Steps

Orchestration — workflow DSL, step types, engine boundary
Scheduling — cron execution and delay steps
Workflow Server — expose persistent workflows over HTTP