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(lenso_bootstrap::function_registry(&modules));
56    let activation_run_ids =
57        lenso_bootstrap::enqueue_lifecycle_activation_jobs(&ctx, &modules, &registry)
58            .await
59            .context("failed to enqueue module lifecycle activation jobs")?;
60    let schedules = lenso_bootstrap::scheduled_functions(&modules, registry.as_ref())
61        .context("failed to collect scheduled runtime functions")?;
62    let event_handlers =
63        lenso_bootstrap::event_handlers_with_runtime_actions(&ctx, &modules, registry.clone());
64
65    info!(
66        functions = registry.all().count(),
67        lifecycle_activation_jobs = activation_run_ids.len(),
68        scheduled_functions = schedules.len(),
69        "starting worker"
70    );
71
72    run_worker_loop(ctx.clone(), event_handlers, registry, schedules).await;
73    Ok(())
74}
75
76async fn run_worker_loop(
77    ctx: AppContext,
78    dispatcher: EventHandlerRegistry,
79    registry: Arc<FunctionRegistry>,
80    schedules: Vec<ScheduledFunctionDefinition>,
81) {
82    let shutdown = ctx.shutdown.clone();
83    let mut shutdown_rx = shutdown.subscribe();
84    let relay = OutboxRelay::new(ctx.db.clone(), "worker-local");
85    let scheduler = RuntimeScheduler::new(ctx.db.clone(), "worker-local");
86    let runtime_worker = RuntimeWorker::new(ctx.db.clone(), registry, "worker-local");
87    loop {
88        let cfg: WorkerRuntimeConfig = ctx
89            .runtime_config
90            .snapshot()
91            .get("worker")
92            .unwrap_or_default();
93        // batch_size is descriptor-capped at 1000, so the u64->i64 cast is lossless.
94        let batch_size = cfg.batch_size as i64;
95        tokio::select! {
96            changed = shutdown_rx.changed() => {
97                if changed.is_ok() && *shutdown_rx.borrow() {
98                    break;
99                }
100            }
101            () = Shutdown::wait_for_signal() => {
102                shutdown.signal();
103            }
104            () = tokio::time::sleep(Duration::from_millis(cfg.poll_interval_ms)) => {
105                match scheduler.enqueue_due(&schedules).await {
106                    Ok(run_ids) => {
107                        if !run_ids.is_empty() {
108                            tracing::debug!(
109                                scheduled_function_runs = run_ids.len(),
110                                "runtime scheduler tick"
111                            );
112                        }
113                    }
114                    Err(error) => {
115                        tracing::warn!(error = ?error, "runtime scheduler tick failed");
116                    }
117                }
118                match relay.relay_once(&dispatcher, batch_size).await {
119                    Ok(count) => {
120                        tracing::debug!(claimed_outbox_events = count, "outbox relay tick");
121                    }
122                    Err(error) => {
123                        tracing::warn!(error = ?error, "outbox relay tick failed");
124                    }
125                }
126                match runtime_worker.claim_and_run_batch(batch_size).await {
127                    Ok(count) => {
128                        tracing::debug!(claimed_function_runs = count, "runtime worker tick");
129                    }
130                    Err(error) => {
131                        tracing::warn!(error = ?error, "runtime worker tick failed");
132                    }
133                }
134            }
135        }
136    }
137}
lenso_worker/lib.rs

lenso_worker/
lib.rs
