Host Strata in a service

This guide runs a graph inside a long-running service. You register a store and a graph, read settings from configuration, run the background drivers, add health checks and telemetry, call the graph from a request handler, and stop without cutting a run short.

In .NET, Datalium.Strata.Hosting plugs into the Generic Host with AddStrata(). In Rust, datalium-strata-host has no container: you open the store and compile the graph yourself, and a Host runs the drivers as tasks. The examples use a refund graph: Prepare runs, the thread waits for approval, then Pay runs.

Before you start

• The packages from Installation. In .NET: Datalium.Strata.Hosting, plus Microsoft.Extensions.Hosting for a worker or the ASP.NET Core SDK for a web service. In Rust: datalium-strata, datalium-strata-memory, datalium-strata-host, serde_json, and tokio with the rt-multi-thread, macros and signal features.
• A checkpoint store. The examples use SQLite because it needs no server. Choose a store compares the options.
• Durable execution explains what the wake and recovery drivers do.

Register the store, the graph and the drivers

C#

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddStrata()
    .AddSqliteCheckpoints()                  // Strata:Sqlite
    .AddInstrumentation()
    .AddHealthChecks()
    .AddGraph(_ => Refunds.Build())          // compiled when the host starts
    .AddWakeDriver<Refund>()                 // Strata:Wake
    .AddRecoveryDriver<Refund>()             // Strata:Recovery
    .AddRetention();                         // Strata:Retention

builder.Services.AddSingleton<RefundDesk>();

using var host = builder.Build();
await host.RunAsync();

public sealed record Refund(string Customer, decimal Amount, string Status);

public static class Refunds
{
    public static CompiledGraph<Refund> Build() =>
        new StateGraph<Refund>()
            .AddNode("Prepare", refund => refund with { Status = "awaiting approval" })
            .AddInterruptBoundary("Prepare")
            .AddNode("Pay", refund => refund with { Status = "paid" })
            .AddEdge("Prepare", "Pay")
            .Compile(new CompilationOptions { Checkpointing = CheckpointPolicy.EveryNode });
}

AddSqliteCheckpoints(), AddPostgresCheckpoints(), AddRedisCheckpoints() or AddInMemoryCheckpoints() registers one checkpoint store under every interface it implements. AddCheckpointStore(factory) registers your own. AddGraph(factory) registers the compiled graph and a StrataRunner<TState>. A container holds one checkpoint store, and one graph per state type.

When the host starts, it migrates the SQL stores, compiles every graph and checks that each driver has what it needs. If one of these fails, StartAsync throws and no driver starts. A graph with a structural error throws GraphCompilationException.

Rust

#[tokio::main]
async fn main() {
    let text = std::fs::read_to_string("strata.json").expect("strata.json is readable");
    let settings: StrataSettings = serde_json::from_str(&text).expect("the settings parse");

    let store = Arc::new(settings.sqlite.open().expect("the store opens"));
    let graph = Arc::new(build());
    let config = settings.runs.config(); // the lease, when runs.lease.ttl is set

    let host = Host::new()
        .wake(graph.clone(), store.clone(), settings.wake.clone().with_config(config.clone()))
        .recover(graph.clone(), store.clone(), settings.recovery.clone().with_config(config.clone()))
        .retain(store.clone(), settings.retention.clone());
    let status = host.status(); // take it before run() consumes the host

    let shutdown = CancellationToken::new();
    let running = tokio::spawn(host.run(shutdown.clone()));

    // Start your own server or consumer here, with graph, store and config.

    tokio::signal::ctrl_c().await.expect("listens for Ctrl+C");
    shutdown.cancel();
    match tokio::time::timeout(Duration::from_secs(30), running).await {
        Ok(Ok(Ok(()))) => {}
        Ok(Ok(Err(refused))) => eprintln!("the host didn't start: {refused}"),
        Ok(Err(panicked)) => eprintln!("the host task panicked: {panicked}"),
        Err(_) => eprintln!("a run outlasted the shutdown timeout"),
    }
}

fn build() -> CompiledGraph<Refund> {
    StateGraph::<Refund>::new()
        .add_node("Prepare", |r: Refund| Refund { status: "awaiting approval".into(), ..r })
        .add_interrupt_boundary("Prepare")
        .add_node("Pay", |r: Refund| Refund { status: "paid".into(), ..r })
        .add_edge("Prepare", "Pay")
        .compile_with(CompileOptions::new().with_checkpointing(CheckpointPolicy::EveryNode))
        .expect("the graph is valid")
}

