Use Postgres

This guide stores your threads and chat history in Postgres. You open the stores, choose how they connect and when their tables are migrated, tune the pool, retries and group commit, register them in a host, and resume a thread in the other language.

Choose a store says when Postgres is the right store.

Before you start

• The packages from Installation. In .NET, the Postgres stores are in Datalium.Strata.Storage, which Datalium.Strata.Hosting brings. In Rust, they are in datalium-strata-memory, and you need tokio.
• A Postgres server, and a user that can create tables in the database. CI tests the stores against Postgres 16. For development, the docker-compose.yml in datalium/strata starts a throwaway one on localhost:5432, with the pgvector/pgvector:pg16 image (see Running Postgres and Redis locally).

Open the stores

The checkpoint store keeps your threads. The history store keeps chat messages by session. Both take the same URL. This example opens both, runs a refund graph until it waits for approval, and appends one message:

C#

using Datalium.Strata.Graph;
using Datalium.Strata.Interruption;
using Datalium.Strata.Memory;
using Datalium.Strata.Storage.Postgres;

var url = "postgres://strata:strata@localhost:5432/strata";
await using var checkpoints = await PostgresCheckpointStore.OpenAsync(url);
await using var history = await PostgresHistoryStore.OpenAsync(url);

var result = await Refunds.Build().RunAsync("refund-1042", new Refund("acme", 120m, "new"), checkpoints);
await history.AppendAsync("refund-1042",
    new ChatMessage(MessageRole.User, "Refund order 1042".AsMemory(), DateTimeOffset.UtcNow));

Console.WriteLine(result is GraphExecutionResult<Refund>.Interrupted waiting
    ? $"refund-1042 waits after {waiting.SuspendedNode}"
    : $"unexpected: {result}");

Rust

use std::sync::Arc;

use chrono::Utc;
use datalium_strata::checkpoint::GraphExecutionResult;
use datalium_strata::history::{ChatMessage, HistoryStore, MessageRole};
use datalium_strata_memory::postgres_store::{PostgresCheckpointStore, PostgresHistoryStore};

let url = "postgres://strata:strata@localhost:5432/strata";
let checkpoints = Arc::new(PostgresCheckpointStore::connect(url).await.map_err(|e| e.0)?);
let history = PostgresHistoryStore::connect(url).await.map_err(|e| e.0)?;

let initial = Refund { customer: "acme".into(), amount: 120.0, status: "new".into() };
let result = build().run("refund-1042", initial, checkpoints).await;
history
    .append("refund-1042", ChatMessage {
        role: MessageRole::User,
        content: Arc::from("Refund order 1042"),
        timestamp: Utc::now(),
    })
    .await
    .map_err(|e| e.0)?;

match result {
    GraphExecutionResult::Interrupted { suspended_node, .. } => {
        println!("refund-1042 waits after {suspended_node}")
    }
    _ => println!("unexpected result"),
}

The graph is the refund graph from Hosting: Prepare runs, the thread waits, then Pay runs. Both programs print:

refund-1042 waits after Prepare

