lenso_worker/
lib.rs

1use anyhow::Context as _;
2use platform_core::{
3    AppConfig, AppContext, EventHandlerRegistry, LoggingEventPublisher, OutboxRelay,
4    PostgresRuntimeConfigProvider, RuntimeConfigRegistry, Shutdown, WorkerRuntimeConfig,
5    connect_pool, telemetry,
6};
7use platform_runtime::{
8    FunctionRegistry, RuntimeScheduler, RuntimeWorker, ScheduledFunctionDefinition,
9};
10use std::path::Path;
11use std::sync::Arc;
12use std::time::Duration;
13use tracing::info;
14
15pub async fn run_from_env() -> anyhow::Result<()> {
16    run_from_env_with_composition(lenso_bootstrap::HostComposition::default()).await
17}
18
19pub async fn run_from_env_with_composition(
20    composition: lenso_bootstrap::HostComposition,
21) -> anyhow::Result<()> {
22    let config = AppConfig::try_from_env().context("invalid application configuration")?;
23    telemetry::init(&config.telemetry)?;
24    let provider_plan = lenso_bootstrap::provider_runtime_plan_from_workspace(Path::new("."))
25        .context("failed to compile Provider Runtime Plan")?;
26
27    let db = connect_pool(&config.database).await?;
28    let ctx = AppContext::new(config, db, Arc::new(LoggingEventPublisher));
29
30    let descriptors =
31        lenso_bootstrap::runtime_config_descriptors_with_composition(&ctx, &composition)
32            .context("failed to collect runtime-config descriptors")?;
33    let groups =
34        lenso_bootstrap::runtime_config_group_descriptors_with_composition(&ctx, &composition)
35            .context("failed to collect runtime-config groups")?;
36    let runtime_config_registry = RuntimeConfigRegistry::try_new_with_groups(descriptors, groups)
37        .context("duplicate runtime-config descriptor registered")?;
38    let runtime_config = PostgresRuntimeConfigProvider::connect(
39        ctx.db.clone(),
40        Arc::new(runtime_config_registry),
41        "worker",
42    )
43    .await
44    .context("failed to load runtime-config snapshot")?;
45    runtime_config.spawn_listener();
46    let ctx = ctx.with_runtime_config_provider(runtime_config);
47
48    let modules = lenso_bootstrap::load_modules_with_composition_and_provider_plan(
49        &ctx,
50        &composition,
51        provider_plan.as_ref(),
52    )
53    .await
54    .context("failed to load modules")?;
55    let registry = Arc::new(
56        lenso_bootstrap::try_function_registry(&modules)
57            .context("invalid module runtime binding")?,
58    );
59    let activation_run_ids =
60        lenso_bootstrap::enqueue_lifecycle_activation_jobs(&ctx, &modules, &registry)
61            .await
62            .context("failed to enqueue module lifecycle activation jobs")?;
63    let schedules = lenso_bootstrap::scheduled_functions(&modules, registry.as_ref())
64        .context("failed to collect scheduled runtime functions")?;
65    let event_handlers =
66        lenso_bootstrap::try_event_handlers_with_runtime_actions(&ctx, &modules, registry.clone())
67            .context("invalid module Event handler binding")?;
68
69    info!(
70        functions = registry.all().count(),
71        lifecycle_activation_jobs = activation_run_ids.len(),
72        scheduled_functions = schedules.len(),
73        "starting worker"
74    );
75
76    run_worker_loop(ctx.clone(), event_handlers, registry, schedules).await;
77    Ok(())
78}
79
80async fn run_worker_loop(
81    ctx: AppContext,
82    dispatcher: EventHandlerRegistry,
83    registry: Arc<FunctionRegistry>,
84    schedules: Vec<ScheduledFunctionDefinition>,
85) {
86    let shutdown = ctx.shutdown.clone();
87    let mut shutdown_rx = shutdown.subscribe();
88    let relay = OutboxRelay::new(ctx.db.clone(), "worker-local");
89    let scheduler = RuntimeScheduler::new(ctx.db.clone(), "worker-local")
90        .with_service_name(ctx.config.service.name.clone());
91    let runtime_worker = RuntimeWorker::new(ctx.db.clone(), registry, "worker-local")
92        .with_service_name(ctx.config.service.name.clone());
93    loop {
94        let cfg: WorkerRuntimeConfig = ctx
95            .runtime_config
96            .snapshot()
97            .get("worker")
98            .unwrap_or_default();
99        // batch_size is descriptor-capped at 1000, so the u64->i64 cast is lossless.
100        let batch_size = cfg.batch_size as i64;
101        tokio::select! {
102            changed = shutdown_rx.changed() => {
103                if changed.is_ok() && *shutdown_rx.borrow() {
104                    break;
105                }
106            }
107            () = Shutdown::wait_for_signal() => {
108                shutdown.signal();
109            }
110            () = tokio::time::sleep(Duration::from_millis(cfg.poll_interval_ms)) => {
111                match scheduler.enqueue_due(&schedules).await {
112                    Ok(run_ids) => {
113                        if !run_ids.is_empty() {
114                            tracing::debug!(
115                                scheduled_function_runs = run_ids.len(),
116                                "runtime scheduler tick"
117                            );
118                        }
119                    }
120                    Err(error) => {
121                        tracing::warn!(error = ?error, "runtime scheduler tick failed");
122                    }
123                }
124                match relay.relay_once(&dispatcher, batch_size).await {
125                    Ok(count) => {
126                        tracing::debug!(claimed_outbox_events = count, "outbox relay tick");
127                    }
128                    Err(error) => {
129                        tracing::warn!(error = ?error, "outbox relay tick failed");
130                    }
131                }
132                match runtime_worker.claim_and_run_batch(batch_size).await {
133                    Ok(count) => {
134                        tracing::debug!(claimed_function_runs = count, "runtime worker tick");
135                    }
136                    Err(error) => {
137                        tracing::warn!(error = ?error, "runtime worker tick failed");
138                    }
139                }
140            }
141        }
142    }
143}
lenso_worker/lib.rs

lenso_worker/
lib.rs