StrataSettings is a set of serde structs, so you read it from any format: JSON, TOML, or the environment through a config crate. Each store’s section has open(). open() and compile_with return their errors where you call them.

Host::run runs the drivers until the token fires. It checks every driver first: if one lacks what it needs, run returns Err(HostError) and starts nothing. Host::task(name, f) runs a loop of your own, such as a queue consumer, beside the drivers.

When the .NET host is up, its log shows one line per driver. The Rust host writes the same events through tracing.

Strata driver wake:Refund started, every 00:00:05
Strata driver recovery:Refund started, every 00:00:30
Strata driver retention started, every 01:00:00

Configure Strata from settings

C#

The options come from the Strata section of configuration. Durations are TimeSpan strings, such as 00:00:30 or 30.00:00:00 for 30 days.

{
  "Strata": {
    "Sqlite": { "Path": "strata.db" },
    "Runs": {
      "Lease": { "Ttl": "00:00:30" }
    },
    "Wake": { "Interval": "00:00:05" },
    "Recovery": { "Interval": "00:00:30" },
    "Retention": { "Interval": "01:00:00", "OlderThan": "30.00:00:00" }
  }
}

Each registration also takes a delegate, applied over configuration: AddRetention(o => o.OlderThan = TimeSpan.FromDays(30)). Configuration is bound by reflection, so if you publish with trimming, use delegates.

AddPostgresCheckpoints() reads the connection string ConnectionStrings:strata by default, and AddRedisCheckpoints() reads ConnectionStrings:strata-redis. Either store can also use an NpgsqlDataSource or IConnectionMultiplexer you registered.

Rust

The keys are snake_case. A duration is a number of seconds, or a number with ms, s, m, h or d.

{
  "sqlite": { "path": "strata.db" },
  "runs": { "lease": { "ttl": "30s" } },
  "wake": { "interval": "5s" },
  "recovery": { "interval": "30s" },
  "retention": { "interval": "1h", "older_than": "30d" }
}

For Postgres, set postgres.url and call settings.postgres.open().await. For Redis, set redis.url. Every section has defaults, so a file only says what differs.

The sections are the same in both languages:

Section (.NET / Rust)	What it sets
Sqlite / sqlite	Path, the database file, plus group commit, deltas and offload.
Postgres / postgres	The connection, AutoMigrate (on by default), retries, group commit, deltas and offload.
Redis / redis	The connection, the key Prefix, FinishedThreadTtl, retries, deltas and offload.
Runs / runs	Lease: the Owner (machine name and process ID by default) and Ttl. Without a ttl, runs take no lease.
Wake / wake	The wake driver: Interval, BatchSize, Metadata, MatchGraph, Priority.
Recovery / recovery	The recovery driver: the same, plus StaleAfter and MaxRecoveries.
Retention / retention	The retention driver: Interval, BatchSize, OlderThan, States.

With AutoMigrate off, the Postgres store only checks its tables, and a database that is behind stops the host from starting. Run the migration as a deploy step.

Use the graph in your code

C#

Inject StrataRunner<TState> wherever you start or answer a thread:

public sealed class RefundDesk(StrataRunner<Refund> runner)
{
    public Task<GraphExecutionResult<Refund>> StartAsync(string id, Refund refund, CancellationToken ct = default) =>
        runner.RunAsync(id, refund, cancellationToken: ct);

    public Task<GraphExecutionResult<Refund>> ApproveAsync(string id, string reviewer, CancellationToken ct = default) =>
        runner.ResumeAsync(id, HumanSignal.Approve(reviewer), cancellationToken: ct);
}

The runner holds the graph (runner.Graph), the store (runner.Store) and the host’s run defaults, and has RunAsync, ResumeAsync, SignalAsync, RecoverAsync and EndAsync. Its runs get the lease from Strata:Runs:Lease, the container’s clock, the registered history store and observer, and the container as NodeContext.Services. runner.CreateOptions() returns these defaults, for example to add metadata.

You can also inject CompiledGraph<Refund> and ICheckpointStore directly. Then you pass run options yourself, and runs take no lease unless you add one.

Rust

