Crate runledger_postgres 

PostgreSQL persistence layer for Runledger durable execution.

This crate owns the SQLx-backed storage and query helpers used by the runtime and application crates. The main entrypoint is the jobs module, which exposes APIs for:

• queueing, resource-aware claiming, heartbeating, and completing jobs
• typed direct-job recovery, successful-job replay, events, metrics, and admin reads
• enqueueing, actively coordinating, querying, and immutably recovering workflow runs and steps
• applying or validating the bundled Runledger schema migrations

Typical consumers share a DbPool with runledger-runtime, then call the exported jobs functions from application setup, admin APIs, or tests.

Security Boundary

This crate is a persistence layer, not an authentication or authorization layer. Public job and workflow APIs that accept organization_id, idempotency keys, workflow IDs, job IDs, or metadata expect those values to come from a trusted service boundary. HTTP or RPC handlers should derive organization scope from authenticated claims or server-side policy, not from untrusted request parameters alone.

QueryError::client_message and QueryError::code are the stable values intended for public error responses and application logs. Detailed internal context remains available through QueryError::internal_message and the standard error source chain for targeted debugging, but may contain payload values, idempotency keys, or database policy details. Do not emit it at general logging boundaries without explicit redaction. Public formatting keeps raw SQLx details sanitized. Runtime lifecycle, workflow mutation, and idempotent enqueue APIs are designed for PostgreSQL’s default READ COMMITTED transaction isolation so they can observe rows committed after lock waits or uniqueness conflicts. Release-sensitive, workflow-append, and keyed-enqueue paths validate this before running because their correctness depends on second reads after waits or conflicts.

For simple embedding, call migrate_after_idempotency_cutover during startup:

let pool = sqlx::PgPool::connect("postgres://localhost/runledger").await?;
runledger_postgres::migrate_after_idempotency_cutover(&pool).await?;

For deployments that manage DDL elsewhere, call ensure_schema_compatible_after_idempotency_cutover instead to fail fast if the schema is missing, drifted, or still has keyed legacy rows without idempotency request snapshots. That check is read-only, but it expects the database to retain SQLx migration history in _sqlx_migrations and, when available, Runledger-owned migration state in runledger_migration_history.

Copy-Paste Examples

• Enqueue one job
• Enqueue a workflow DAG
• Use an external workflow gate
• Create a scheduled job entrypoint
• Adopt continuation, coordination, replay, and recovery

Import runledger_runtime::prelude::* and use the worker binary example when adding a worker process for the jobs and workflows enqueued through this crate.

Prelude

use runledger_core::prelude::*;
use runledger_postgres::prelude::*;

The PostgreSQL prelude exports persistence functions and record/input types. It intentionally does not re-export core contract types, so import runledger_core::prelude::* beside it when building job or workflow inputs.

Enqueue One Job

Use direct job enqueue for one independent retried unit of work.

use runledger_core::prelude::*;
use runledger_postgres::prelude::*;

let payload = serde_json::json!({"email_id": "email_123"});
let job = JobEnqueue {
    job_type: JobType::new("jobs.email.send"),
    organization_id: None,
    payload: &payload,
    priority: None,
    max_attempts: None,
    timeout_seconds: None,
    next_run_at: None,
    idempotency_key: Some("email:email_123:send"),
    stage: None,
};

let _job_id = enqueue_job(&pool, &job).await?;

Record A Durable Transactional Handoff

Use an enqueue intent when application state and a future job request must commit atomically before the job definition is available. Recording does not read job_definitions or create a queue row. A standard worker later promotes pending intents for its registered job types through the ordinary enqueue path.

use runledger_core::prelude::*;
use runledger_postgres::prelude::*;

let payload = serde_json::json!({"invoice_id": "invoice_123"});
let intent = JobEnqueueIntent::new(
    JobType::new("billing.invoice.capture"),
    &payload,
    "invoice:invoice_123:capture",
);

