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

lenso_worker/
lib.rs