Keep the graph, the store and the run config together in a struct of your own, and clone it into whatever needs it:

#[derive(Clone)]
pub struct Refunds {
    graph: Arc<CompiledGraph<Refund>>,
    store: Arc<SqliteCheckpointStore>,
    config: RunConfig,
}

impl Refunds {
    pub async fn start(&self, id: &str, refund: Refund) -> GraphExecutionResult<Refund> {
        self.graph
            .run_with(id, refund, self.store.clone(), self.config.clone())
            .await
    }

    pub async fn approve(&self, id: &str, reviewer: &str) -> GraphExecutionResult<Refund> {
        self.graph
            .resume_with(id, HumanSignal::approve(reviewer), self.store.clone(), self.config.clone())
            .await
    }
}

Use the same config for your own runs and for the drivers (with_config), so that every run on a thread takes the lease. Add services for your nodes with config.with_service(...).

Run the background drivers

The drivers are the loops you would otherwise write yourself. Each one runs a round, waits its interval, and goes again.

Driver	.NET	Rust	Default interval	What it does
Wake	AddWakeDriver<TState>()	Host::wake	5 seconds	Resumes timers that are due and times out waits whose timeout has passed.
Recovery	AddRecoveryDriver<TState>()	Host::recover	30 seconds	Continues running threads whose process died.
Retention	AddRetention()	Host::retain	1 hour	Deletes completed and ended threads whose latest checkpoint is older than OlderThan.

• The first round runs when the host starts, so a host that comes back after a crash recovers at once.
• A round that fills its batch goes again without waiting. The interval is the wait once a round has caught up.
• A round that fails is logged, counted and tried again after a wait that doubles, up to 5 minutes. A driver never stops the host.
• Every replica can run every driver. Wakes and recoveries claim each thread with a conditional save, and purging twice deletes nothing more.

What the drivers need

The host refuses to start a driver that can’t work. Check these before you deploy:

• Wake and recovery: a graph registered for the state type (in .NET, AddGraph<TState>), and a store that lists threads. The SQLite, Postgres, Redis and in-memory stores do.
• Recovery: a graph compiled with CheckpointPolicy.EveryNode. Under the default policy, a crashed run leaves nothing to recover.
• Recovery: a lease ttl in Runs:Lease (in Rust, passed to the driver with with_config), or a StaleAfter longer than your slowest node. Without leases, the age of the latest checkpoint is the only sign that a run died. See how Strata decides a thread is dead.
• Retention: an OlderThan. There is no default: you decide how long you keep threads.
• Retention: a store that purges. The SQLite, Postgres, Redis and in-memory stores do. The store for the Strata Cloud private beta doesn’t.
• All drivers: an interval above zero, and a batch size from 1 to 1000.