let mut tx = pool.begin().await?;
// Persist the application's business/audit mutation with this same `tx`.
let outcome = record_job_enqueue_intent_tx(&mut tx, &intent).await?;
if outcome.status() == JobEnqueueIntentStatus::Conflicted {
    return Err("the existing durable handoff is conflicted".into());
}
tx.commit().await?;

The returned status is a point-in-time observation, not a promotion guarantee. An existing intent can be promoted or become conflicted concurrently, including while a caller-owned record transaction remains open. Continue monitoring pending age and conflicts after accepting a pending handoff.

Concurrent calls that record the same (job_type, organization_id, idempotency_key) may wait for the transaction that first claimed that unique key. Include that wait in the caller-owned transaction’s lock ordering and timeout budget. Call jobs::record_job_enqueue_intent_tx before any operation in the same transaction that can lock a job_queue row. A job-first recorder can create an inverse lock cycle with retention’s canonical intent-before-job order.

Intent payloads and idempotency keys cross the same trusted persistence boundary as ordinary queue inputs. Do not place secrets in them or emit them in logs. Monitor jobs::get_job_enqueue_intent_metrics for pending age, retrying count, maximum promotion attempts, and conflicts during the preceding 24 hours; Runledger never automatically deletes conflicted intent evidence. Database-level promotion failures leave the intent pending with bounded jittered exponential backoff, and read APIs expose the attempt/error metadata. They retry indefinitely so a prolonged outage cannot silently discard work; operators must alert on pending age and maximum attempts. Promoted intents retain their linked jobs. A queue-retention transaction must call jobs::delete_promoted_job_enqueue_intents_for_jobs_tx with its exact selected job IDs before deleting those jobs. Select candidate IDs without row locks, then call the helper as the transaction’s first lock-taking operation. The transaction must use READ COMMITTED; stronger isolation levels return job.intent_retention_unsupported_isolation before fence acquisition. The helper waits for active promotions, fences new promotions, deletes promoted-intent links, then locks those jobs for the rest of the transaction. This intent-before-job order composes with duplicate recorders without an inverse lock cycle. Keep the transaction short and commit promptly. Time cutoffs alone are insufficient because a newly promoted intent may link to an older existing job.

Create A Scheduled Job Entrypoint

Use jobs::JobScheduleUpsert to create or update the cron row consumed by the runtime scheduler. Schedules are UTC-only. Updating an existing schedule refreshes its definition while preserving is_active and organization_id; next_fire_at is refreshed when cron_expr changes. Cron expressions are validated with the same parser used by runledger-runtime. Use jobs::set_job_schedule_active to pause or resume a schedule, and jobs::set_job_schedule_next_fire_at to manually retime its cursor.

use chrono::Utc;
use runledger_core::prelude::*;
use runledger_postgres::prelude::*;

let payload_template = serde_json::json!({"source": "api"});
let schedule = JobScheduleUpsert {
    name: "profile-refresh-hourly",
    job_type: JobType::new("profiles.refresh"),
    organization_id: None,
    payload_template: &payload_template,
    cron_expr: "0 0 * * * *",
    is_active: true,
    next_fire_at: Utc::now(),
    max_jitter_seconds: 0,
};

let _schedule = upsert_job_schedule(&pool, &schedule).await?;

Enqueue A Workflow DAG

Use workflows when the work has step dependencies, fan-out/fan-in, external gates, cancellation as one logical run, or workflow-level idempotency.

use runledger_core::prelude::*;
use runledger_postgres::prelude::*;

let crawl_payload = serde_json::json!({"profile_id": "p_123"});
let classify_payload = serde_json::json!({"profile_id": "p_123"});
let metadata = serde_json::json!({"source": "api"});

let run = WorkflowDagBuilder::new("profiles.research", &metadata)
    .idempotency_key("profile:p_123:research")
    .job("crawl", "profiles.crawl", &crawl_payload)?
    .job("classify", "profiles.classify", &classify_payload)?
    .after_success("classify", ["crawl"])?
    .build()?;