OpenAsync (C#) and connect (Rust) connect and create the tables before they return, so a server that can’t be reached fails at startup. A .NET store you create with new prepares its tables on its first call instead.

To see the thread, point the strata CLI at the same URL:

strata thread list --storage postgres://strata:strata@localhost:5432/strata

+-------------+--------------+-------------+-------------+-------------+----------------------+
| Thread ID   | Current Node | Sequence ID | State       | Waiting For | Updated At           |
+-------------+--------------+-------------+-------------+-------------+----------------------+
| refund-1042 | Prepare      | 0           | interrupted | -           | 2026-10-30T14:44:59Z |
+-------------+--------------+-------------+-------------+-------------+----------------------+

Connection strings and URLs

Both languages take a URL of the form postgres://user:password@host:port/database. postgresql:// works too.

• .NET also takes an Npgsql connection string, such as Host=db;Username=strata;Password=...;Database=strata, and passes it to Npgsql unchanged. In a URL, the only query parameter it accepts is sslmode, for example ?sslmode=require. Any other parameter is an error: Unsupported query parameter 'connect_timeout' in the postgres URL; use an Npgsql connection string instead. A URL gets a 5-second connect timeout.
• Rust parses the URL with tokio-postgres, so its query parameters work, such as connect_timeout or application_name. Without connect_timeout, the store waits 5 seconds for a connection.

The stores create their tables in the first schema of the search path. To use another schema, put Search Path=app in a .NET connection string, or ?options=-c%20search_path%3Dapp in a URL for Rust and the CLI.

TLS and your own connection

In .NET, sslmode in the URL turns on TLS. For more control, such as certificates, build your own NpgsqlDataSource and give it to the stores. Rust’s stores don’t connect over TLS. The workaround is to connect a tokio_postgres::Client with your own TLS connector and pass it to from_client:

C#

using Npgsql;

var builder = new NpgsqlDataSourceBuilder(
    "Host=db.internal;Username=strata;Password=...;Database=strata;SSL Mode=VerifyFull;Maximum Pool Size=20");
await using var dataSource = builder.Build();

await using var checkpoints = new PostgresCheckpointStore(dataSource);
await using var history = new PostgresHistoryStore(dataSource);
await checkpoints.MigrateAsync();
await history.MigrateAsync();

Both stores share the data source and its pool. A store doesn’t dispose a data source you give it, unless you pass ownsDataSource: true.

Rust

let connector = postgres_native_tls::MakeTlsConnector::new(native_tls::TlsConnector::new()?);
let (client, connection) = tokio_postgres::connect(
    "host=db.internal user=strata password=... dbname=strata sslmode=require",
    connector,
)
.await?;
tokio::spawn(connection);
let checkpoints = PostgresCheckpointStore::from_client(client).await.map_err(|e| e.0)?;

A store built from a client has that one connection. It has no pool, and it can’t reconnect: once the connection drops, every later call fails. Recreate the store when that happens.

The strata CLI has no TLS either. See Known limitations for both.

Size the connection pool

Each store you open from a URL keeps its own pool of connections. A call takes a connection for each attempt and gives it back.

• .NET uses Npgsql’s pool, with Npgsql’s defaults. Set its size with Maximum Pool Size in a connection string. To let several stores share one pool, give them one NpgsqlDataSource, as above.
• Rust keeps up to 16 connections per store. Change it with PostgresStoreOptions::new().with_pool_size(n). A checkpoint store and a history store are two pools, up to 32 connections.

Add up the pools of every process against the server’s max_connections. With group commit on, a store runs up to 4 flushes at once, each on its own connection.

Decide when the tables are migrated

Each database records its layout in a strata_schema_version table, with one row per component (checkpoints, history). A store that opens a database brings it up to date in one transaction, under a lock that every store and both languages share, so many processes can start at once. A database that a newer version of Strata wrote is refused, and left as it is:

database is at checkpoints schema version 3, and this build knows 2: upgrade the runtime

There are no down migrations. Once a newer Strata has migrated a database, an older one refuses to open it. The strata CLI is one of them, and opening a database with the CLI migrates it too, so use a CLI of the same version as your services.

To migrate in a deploy step instead, turn AutoMigrate off in your application. Its stores then only check the tables:

C#

// The deploy step:
await using (var migrate = await PostgresCheckpointStore.OpenAsync(url)) { }
await using (var migrate = await PostgresHistoryStore.OpenAsync(url)) { }

// The application:
var options = new PostgresStoreOptions { AutoMigrate = false };
await using var checkpoints = new PostgresCheckpointStore(url, options);
var latest = await checkpoints.LoadAsync("refund-1042"); // checks the tables first

OpenAsync and MigrateAsync always migrate, whatever AutoMigrate says. In the application, create the store with new and let its first call check the tables.

Rust

// The deploy step:
PostgresCheckpointStore::connect(url).await.map_err(|e| e.0)?;
PostgresHistoryStore::connect(url).await.map_err(|e| e.0)?;

// The application:
let options = PostgresStoreOptions::new().with_auto_migrate(false);
let checkpoints = PostgresCheckpointStore::connect_with(url, options).await.map_err(|e| e.0)?;

connect_with checks the tables before it returns. migrate() on a store migrates whatever its options say.

If the deploy step hasn’t run, the application fails with:

database is at checkpoints schema version 0, and this build needs 2: run its migration

Run many writers

Many processes can write to one database. Before a save picks the next sequence number for a thread, it takes a transaction-scoped advisory lock on that thread, so a second writer to the same thread waits instead of failing. The lock key is computed on the server from the thread ID, so C# and Rust writers share it. Saves to different threads don’t wait for each other. The history store locks each session the same way.

The lock keeps saves in order. It doesn’t stop two workers from running the same thread at once. For that, use thread leases.

Set retries

The Postgres stores repeat a call that failed for a transient reason. By default, a call is made up to 4 times. The first repeat waits up to 50 ms, and each wait after it doubles, up to 2 s. Each wait is random within that limit, so stores that failed together don’t come back together.

A failure is transient when Postgres reports a connection error (SQLSTATE class 08), a shutdown or startup (57P01 to 57P03), a serialization failure (40001), a deadlock (40P01) or too many connections (53300), or when the connection breaks or times out. Anything else, such as a wrong password, fails at once.

Repeating a call is safe:

• A save that may have been committed is looked up before it is made again, so it is never written twice.
• A history append, or a lease acquisition, is repeated only if it failed before its commit was sent. After that, the failure is yours, because the store can’t tell whether it was written.
• A dropped connection is replaced. The pool opens a new connection for the next attempt.

Change the policy in the store’s options:

C#

using Datalium.Strata.Storage;
using Datalium.Strata.Storage.Postgres;

var options = new PostgresStoreOptions
{
    Retry = new StoreRetryPolicy { MaxAttempts = 6, MaxDelay = TimeSpan.FromSeconds(5) },
    GroupCommit = new GroupCommitOptions(),
};
await using var checkpoints = await PostgresCheckpointStore.OpenAsync(url, options);

StoreRetryPolicy.None turns retries off. A cancellation token you pass to a store call stops the wait between attempts.

Rust

use std::time::Duration;

use datalium_strata_memory::{GroupCommit, PostgresStoreOptions, StoreRetry};

let options = PostgresStoreOptions::new()
    .with_retry(StoreRetry::new().with_max_attempts(6).with_max_delay(Duration::from_secs(5)))
    .with_pool_size(32)
    .with_group_commit(GroupCommit::new());
let checkpoints = PostgresCheckpointStore::connect_with(url, options).await.map_err(|e| e.0)?;

StoreRetry::none() turns retries off.

Each repeat adds one to the strata.store.retries counter, by store and operation.

Turn on group commit

Group commit is off by default. The examples above turn it on with GroupCommit = new GroupCommitOptions() (C#) or with_group_commit(GroupCommit::new()) (Rust).

With it on, a save that arrives while the store’s flushes are all busy waits in a queue, and the next flush writes the waiting saves together, in one transaction. There is no timer: a save that finds a flush free is written at once. Each save is answered only after its flush commits, with the answer it would have had alone, and saves to one thread keep their order. A flush takes up to 256 saves (MaxBatch, with_max_batch), and up to 4 flushes run at once (MaxConcurrentFlushes, with_max_concurrent_flushes). The history store ignores the option.

It helps when one process saves for many runs at once. In the benchmarks, 64 writers saved 2.5 times as many checkpoints a second in Rust and 1.8 times as many in .NET. A single writer saved a little less with it on. See Performance for the numbers and their caveats.

Delta checkpoints and value offload go in the same options. Retention and storage size covers them, and purging old threads.

Register the stores in a host

In a host, the stores come from configuration. Hosting covers the rest.

C#

Register the stores with AddPostgresCheckpoints() and AddPostgresHistory(). They read the connection string ConnectionStrings:strata (a URL or an Npgsql connection string), and their options from Strata:Postgres:

{
  "ConnectionStrings": {
    "strata": "postgres://strata:strata@localhost:5432/strata"
  },
  "Strata": {
    "Postgres": {
      "AutoMigrate": true,
      "Retry": { "MaxAttempts": 6 },
      "GroupCommit": { "Enabled": true }
    }
  }
}

When the host starts, it migrates the tables, or only checks them with AutoMigrate off. A database that can’t be reached, or that fails the check, stops the host. To use an NpgsqlDataSource you registered, set UseRegisteredDataSource, or DataSourceKey for a keyed one.

Rust

Put a postgres section in your StrataSettings, then call settings.postgres.open().await for the checkpoint store and settings.postgres.open_history().await for the history store:

{
  "postgres": {
    "url": "postgres://strata:strata@localhost:5432/strata",
    "pool_size": 16,
    "retry": { "max_attempts": 6 },
    "group_commit": { "enabled": true }
  }
}

open() connects, and migrates or checks the tables, so an error comes back where you call it.

Share threads between C# and Rust

Both engines write the same tables, so a thread that one engine paused can be resumed by the other, and both read the same history. The state’s JSON field names must match. Pin the C# names with [JsonPropertyName], as Architecture explains:

C#

using System.Text.Json.Serialization;

var result = await Refunds.Build().ResumeAsync("refund-1042", HumanSignal.Approve("jdoe"), checkpoints);
if (result is GraphExecutionResult<Refund>.Completed done)
{
    Console.WriteLine($"refund-1042 is {done.State.Status}");
}

await foreach (var message in history.GetHistoryAsync("refund-1042", 10))
{
    Console.WriteLine($"{message.Role}: {message.Content}");
}

public sealed record Refund(
    [property: JsonPropertyName("customer")] string Customer,
    [property: JsonPropertyName("amount")] decimal Amount,
    [property: JsonPropertyName("status")] string Status);

Rust

use datalium_strata::checkpoint::HumanSignal;

match build().resume("refund-1042", HumanSignal::approve("jdoe"), checkpoints).await {
    GraphExecutionResult::Completed(refund) => println!("refund-1042 is {}", refund.status),
    _ => println!("unexpected result"),
}
for message in history.get_history("refund-1042", 10).await.map_err(|e| e.0)? {
    println!("{:?}: {}", message.role, message.content);
}

#[derive(Clone, Serialize, Deserialize)]
struct Refund {
    customer: String,
    amount: f64,
    status: String,
}

Start the thread with the first example on this page in one language, and resume it in the other. Either way, the resuming program prints:

refund-1042 is paid
User: Refund order 1042

Demo 12, pluggable storage, makes the same hand-off with its freeze and resume modes (C#, Rust).

Related

• Choose a store: how Postgres compares with the other stores.
• Retention and storage size: purging finished threads, delta checkpoints and value offload.
• Hosting: the host, its drivers and health checks.
• Durable execution: recovery and thread leases.
• Performance: the store benchmarks.
• design/34-Storage-Hardening.md and design/41-Checkpoint-Efficiency.md: the design behind retries, migrations, the pool and group commit.