relay_knowledge/application/runtime/
worker.rs

1use std::{error::Error, fmt};
2
3use crate::{domain::WorkerKind, env::EnvironmentConfig};
4
5/// External worker runtime configuration and deterministic fallback policy.
6#[derive(Debug, Clone, PartialEq, Eq)]
7pub struct WorkerRuntimeConfig {
8    pub embedding_endpoint: Option<String>,
9    pub ocr_endpoint: Option<String>,
10    pub vision_endpoint: Option<String>,
11    pub extractor_endpoint: Option<String>,
12    pub max_in_flight: usize,
13    pub code_index_max_in_flight: usize,
14    pub silent_updates_enabled: bool,
15}
16
17impl WorkerRuntimeConfig {
18    pub const DEFAULT_MAX_IN_FLIGHT: usize = 2;
19    pub const DEFAULT_CODE_INDEX_MAX_IN_FLIGHT: usize = 2;
20    pub const MAX_CODE_INDEX_MAX_IN_FLIGHT: usize = 8;
21
22    /// Builds worker config from typed environment overrides.
23    pub fn from_environment(
24        environment: &EnvironmentConfig,
25    ) -> Result<Self, WorkerRuntimeConfigError> {
26        Ok(Self {
27            embedding_endpoint: validate_worker_endpoint(
28                environment.workers.embedding_endpoint.clone(),
29            )?,
30            ocr_endpoint: validate_worker_endpoint(environment.workers.ocr_endpoint.clone())?,
31            vision_endpoint: validate_worker_endpoint(environment.workers.vision_endpoint.clone())?,
32            extractor_endpoint: validate_worker_endpoint(
33                environment.workers.extractor_endpoint.clone(),
34            )?,
35            max_in_flight: environment
36                .workers
37                .max_in_flight
38                .unwrap_or(Self::DEFAULT_MAX_IN_FLIGHT),
39            code_index_max_in_flight: environment
40                .workers
41                .code_index_max_in_flight
42                .unwrap_or(Self::DEFAULT_CODE_INDEX_MAX_IN_FLIGHT)
43                .min(Self::MAX_CODE_INDEX_MAX_IN_FLIGHT),
44            silent_updates_enabled: environment.workers.silent_updates_enabled.unwrap_or(false),
45        })
46    }
47
48    /// Returns the configured endpoint for a worker kind.
49    pub fn endpoint_for(&self, kind: WorkerKind) -> Option<&str> {
50        match kind {
51            WorkerKind::Embedding => self.embedding_endpoint.as_deref(),
52            WorkerKind::Ocr => self.ocr_endpoint.as_deref(),
53            WorkerKind::Vision => self.vision_endpoint.as_deref(),
54            WorkerKind::Extractor => self.extractor_endpoint.as_deref(),
55        }
56    }
57}
58
59/// Worker runtime configuration validation error.
60#[derive(Debug, Clone, PartialEq, Eq)]
61pub enum WorkerRuntimeConfigError {
62    InvalidEndpoint(String),
63}
64
65impl fmt::Display for WorkerRuntimeConfigError {
66    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
67        match self {
68            Self::InvalidEndpoint(value) => write!(
69                formatter,
70                "worker endpoint '{value}' must use http:// and include a host"
71            ),
72        }
73    }
74}
75
76impl Error for WorkerRuntimeConfigError {}
77
78fn validate_worker_endpoint(
79    value: Option<String>,
80) -> Result<Option<String>, WorkerRuntimeConfigError> {
81    value
82        .map(|endpoint| {
83            let trimmed = endpoint.trim();
84            if is_valid_worker_http_endpoint(trimmed) {
85                Ok(trimmed.to_owned())
86            } else {
87                Err(WorkerRuntimeConfigError::InvalidEndpoint(endpoint))
88            }
89        })
90        .transpose()
91}
92
93fn is_valid_worker_http_endpoint(value: &str) -> bool {
94    let Some(remainder) = value.strip_prefix("http://") else {
95        return false;
96    };
97    let authority = remainder
98        .split_once('/')
99        .map_or(remainder, |(authority, _)| authority);
100    if authority.is_empty() || authority.contains(char::is_whitespace) {
101        return false;
102    }
103    if let Some((host, port)) = authority.rsplit_once(':') {
104        return !host.is_empty() && port.parse::<u16>().is_ok_and(|port| port > 0);
105    }
106
107    !authority.is_empty()
108}
109
110#[cfg(test)]
111#[path = "worker_tests.rs"]
112mod tests;
relay_knowledge/application/runtime/worker.rs

relay_knowledge/application/runtime/
worker.rs
