Skip to content

OPERATIONS.md

Latest commit

 

History

History
466 lines (342 loc) · 13.8 KB

OPERATIONS.md

File metadata and controls

466 lines (342 loc) · 13.8 KB

Operations Guide

Runtime operations, observability, persistence, testing, and production deployment.

Event Streaming & Observability {#event-streaming}

Weavegraph provides multiple patterns for streaming workflow events with JSON serialization support.

Event Sinks

Built-in sinks for different use cases:

  • StdOutSink: Human-readable console output
  • MemorySink: In-memory capture for testing
  • ChannelSink: Async streaming to channels
  • JsonLinesSink: Machine-readable JSON Lines format

Events can be serialized to JSON using event.to_json_value(), event.to_json_string(), or event.to_json_pretty().

Simple Pattern (CLI Tools)

let (result, events) = app.invoke_with_channel(initial_state).await;

// Collect events while processing
tokio::spawn(async move {
    while let Ok(event) = events.recv_async().await {
        println!("Event: {:?}", event);
    }
});

Multiple Sinks

use weavegraph::event_bus::{StdOutSink, ChannelSink};

app.invoke_with_sinks(
    initial_state,
    vec![
        Box::new(StdOutSink::default()),
        Box::new(ChannelSink::new(tx)),
    ]
).await?;

Web Servers (SSE/WebSockets)

Use App::invoke_streaming to run a workflow while streaming events to clients. See STREAMING.md for full details.

Sink Diagnostics

Monitor event sink health and failures:

use weavegraph::event_bus::EventBus;

let bus = EventBus::with_sinks(vec![/* your sinks */]);

// Subscribe to diagnostics (optional)
let mut diags = bus.diagnostics();
tokio::spawn(async move {
    while let Ok(diagnostic) = diags.recv().await {
        eprintln!("Sink '{}' error: {}", diagnostic.sink, diagnostic.error);
    }
});

// Query health snapshot
let health = bus.sink_health();
for entry in health {
    println!("{}: {} errors", entry.sink, entry.error_count);
}

Tracing

Rich tracing integration with configurable log levels:

# Debug level for weavegraph modules
RUST_LOG=debug cargo run --example basic_nodes --features examples

# Error level globally, debug for weavegraph
RUST_LOG=error,weavegraph=debug cargo run --example advanced_patterns --features examples

Persistence {#persistence}

Weavegraph supports SQLite and PostgreSQL checkpointing, as well as in-memory state for workflows.

SQLite Checkpointing

Automatic state persistence with configurable database location:

use weavegraph::runtimes::{AppRunner, CheckpointerType};

// Using the builder pattern (recommended)
let runner = AppRunner::builder()
    .app(app)
    .checkpointer(CheckpointerType::SQLite)
    .build()
    .await;

SQLite URL resolution order (when CheckpointerType::SQLite is selected):

  1. WEAVEGRAPH_SQLITE_URL environment variable (full URL)
  2. SQLITE_DB_NAME environment variable (filename only; used as sqlite://{name})
  3. Default: sqlite://weavegraph.db

Tip: RuntimeConfig also loads .env automatically (via dotenvy) for local dev.

PostgreSQL Checkpointing

For production deployments requiring horizontal scaling or shared state:

use weavegraph::runtimes::{AppRunner, CheckpointerType};

// Using the builder pattern
let runner = AppRunner::builder()
    .app(app)
    .checkpointer(CheckpointerType::Postgres)
    .build()
    .await;

Database URL resolution order:

  1. WEAVEGRAPH_POSTGRES_URL environment variable
  2. DATABASE_URL environment variable (common convention)
  3. Fallback: postgresql://localhost/weavegraph

PostgreSQL vs SQLite:

Aspect SQLite PostgreSQL
Deployment Single-file, embedded Server-based
Concurrency Single-writer Multi-writer
Scaling Single instance Horizontal scaling
Best for Development, single-instance Production, distributed

Migrations: PostgreSQL migrations are in migrations/postgres/. Run with:

# Using sqlx-cli
sqlx migrate run --source migrations/postgres

In-Memory Mode

For testing and ephemeral workflows:

let runner = AppRunner::builder()
    .app(app)
    .checkpointer(CheckpointerType::InMemory)
    .build()
    .await?;

Iterative Checkpointed Workflows

Use iterative sessions when one logical run should process many inputs while keeping one checkpoint lineage. This is useful for event-driven systems that repeatedly restore the latest durable state, apply the next input, run the graph, and checkpoint the result.

use weavegraph::node::NodePartial;
use weavegraph::runtimes::{AppRunner, CheckpointerType};
use weavegraph::state::VersionedState;
use weavegraph::types::NodeKind;
use weavegraph::utils::collections::new_extra_map;

# async fn example(app: weavegraph::app::App) -> Result<(), Box<dyn std::error::Error>> {
let mut runner = AppRunner::builder()
    .app(app)
    .checkpointer(CheckpointerType::SQLite)
    .autosave(true)
    .build()
    .await;

let run_id = "market-run-2026-05-08".to_string();
runner
    .create_iterative_session(
        run_id.clone(),
        VersionedState::new_with_user_message("start"),
        NodeKind::Start,
    )
    .await?;

for tick in [1, 2, 3] {
    let mut extra = new_extra_map();
    extra.insert("tick".to_string(), serde_json::json!(tick));

    runner
        .invoke_next(&run_id, NodePartial::new().with_extra(extra), NodeKind::Start)
        .await?;
}
# Ok(())
# }

NodeKind::Start means the same initial frontier as a normal session: the graph's outgoing edges from the virtual Start node. A registered custom node can be used to resume from a narrower entry point. NodeKind::End is rejected because it is terminal.

The runner keeps SessionState.step monotonic across invocations. It also clears scheduler version-gating state for each invoke_next call, so the entry path runs for each logical input even when two consecutive input patches are identical.

If you subscribe with AppRunner::event_stream() before an iterative run, each invoke_next(...) emits INVOCATION_END_SCOPE and leaves the stream open for the next input. After the final input, call finish_iterative_session(...) to emit STREAM_END_SCOPE and close the stream for consumers that expect the standard terminal sentinel.

Typed State Slots

Use StateKey<T> when checkpointed extra state needs a documented schema and compile-time payload type while staying JSON-compatible across backends.

use serde::{Deserialize, Serialize};
use weavegraph::node::NodePartial;
use weavegraph::state::{StateKey, StateSnapshot};

#[derive(Serialize, Deserialize)]
struct PortfolioState {
    cash_cents: i64,
}

const PORTFOLIO: StateKey<PortfolioState> = StateKey::new("wq", "portfolio", 1);

fn load(snapshot: &StateSnapshot) -> Result<PortfolioState, weavegraph::state::StateSlotError> {
    snapshot.require_typed(PORTFOLIO)
}

fn store(value: PortfolioState) -> Result<NodePartial, weavegraph::state::StateSlotError> {
    NodePartial::new().with_typed_extra(PORTFOLIO, value)
}

The generated storage key is namespace:name:v{schema_version}, so old and new schemas can coexist during migrations.

Deterministic Clock And Run Metadata

Inject a clock when simulations, replay, or tests need logical time to be independent of wall-clock time. The same clock is available from NodeContext::now_unix_ms() and is attached to node event metadata when present.

use std::sync::Arc;
use weavegraph::runtimes::{AppRunner, CheckpointerType};
use weavegraph::utils::clock::MockClock;

let runner = AppRunner::builder()
    .app(app)
    .checkpointer(CheckpointerType::InMemory)
    .clock(Arc::new(MockClock::new(1_700_000_000)))
    .build()
    .await;

let metadata = runner.run_metadata();
println!("graph={} runtime={} clock={}", metadata.graph_hash, metadata.runtime_config_hash, metadata.clock_mode);

App::graph_metadata() and App::graph_definition_hash() are useful for replay manifests and checkpoint labels. The graph hash covers the graph definition surface, including node kinds, edges, conditional edge registrations, and reducer definition labels. Custom reducers can override Reducer::definition_label(...) when a stable domain label is more appropriate than the default Rust type path.

Replay Conformance Checks

Replay helpers compare normalized events and final state snapshots for uninterrupted/resumed run parity.

use weavegraph::runtimes::{ReplayRun, compare_replay_runs};

let expected = ReplayRun::new(expected_state, expected_events);
let actual = ReplayRun::new(actual_state, actual_events);

compare_replay_runs(&expected, &actual).assert_matches()?;

Use compare_event_sequences_with(...) or compare_replay_runs_with(...) when domain events need custom normalization before comparison.

Storage Management

InMemoryCheckpointer stores only the latest checkpoint per session (automatic retention). No storage management needed.

SQLiteCheckpointer stores complete step history for audit trails and debugging. Storage grows with (sessions × steps_per_session × state_size).

For long-running applications, implement periodic cleanup:

Option 1: Direct SQL maintenance (recommended)

# Delete checkpoints older than 30 days
sqlite3 workflow.db "DELETE FROM steps WHERE created_at < datetime('now', '-30 days')"

# Keep only latest 100 steps per session
sqlite3 workflow.db "
  DELETE FROM steps 
  WHERE step NOT IN (
    SELECT step FROM steps 
    WHERE session_id = steps.session_id 
    ORDER BY step DESC 
    LIMIT 100
  )
"

# Vacuum to reclaim space
sqlite3 workflow.db "VACUUM"

Option 2: Application-level session management

Delete entire sessions when workflows complete:

// Using sqlx directly
sqlx::query("DELETE FROM sessions WHERE id = ?")
    .bind(&session_id)
    .execute(&pool)
    .await?;
// Cascading delete removes all associated steps

Storage monitoring:

# Check database size
ls -lh workflow.db

# Count checkpoints per session
sqlite3 workflow.db "
  SELECT session_id, COUNT(*) as checkpoint_count 
  FROM steps 
  GROUP BY session_id
"

Testing {#testing}

Weavegraph supports comprehensive testing, including property-based tests and event capture.

Running Tests

# All tests with output
cargo test --all -- --nocapture

# Specific test categories
cargo test schedulers:: -- --nocapture
cargo test channels:: -- --nocapture
cargo test integration:: -- --nocapture

Event Capture in Tests

Use MemorySink for synchronous event capture:

use weavegraph::event_bus::{EventBus, MemorySink};
use weavegraph::runtimes::{AppRunner, CheckpointerType};
use weavegraph::state::VersionedState;

# async fn example(app: weavegraph::app::App) -> Result<(), Box<dyn std::error::Error>> {
let sink = MemorySink::new();
let event_bus = EventBus::with_sink(sink.clone());

let mut runner = AppRunner::builder()
    .app(app)
    .checkpointer(CheckpointerType::InMemory)
    .event_bus(event_bus)
    .autosave(false)
    .start_listener(true)
    .build()
    .await;

let session_id = "test-session".to_string();
runner
    .create_session(session_id.clone(), VersionedState::new_with_user_message("Hi"))
    .await?;
runner.run_until_complete(&session_id).await?;

let events = sink.snapshot();
assert!(!events.is_empty());
# Ok(())
# }

Property-Based Testing

Weavegraph uses proptest to ensure correctness across edge cases. See the test suite for examples of property-based validation of schedulers, channels, and state management.

Error Handling {#errors}

Weavegraph provides structured, matchable error enums via thiserror. Rich diagnostic metadata is available behind the optional diagnostics feature.

Basic Usage

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Your workflow code here
    Ok(())
}

Handling Graph Compilation Errors

use weavegraph::graphs::{GraphBuilder, GraphCompileError};
use weavegraph::types::NodeKind;

fn build_app() -> Result<weavegraph::app::App, weavegraph::graphs::GraphCompileError> {
    match GraphBuilder::new()
        .add_edge(NodeKind::Start, NodeKind::Custom("A".into()))
        .add_edge(NodeKind::Custom("A".into()), NodeKind::End)
        .compile()
    {
        Ok(app) => Ok(app),
        Err(GraphCompileError::MissingEntry) => Err(GraphCompileError::MissingEntry),
        Err(GraphCompileError::UnknownNode(nk)) => Err(GraphCompileError::UnknownNode(nk)),
        Err(e) => Err(e),
    }
}

Features:

  • Match on error variants for custom handling
  • Lightweight core error model for library consumers
  • Optional diagnostic metadata (--features diagnostics) for richer terminal reporting

See examples/errors_pretty.rs for comprehensive error handling patterns.

Production Considerations {#production}

Performance

  • Bounded concurrency prevents resource exhaustion
  • Snapshot isolation eliminates state races
  • Channel-based architecture enables efficient partial updates
  • SQLite checkpointing supports failure recovery

Monitoring

  • Structured event streaming for observability platforms
  • Rich tracing spans for distributed tracing
  • Error aggregation and pretty diagnostics
  • Custom event sinks for metrics collection

Deployment

  • Docker-ready with provided docker-compose.yml
  • Environment-based configuration for flexible deployment
  • Graceful shutdown handling for clean termination
  • Migration support for schema evolution

Production Patterns

For web servers with per-request isolation:

use weavegraph::event_bus::{EventBus, ChannelSink};
use weavegraph::runtimes::{AppRunner, CheckpointerType};

// Per-request EventBus with isolated channel
let (tx, rx) = flume::unbounded();
let bus = EventBus::with_sinks(vec![Box::new(ChannelSink::new(tx))]);

let mut runner = AppRunner::builder()
    .app(app.clone())
    .checkpointer(CheckpointerType::InMemory)
    .event_bus(bus)
    .autosave(true)
    .start_listener(true)
    .build()
    .await;

See also: Quickstart, Architecture