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        .with_service_name(ctx.config.service.name.clone());
87    let runtime_worker = RuntimeWorker::new(ctx.db.clone(), registry, "worker-local")
88        .with_service_name(ctx.config.service.name.clone());
89    loop {
90        let cfg: WorkerRuntimeConfig = ctx
91            .runtime_config
92            .snapshot()
93            .get("worker")
94            .unwrap_or_default();
95        // batch_size is descriptor-capped at 1000, so the u64->i64 cast is lossless.
96        let batch_size = cfg.batch_size as i64;
97        tokio::select! {
98            changed = shutdown_rx.changed() => {
99                if changed.is_ok() && *shutdown_rx.borrow() {
100                    break;
101                }
102            }
103            () = Shutdown::wait_for_signal() => {
104                shutdown.signal();
105            }
106            () = tokio::time::sleep(Duration::from_millis(cfg.poll_interval_ms)) => {
107                match scheduler.enqueue_due(&schedules).await {
108                    Ok(run_ids) => {
109                        if !run_ids.is_empty() {
110                            tracing::debug!(
111                                scheduled_function_runs = run_ids.len(),
112                                "runtime scheduler tick"
113                            );
114                        }
115                    }
116                    Err(error) => {
117                        tracing::warn!(error = ?error, "runtime scheduler tick failed");
118                    }
119                }
120                match relay.relay_once(&dispatcher, batch_size).await {
121                    Ok(count) => {
122                        tracing::debug!(claimed_outbox_events = count, "outbox relay tick");
123                    }
124                    Err(error) => {
125                        tracing::warn!(error = ?error, "outbox relay tick failed");
126                    }
127                }
128                match runtime_worker.claim_and_run_batch(batch_size).await {
129                    Ok(count) => {
130                        tracing::debug!(claimed_function_runs = count, "runtime worker tick");
131                    }
132                    Err(error) => {
133                        tracing::warn!(error = ?error, "runtime worker tick failed");
134                    }
135                }
136            }
137        }
138    }
139}
lenso_worker/lib.rs

lenso_worker/
lib.rs