let _workflow_run = enqueue_workflow_run(&pool, &run).await?;

Use An External Workflow Gate

Create the gate with WorkflowStepEnqueueBuilder::new_external, then complete it from a trusted service boundary when the external condition is known.

use runledger_core::prelude::*;
use runledger_postgres::prelude::*;

let input = CompleteExternalWorkflowStepInput {
    workflow_run_id,
    organization_id: None,
    step_key: StepKey::new("approval"),
    outcome: ExternalWorkflowStepTerminalOutcome::Succeeded { output: None },
    status_reason: Some("approved"),
    last_error_code: None,
    last_error_message: None,
};

let _step = complete_external_workflow_step(&pool, &input).await?;

Inspect Workflow State

use runledger_postgres::prelude::*;

let scope = WorkflowRunReadScope::Admin;
let _run = get_workflow_run_by_id_with_scope(&pool, scope, workflow_run_id).await?;
let _steps = list_workflow_steps_with_scope(&pool, scope, workflow_run_id).await?;
let _dependencies =
    list_workflow_step_dependencies_with_scope(&pool, scope, workflow_run_id).await?;

Coordinate And Recover Durable Work

• Use jobs::enqueue_job_with_execution_resource or runledger_core::jobs::WorkflowStepEnqueueBuilder::execution_resource for lease-scoped single-permit resources.
• Use jobs::enqueue_or_get_active_workflow with a workflow runledger_core::jobs::WorkflowRunEnqueueBuilder::active_key and inspect every jobs::EnqueueActiveWorkflowOutcome.
• Use jobs::compare_and_requeue_job for a failed/canceled/dead-lettered direct job and jobs::compare_and_replay_succeeded_job for intentional successful replay.
• Use jobs::recover_workflow_run to reconstruct a terminal workflow as a new lineage-linked run rather than rewriting source history.

The caller-transaction variants of recovery and replay require PostgreSQL READ COMMITTED. See the downstream guide for rollout fences, request idempotency, and retention behavior.

Handle Errors Safely

QueryError::client_message and QueryError::code are safe for public responses. QueryError::internal_message is for trusted diagnostics.

match runledger_postgres::jobs::enqueue_job(&pool, &job).await {
    Ok(_job_id) => {}
    Err(runledger_postgres::Error::QueryError(query_error)) => {
        let _public_code = query_error.code();
        let _public_message = query_error.client_message();
        let _private_diagnostic = query_error.internal_message();
    }
    Err(error) => return Err(error),
}

Modules

jobs

Durable job, schedule, workflow, and admin persistence APIs.

prelude

Common runledger-postgres imports for integration crates.

Structs

FrameworkConstraintSpec

QueryError

WorkflowJobLinkTriggerDiagnostic

Validation details for one unsafe expand-window workflow/job-link trigger.

Enums

Error

QueryErrorCategory

QueryErrorKind

Stable semantic kinds for database errors that drive runtime policy.

SchemaCompatibilityError

WorkflowJobLinkTriggerProblem

One reason an expand-window workflow/job-link trigger is unsafe.

Statics

MIGRATOR

Raw SQLx migrator for inspecting the migrations bundled with this crate version.

Functions

classify_framework_constraint

classify_query_error

classify_query_error_with_constraint_classifier

ensure_schema_compatibleDeprecated

Validate that the target database’s SQLx migration history matches the bundled Runledger migrations.

ensure_schema_compatible_after_idempotency_cutover

Validate that the target database’s SQLx migration history matches the bundled Runledger migrations.

has_framework_constraint_classifier

migrateDeprecated

Apply the bundled Runledger schema migrations to a PostgreSQL pool.

migrate_after_idempotency_cutover

Apply the bundled Runledger schema migrations to a PostgreSQL pool, then enforce the idempotency snapshot cutover.

Type Aliases

DbPool

DbTx

Result