Function instance_monitor

crates/cargo-lambda-watch/src/scheduler.rs:125–178

Background monitor that checks queue depths and spawns instances as needed

(
    subsys: SubsystemHandle,
    state: RuntimeState,
    cargo_options: CargoOptions,
    watcher_config: WatcherConfig,
    gc_tx: Sender<(String, InstanceId)>,
)

/// Background monitor that checks queue depths and spawns instances as needed
async fn instance_monitor(
    subsys: SubsystemHandle,
    state: RuntimeState,
    cargo_options: CargoOptions,
    watcher_config: WatcherConfig,
    gc_tx: Sender<(String, InstanceId)>,
) -> Result<(), ServerError> {
    let mut interval = tokio::time::interval(Duration::from_millis(100));

    loop {
        tokio::select! {
            _ = interval.tick() => {
                let function_names = state.req_cache.keys().await;

                for function_name in function_names {
                    let queue_depth = state.req_cache.queue_depth(&function_name).await;

                    if queue_depth == 0 {
                        continue;
                    }

                    let mut pools = state.instance_pools.write().await;
                    let pool = pools
                        .entry(function_name.clone())
                        .or_insert_with(|| InstancePool::new(state.max_concurrency));
                    let pool_clone = pool.clone();
                    drop(pools);

                    if pool_clone.should_spawn_instance(queue_depth).await {
                        debug!(
                            function = ?function_name,
                            queue_depth,
                            "spawning additional instance"
                        );

                        spawn_function_instance(
                            &subsys,
                            &state,
                            &function_name,
                            &cargo_options,
                            &watcher_config,
                            &gc_tx,
                        )
                        .await?;
                    }
                }
            }
            _ = subsys.on_shutdown_requested() => {
                info!("terminating instance monitor");
                return Ok(());
            }
        }
    }
}

/// Spawn a new function instance
async fn spawn_function_instance(