haste_health/commands/
worker.rs

1use std::sync::Arc;
2
3use clap::Subcommand;
4use figment::{
5    Figment,
6    providers::{Env, Format as _, Toml},
7};
8use haste_fhir_model::r4::generated::terminology::IssueType;
9use haste_fhir_operation_error::OperationOutcomeError;
10use haste_server::{config::ServerConfig, jobs::operation::OperationJobHandler, services};
11use haste_worker::{
12    async_jobs::{AsyncJobWorker, JobHandler},
13    search_indexing,
14    traits::Worker as _,
15};
16use tokio::task::{JoinError, JoinHandle};
17
18/// Run background workers. With no subcommand, runs all of them in one process.
19#[derive(Subcommand, Debug)]
20pub(crate) enum WorkerCommands {
21    /// Run only the search-indexing worker.
22    Indexing,
23    /// Run only the async-job worker.
24    AsyncJobs,
25}
26
27fn config_error(e: figment::Error) -> OperationOutcomeError {
28    OperationOutcomeError::error(IssueType::exception(), e.to_string())
29}
30
31fn figment() -> Figment {
32    Figment::new()
33        .merge(Toml::file("haste.toml"))
34        .merge(Env::prefixed("HASTE_"))
35}
36
37async fn start_indexing() -> Result<JoinHandle<()>, OperationOutcomeError> {
38    let config: Arc<search_indexing::WorkerEnvironment> =
39        Arc::new(figment().extract().map_err(config_error)?);
40
41    search_indexing::IndexingWorker::new(config)
42        .await?
43        .run()
44        .await
45}
46
47async fn start_async_jobs() -> Result<JoinHandle<()>, OperationOutcomeError> {
48    // Jobs run through the server's own FHIR client, so the job worker is
49    // built from the server's services (without the index migrations the
50    // server itself owns).
51    let config: Arc<ServerConfig> = Arc::new(figment().extract().map_err(config_error)?);
52    let state = services::create_worker_services(config.clone()).await?;
53    let handlers: Vec<Arc<dyn JobHandler>> =
54        vec![Arc::new(OperationJobHandler::new(state.clone()))];
55
56    AsyncJobWorker::new(state.repo.clone(), handlers, config.jobs.worker_config())
57        .run()
58        .await
59}
60
61/// A worker's loop only ends on a panic, which fails the command.
62fn finished(name: &str, result: Result<(), JoinError>) -> Result<(), OperationOutcomeError> {
63    result.map_err(|e| {
64        OperationOutcomeError::fatal(
65            IssueType::exception(),
66            format!("{name} worker task failed: {e:?}"),
67        )
68    })
69}
70
71/// Runs the `worker` command group. `command` is `None` when the bare `worker`
72/// invocation is used, which runs every worker.
73pub(crate) async fn run(command: &Option<WorkerCommands>) -> Result<(), OperationOutcomeError> {
74    match command {
75        None => {
76            let indexing = start_indexing().await?;
77            let jobs = start_async_jobs().await?;
78
79            // Either loop ending ends the process, so a supervisor restarts
80            // both rather than one running on alone.
81            tokio::select! {
82                result = indexing => finished("Indexing", result),
83                result = jobs => finished("Async job", result),
84            }
85        }
86        Some(WorkerCommands::Indexing) => finished("Indexing", start_indexing().await?.await),
87        Some(WorkerCommands::AsyncJobs) => finished("Async job", start_async_jobs().await?.await),
88    }
89}
haste_health/commands/worker.rs

haste_health/commands/
worker.rs
