riverbase_task

riverbase_taskExtensions · v0.2.0

Worker runtime and job tracker: consume bus messages, record jobs.

$cargo add riverbase_task --git https://github.com/riverbase-io/riverbase --features nats-io

• postgresdefault
• nats-io

riverbase_task runs workers: processes that consume integration messages from the bus and record each unit of work as a tracked job. Use it for side effects that must happen after a command commits, such as sending email, calling a provider, or updating a search index.

Example: a NATS consumer

worker/src/main.rs

use std::sync::Arc;

use riverbase_core::datastore::postgres::establish_dbpool;
use riverbase_core::logstore::PostgresDomainLogStore;
use riverbase_core::transport::NatsMessageBus;
use riverbase_task::{
    establish_tracker_dbpool, run_tracker_migrations_url, PostgresTrackerStore, WorkerRuntime,
};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    riverbase_core::applog::init_logging("info", riverbase_core::LogFormat::Compact);
    let db_url = std::env::var("RIVERBASE_DB_URL")?;

    let bus = NatsMessageBus::connect("nats://127.0.0.1:4222").await?;
    let logs = PostgresDomainLogStore::new(establish_dbpool(&db_url).await?).into_bundle();

    run_tracker_migrations_url(&db_url)?;
    let tracker = Arc::new(PostgresTrackerStore::new(establish_tracker_dbpool(&db_url)?));
    let runtime = WorkerRuntime::with_tracker(tracker);

    // Subscribe to every todo integration message; record one tracked job per message.
    runtime
        .run_nats_consumer(&bus, "riverbase.todo.>", "nats-todo", Some(&*logs.activities))
        .await?;
    Ok(())
}

Pieces

Type	Role
WorkerRuntime	Runs consumers against a tracker. run_nats_consumer subscribes to a NATS subject; run_command_side_effect wraps one unit of work.
MessageConsumer	Drains a StreamBus topic (drain_once). Works with any bus.
NatsMessageConsumer	A long-running NATS subscription with a handler closure (feature nats-io)
TrackerStore	Persists Worker, WorkerJob, and JobRelation records
PostgresTrackerStore	The Postgres tracker (feature postgres, default)
JobStatus, WorkerStatus	Lifecycle states

Tracker tables are created by run_tracker_migrations / run_tracker_migrations_url.

Guidelines

• Be idempotent. Outbox delivery is at least once, so the same message can arrive twice. Key side effects by the message id.
• Workers are not workflows. A job runs once. For multi-step processes with state and compensation, use a process manager (see Persistence, audit, and messaging).
• Authorize writes. When a worker sends commands back into a domain, the command's CommandPolicy still applies.

A typical setup is two binaries: the HTTP service publishes integration messages through the outbox, and a worker like the one above consumes them.