To see the wake driver’s engine call on its own, run demo 19, waits and timers (C#, Rust).

Check health

C#

AddHealthChecks() adds two checks to the host’s health check service:

• strata-checkpoints loads the latest checkpoint of a thread nobody has (strata.health). An answer is healthy. A failure, or no answer within 5 seconds, is unhealthy. The description is the exception’s type, not its message, because a store’s message can hold a connection string.
• strata-drivers is degraded while any driver’s last round failed, and names those drivers. It is never unhealthy, because a failing driver keeps the host running.

AddHealthChecks(tags, timeout) sets tags and a different timeout. In ASP.NET Core, expose the checks with app.MapHealthChecks("/health"). To read the drivers’ state yourself, inject IStrataDriverStatus.

Rust

check_store runs the same probe as .NET, and Host::status() says how each driver stands. Combine them in your readiness endpoint:

let ready = match check_store(&*store, Duration::from_secs(5)).await {
    StoreHealth::Healthy { .. } => status.failing().is_empty(),
    StoreHealth::Unhealthy { .. } => false,
};

StoreHealth::Unhealthy carries the store’s own message, which can hold a connection string. Don’t put it on a public endpoint. status.drivers() lists every driver with its rounds, failures in a row, last round and last failure.

Send telemetry to OpenTelemetry

C#

AddInstrumentation() adds Strata’s activity sources and meters to the OpenTelemetry providers you set up. It creates no provider and exports nothing, so you still add OpenTelemetry and an exporter:

builder.Services.AddOpenTelemetry()
    .WithTracing(tracing => tracing.AddConsoleExporter())
    .WithMetrics(metrics => metrics.AddConsoleExporter());

The names, if you wire them by hand (StrataInstrumentation.ActivitySourceNames and StrataInstrumentation.MeterNames):

• Activity sources: Datalium.Strata, Datalium.Strata.Cloud.
• Meters: Datalium.Strata, Datalium.Strata.Storage, Datalium.Strata.Cloud, Datalium.Strata.Hosting.

The drivers’ own metrics are strata.host.driver.rounds, by driver and outcome, and strata.host.driver.threads, the threads woken, recovered or purged.

Rust

There is no instrumentation call. Strata writes spans and events through tracing and metrics through metrics, which are global facades. Install a tracing subscriber and a metrics recorder in your application, and Strata’s spans and metrics arrive in them.

The drivers count strata.host.driver.rounds, by strata.driver and strata.outcome, and strata.host.driver.threads, by strata.driver.

Run the graph from a request handler

A handler starts a thread with an ID it chooses, or resumes one by ID. The thread lives in the store, so any replica can take the next request about it.

C#

With ASP.NET Core minimal APIs, the handler asks for the runner as a parameter:

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddStrata()
    .AddSqliteCheckpoints()
    .AddHealthChecks()
    .AddGraph(_ => Refunds.Build())
    .AddWakeDriver<Refund>()
    .AddRecoveryDriver<Refund>();

var app = builder.Build();

app.MapPost("/refunds/{id}", async (string id, Refund refund, StrataRunner<Refund> runner) =>
    Describe(await runner.RunAsync(id, refund)));

app.MapPost("/refunds/{id}/approve", async (string id, string reviewer, StrataRunner<Refund> runner) =>
    Describe(await runner.ResumeAsync(id, HumanSignal.Approve(reviewer))));

app.MapHealthChecks("/health");

app.Run();

static IResult Describe(GraphExecutionResult<Refund> result) => result switch
{
    GraphExecutionResult<Refund>.Completed done => Results.Ok(done.State),
    GraphExecutionResult<Refund>.Interrupted waiting => Results.Accepted(value: new { waitingAt = waiting.SuspendedNode }),
    GraphExecutionResult<Refund>.Failed failed => Results.Conflict(new { error = failed.Message }),
    _ => Results.StatusCode(StatusCodes.Status503ServiceUnavailable),
};

Rust

Give your web framework’s state a clone of Refunds from Use the graph in your code, and call it from the handler:

let started = refunds
    .start("refund-1042", Refund { customer: "acme".into(), amount: 120.0, status: "new".into() })
    .await;
// Interrupted: the thread waits for approval.

let approved = refunds.approve("refund-1042", "jdoe").await;
// Completed(Refund { status: "paid", .. })

Map GraphExecutionResult to your response: Completed to success, Interrupted to “accepted, waiting”, and Failed to an error.

The first request returns 202 Accepted with {"waitingAt":"Prepare"}. The approval returns 200 OK with the state, "status":"paid". A second approval of the same thread returns 409 Conflict, because a completed thread can’t be resumed.

The handlers don’t pass the request’s cancellation token to the run. A client that disconnects would otherwise cancel the run and leave the thread cancelled.

Stop the service

When the host stops, each driver starts nothing more and lets the run in flight finish. Nothing is cancelled. A driver that is waiting between rounds stops at once. This covers the runs the drivers start. Runs your own code starts end as your code decides.

C#

The Generic Host waits for the drivers up to its shutdown timeout, 30 seconds by default. If your nodes can take longer, raise it:

builder.Services.Configure<HostOptions>(options => options.ShutdownTimeout = TimeSpan.FromMinutes(2));

Rust

Cancel the token you passed to Host::run, then await the task. run returns once every driver has let its round finish. It has no timeout of its own: put one around the await, as the main above does with tokio::time::timeout.

If a run outlasts the timeout and the process exits, its thread stays running. With EveryNode and the recovery driver, the next host continues it.

Related

• Durable execution: timers, recovery, leases and the drivers at concept level.
• Choose a store: which checkpoint store to register.
• Human-in-the-loop: pausing a thread and resuming it.
• Architecture: where the hosting packages fit.
• Demo 28, hosting: a host with a wake driver, a clean stop and health checks under an outage (C#, Rust).
• design/35-Hosting-Integration.md: the design behind both packages.