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

relay_knowledge/application/runtime/
worker.rs
