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

lenso_worker/
lib.rs
