Brummel/doctate
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(server): plumb Arc<Settings> through workers and routes

Browse Source 
Config sheds its 14 Bucket-B fields (retention, whisper, ollama, llm) and
gains a sibling Arc<Settings> in AppState, loaded from settings.toml at
startup via Settings::load_or_default. Workers (transcribe, analyze,
recovery, auto_trigger) and routes (health, bulk, case_actions,
user_web) take settings as a separate parameter or State<> extractor;
Config::llm_configured moves to Settings::llm_configured.

TestConfig::build_pair() returns (Arc<Config>, Arc<Settings>); the new
create_router_with_settings / create_router_and_session_store_with_settings
helpers wire both into integration tests that exercise LLM/Whisper/Ollama.
The legacy single-Arc create_router falls back to Settings::default()
so the 17 non-LLM tests stay untouched.

Verified: cargo build clean, clippy --all-targets clean, 318 tests pass.
This commit is contained in:
Brummel
2026-04-27 11:48:41 +02:00
parent 6bbbf10f77
commit a510c20e75
21 changed files with 310 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/analyze/auto_trigger.rs
+11 -11
View File
@@ -19,10 +19,10 @@ use tokio::io::AsyncWriteExt;
use tracing::warn;
use super::{ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE};
use crate::config::Config;
use crate::events::{self, CaseEventKind, EventSender};
use crate::paths;
use crate::routes::case_actions::build_analysis_input;
use crate::settings::Settings;
/// Filename of the per-case failure marker. Written by the worker on
/// error, deleted on success. Presence with a matching
@@ -99,10 +99,10 @@ pub async fn evaluate_case(case_dir: &Path) -> AutoDecision {
pub async fn try_enqueue(
case_dir: &Path,
tx: &AnalyzeSender,
config: &Arc<Config>,
settings: &Arc<Settings>,
events_tx: &EventSender,
) -> bool {
if !config.llm_configured() {
if !settings.llm_configured() {
return false;
}
@@ -159,7 +159,7 @@ pub async fn try_enqueue(
pub async fn try_enqueue_all_for_user(
user_root: &Path,
tx: &AnalyzeSender,
config: &Arc<Config>,
settings: &Arc<Settings>,
events_tx: &EventSender,
) {
let Ok(mut entries) = tokio::fs::read_dir(user_root).await else {
@@ -179,7 +179,7 @@ pub async fn try_enqueue_all_for_user(
if crate::paths::is_closed(&case_path).await {
continue;
}
try_enqueue(&case_path, tx, config, events_tx).await;
try_enqueue(&case_path, tx, settings, events_tx).await;
}
}
@@ -466,8 +466,8 @@ mod tests {
#[tokio::test]
async fn try_enqueue_all_sends_only_eligible_cases() {
use crate::analyze::channel as analyze_channel;
use crate::config::Config;
use crate::events::channel as events_channel;
use crate::settings::Settings;
let user_root = TempDir::new().unwrap();
@@ -515,12 +515,12 @@ mod tests {
let (tx, mut rx) = analyze_channel();
let events_tx = events_channel();
let mut cfg = Config::test_default();
cfg.llm_url = "http://localhost:9999".into();
cfg.llm_model = "test".into();
let cfg = Arc::new(cfg);
let mut settings = Settings::default();
settings.llm.url = "http://localhost:9999".into();
settings.llm.model = "test".into();
let settings = Arc::new(settings);
try_enqueue_all_for_user(user_root.path(), &tx, &cfg, &events_tx).await;
try_enqueue_all_for_user(user_root.path(), &tx, &settings, &events_tx).await;
// Drop the sender half so recv closes after draining.
drop(tx);
server/src/analyze/worker.rs
+20 -12
View File
@@ -8,9 +8,9 @@ use tracing::{error, info, warn};
use super::{
ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeReceiver, DOCUMENT_FILE, auto_trigger, llm, prompt,
};
use crate::config::Config;
use crate::events::{self, CaseEventKind, EventSender};
use crate::gazetteer::Gazetteer;
use crate::settings::Settings;
use crate::{BusyGuard, WorkerBusy};
/// Consume analyze jobs sequentially. The external LLM tolerates parallelism,
@@ -18,18 +18,26 @@ use crate::{BusyGuard, WorkerBusy};
/// provider from bursts. Trivial to lift later via `tokio::spawn(process(..))`.
pub async fn run(
mut rx: AnalyzeReceiver,
config: Arc<Config>,
settings: Arc<Settings>,
client: reqwest::Client,
worker_busy: WorkerBusy,
vocab: Arc<Gazetteer>,
events_tx: EventSender,
) {
info!(vocab_entries = vocab.len(), "Analyze worker started");
let timeout = Duration::from_secs(config.llm_timeout_seconds);
let timeout = Duration::from_secs(settings.llm.timeout_seconds);
while let Some(job) = rx.recv().await {
let _guard = BusyGuard::new(worker_busy.clone());
process(&job.case_dir, &config, &client, &vocab, timeout, &events_tx).await;
process(
&job.case_dir,
&settings,
&client,
&vocab,
timeout,
&events_tx,
)
.await;
}
warn!("Analyze worker stopped (channel closed)");
@@ -37,7 +45,7 @@ pub async fn run(
async fn process(
case_dir: &Path,
config: &Config,
settings: &Settings,
client: &reqwest::Client,
vocab: &Gazetteer,
timeout: Duration,
@@ -106,16 +114,16 @@ async fn process(
// document write below, the recovery scan will re-enqueue this job and
// the call will be made again. Accepted cost — rare event, small
// per-call price. Response-caching in a `.tmp` file would prevent it.
let settings = llm::LlmSettings {
base_url: &config.llm_url,
api_key: &config.llm_api_key,
model: &config.llm_model,
temperature: config.llm_temperature,
let llm_settings = llm::LlmSettings {
base_url: &settings.llm.url,
api_key: &settings.llm.api_key,
model: &settings.llm.model,
temperature: settings.llm.temperature,
};
let document = match llm::chat_once(
client,
&settings,
&config.llm_system_prompt,
&llm_settings,
&settings.llm.system_prompt,
&user_content,
timeout,
)
server/src/config.rs
+3 -63
View File
@@ -75,6 +75,9 @@ struct UsersFile {
user: Vec<User>,
}
/// Bootstrap config loaded once from `.env` at startup. Runtime-tunable
/// values (retention, whisper/ollama/llm endpoints, prompts) live in
/// `Settings` (loaded from `settings.toml`).
pub struct Config {
// Phase 1 — required
pub server_port: u16,
@@ -86,24 +89,6 @@ pub struct Config {
pub api_keys: HashMap<String, String>, // api_key_value → slug
// Phase 2+ — optional with defaults
pub retention_audio_days: u32,
pub retention_transcript_days: u32,
pub retention_document_days: u32,
pub whisper_url: String,
pub whisper_timeout_seconds: u64,
pub ollama_url: String,
pub ollama_model: String,
pub ollama_keep_alive: u32,
pub llm_url: String,
pub llm_api_key: String,
pub llm_model: String,
pub llm_temperature: f32,
pub llm_timeout_seconds: u64,
/// System prompt sent to the analysis LLM. Defaults to
/// `crate::analyze::prompt::SYSTEM_PROMPT` if `LLM_SYSTEM_PROMPT` is unset
/// or empty. Runtime-configurable so admins can iterate and use the
/// "Neu analysieren" button to re-run cases.
pub llm_system_prompt: String,
/// Directory containing gazetteer `*.txt` files (anatomy, substances,
/// medications). Loaded once at startup; a missing directory is a
/// non-fatal WARN — analysis runs without proper-name correction.
@@ -135,27 +120,6 @@ impl Config {
users,
api_keys,
retention_audio_days: optional_env_parsed("RETENTION_AUDIO_DAYS", 30),
retention_transcript_days: optional_env_parsed("RETENTION_TRANSCRIPT_DAYS", 30),
retention_document_days: optional_env_parsed("RETENTION_DOCUMENT_DAYS", 0),
whisper_url: optional_env("WHISPER_URL", "http://localhost:10300"),
whisper_timeout_seconds: optional_env_parsed("WHISPER_TIMEOUT_SECONDS", 120),
ollama_url: optional_env("OLLAMA_URL", "http://localhost:11434"),
ollama_model: optional_env("OLLAMA_MODEL", "gemma3:4b"),
ollama_keep_alive: optional_env_parsed("OLLAMA_KEEP_ALIVE", 0),
llm_url: optional_env("LLM_URL", ""),
llm_api_key: optional_env("LLM_API_KEY", ""),
llm_model: optional_env("LLM_MODEL", ""),
llm_temperature: optional_env_parsed("LLM_TEMPERATURE", 0.0),
llm_timeout_seconds: optional_env_parsed("LLM_TIMEOUT_SECONDS", 180),
llm_system_prompt: {
let v = optional_env("LLM_SYSTEM_PROMPT", "");
if v.is_empty() {
crate::analyze::prompt::SYSTEM_PROMPT.to_string()
} else {
v
}
},
vocab_dir: PathBuf::from(optional_env("VOCAB_DIR", "./vocab")),
hunspell_dict_path: PathBuf::from(optional_env(
"HUNSPELL_DICT",
@@ -166,16 +130,6 @@ impl Config {
}
}
/// True iff the external analysis LLM is configured. `llm_api_key` is
/// intentionally **not** required — Ollama's OpenAI-compatible endpoint
/// at `/v1/chat/completions` accepts unauthenticated requests and is a
/// supported deployment target. For hosted providers (Ionos, OpenAI) the
/// key remains necessary in practice; the provider will reject with 401
/// if it is missing, which is correct feedback.
pub fn llm_configured(&self) -> bool {
!self.llm_url.is_empty() && !self.llm_model.is_empty()
}
/// Sane defaults for integration tests. Not a `Default` impl on purpose:
/// production code must go through `from_env()`, and `Default::default()`
/// carries an implicit "safe fallback" connotation this value does not
@@ -199,20 +153,6 @@ impl Config {
log_max_days: 90,
users: Vec::new(),
api_keys: HashMap::new(),
retention_audio_days: 30,
retention_transcript_days: 30,
retention_document_days: 0,
whisper_url: "http://localhost:10300".into(),
whisper_timeout_seconds: 120,
ollama_url: "http://localhost:11434".into(),
ollama_model: "gemma3:4b".into(),
ollama_keep_alive: 0,
llm_url: String::new(),
llm_api_key: String::new(),
llm_model: String::new(),
llm_temperature: 0.0,
llm_timeout_seconds: 180,
llm_system_prompt: crate::analyze::prompt::SYSTEM_PROMPT.to_string(),
vocab_dir: PathBuf::new(),
hunspell_dict_path: PathBuf::new(),
session_timeout_hours: 8,
server/src/lib.rs
+25
View File
@@ -12,6 +12,7 @@ pub mod oneliner_locks;
pub mod paths;
pub mod retention;
pub mod routes;
pub mod settings;
pub mod transcribe;
pub mod web_session;
@@ -28,6 +29,7 @@ use config::Config;
use gazetteer::Gazetteer;
use magic_link::MagicLinkStore;
use oneliner_locks::OnelinerLocks;
use settings::Settings;
use transcribe::TranscribeSender;
use web_session::SessionStore;
@@ -88,6 +90,7 @@ pub struct PipelineState {
#[derive(Clone)]
pub struct AppState {
pub config: Arc<Config>,
pub settings: Arc<Settings>,
pub transcribe_tx: TranscribeSender,
pub analyze_tx: AnalyzeSender,
pub session_store: SessionStore,
@@ -107,6 +110,12 @@ impl FromRef<AppState> for Arc<Config> {
}
}
impl FromRef<AppState> for Arc<Settings> {
fn from_ref(state: &AppState) -> Self {
state.settings.clone()
}
}
impl FromRef<AppState> for TranscribeSender {
fn from_ref(state: &AppState) -> Self {
state.transcribe_tx.clone()
@@ -199,11 +208,27 @@ pub fn create_router(config: Arc<Config>) -> Router {
/// shared with the router (`Arc`), so any session created via `/web/login`
/// or `/web/magic` becomes observable through it.
pub fn create_router_and_session_store(config: Arc<Config>) -> (Router, SessionStore) {
create_router_and_session_store_with_settings(config, Arc::new(Settings::default()))
}
/// Variant of [`create_router`] that takes both `Config` and `Settings`.
/// Use from tests that exercise Bucket-B paths (LLM, Whisper, Ollama URLs);
/// otherwise the [`create_router`] convenience with default Settings is enough.
pub fn create_router_with_settings(config: Arc<Config>, settings: Arc<Settings>) -> Router {
create_router_and_session_store_with_settings(config, settings).0
}
/// Variant of [`create_router_and_session_store`] that takes both `Config` and `Settings`.
pub fn create_router_and_session_store_with_settings(
config: Arc<Config>,
settings: Arc<Settings>,
) -> (Router, SessionStore) {
let (transcribe_tx, _transcribe_rx) = transcribe::channel();
let (analyze_tx, _analyze_rx) = analyze::channel();
let session_store = web_session::new_store();
let router = create_router_with_state(AppState {
config,
settings,
transcribe_tx,
analyze_tx,
session_store: session_store.clone(),
server/src/main.rs
+15 -15
View File
@@ -13,6 +13,7 @@ use doctate_server::analyze;
use doctate_server::config::Config;
use doctate_server::events;
use doctate_server::gazetteer::{Gazetteer, SpellbookDict};
use doctate_server::settings::Settings;
use doctate_server::transcribe;
#[tokio::main]
@@ -20,6 +21,9 @@ async fn main() {
dotenvy::dotenv().ok();
let config = Arc::new(Config::from_env());
let settings_path = std::env::var("SETTINGS_FILE").unwrap_or_else(|_| "settings.toml".into());
let settings = Arc::new(Settings::load_or_default(&settings_path));
// Logging: stdout + daily rotating log files.
let env_filter =
EnvFilter::try_new(&config.log_level).unwrap_or_else(|_| EnvFilter::new("info"));
@@ -43,23 +47,17 @@ async fn main() {
.init();
// Surface configuration gaps that silently disable features.
if !config.llm_configured() {
if !settings.llm_configured() {
warn!(
"LLM not configured — 'Fall abschließen' is disabled. \
Set LLM_URL, LLM_API_KEY and LLM_MODEL in .env to enable."
Set [llm].url and [llm].model in settings.toml to enable."
);
}
{
let custom = std::env::var("LLM_SYSTEM_PROMPT")
.map(|v| !v.is_empty())
.unwrap_or(false);
info!(
prompt_source = if custom { "env" } else { "default" },
prompt_chars = config.llm_system_prompt.chars().count(),
"system prompt loaded"
);
}
info!(
prompt_chars = settings.llm.system_prompt.chars().count(),
"system prompt loaded"
);
// Shared HTTP client for both downstream pipelines.
let http_client = reqwest::Client::builder()
@@ -135,6 +133,7 @@ async fn main() {
tokio::spawn(transcribe::worker::run(
transcribe_rx,
config.clone(),
settings.clone(),
http_client.clone(),
transcribe_busy.clone(),
vocab.clone(),
@@ -145,7 +144,7 @@ async fn main() {
let tx = transcribe_tx.clone();
let data_path = config.data_path.clone();
let client = http_client.clone();
let cfg = config.clone();
let settings = settings.clone();
let vocab = vocab.clone();
let events = events_tx.clone();
let heal_busy = oneliner_heal_busy.clone();
@@ -158,7 +157,7 @@ async fn main() {
// page-load does not spawn a second parallel regen loop.
let _guard = doctate_server::BusyGuard::new(heal_busy);
transcribe::recovery::regenerate_missing_oneliners(
&data_path, &client, &cfg, &vocab, &events, &locks,
&data_path, &client, &settings, &vocab, &events, &locks,
)
.await;
});
@@ -170,7 +169,7 @@ async fn main() {
std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
tokio::spawn(analyze::worker::run(
analyze_rx,
config.clone(),
settings.clone(),
http_client.clone(),
analyze_busy.clone(),
vocab.clone(),
@@ -186,6 +185,7 @@ async fn main() {
let state = AppState {
config: config.clone(),
settings: settings.clone(),
transcribe_tx,
analyze_tx,
session_store: doctate_server::web_session::new_store(),
server/src/routes/bulk.rs
+5 -3
View File
@@ -20,6 +20,7 @@ use crate::routes::case_actions::{
build_analysis_input, reset_case_artefacts, write_input_create_new,
};
use crate::routes::user_web::locate_case;
use crate::settings::Settings;
#[derive(Debug, Deserialize)]
pub struct BulkForm {
@@ -44,6 +45,7 @@ impl HasCsrfToken for BulkForm {
pub async fn handle_bulk_action(
user: AuthenticatedWebUser,
State(config): State<Arc<Config>>,
State(settings): State<Arc<Settings>>,
State(analyze_tx): State<AnalyzeSender>,
State(events_tx): State<EventSender>,
State(locks): State<OnelinerLocks>,
@@ -70,7 +72,7 @@ pub async fn handle_bulk_action(
match action {
BulkAction::Analyze => {
bulk_analyze(
&config,
&settings,
&analyze_tx,
&events_tx,
&user.slug,
@@ -89,14 +91,14 @@ pub async fn handle_bulk_action(
}
async fn bulk_analyze(
config: &Config,
settings: &Settings,
analyze_tx: &AnalyzeSender,
events_tx: &EventSender,
slug: &str,
user_root: &std::path::Path,
case_ids: &[String],
) {
if !config.llm_configured() {
if !settings.llm_configured() {
warn!(slug = %slug, "bulk-analyze: LLM not configured, skipping all");
return;
}
server/src/routes/case_actions.rs
+3 -1
View File
@@ -30,6 +30,7 @@ use crate::paths::{
};
use crate::routes::user_web::locate_case_or_404;
use crate::routes::web::validate_filename;
use crate::settings::Settings;
/// POST /web/cases/{case_id}/analyze
///
@@ -41,12 +42,13 @@ use crate::routes::web::validate_filename;
pub async fn handle_analyze_case(
user: AuthenticatedWebUser,
State(config): State<Arc<Config>>,
State(settings): State<Arc<Settings>>,
State(analyze_tx): State<AnalyzeSender>,
State(events_tx): State<EventSender>,
CaseIdPath(case_id): CaseIdPath,
CsrfForm(form): CsrfForm<CsrfOnlyForm>,
) -> Result<Redirect, AppError> {
if !config.llm_configured() {
if !settings.llm_configured() {
return Err(AppError::ServiceUnavailable(
"LLM-Analyse nicht konfiguriert".into(),
));
server/src/routes/health.rs
+7 -3
View File
@@ -5,14 +5,18 @@ use axum::extract::State;
use serde_json::{Value, json};
use crate::config::Config;
use crate::settings::Settings;
pub async fn handle_health(State(config): State<Arc<Config>>) -> Json<Value> {
pub async fn handle_health(
State(config): State<Arc<Config>>,
State(settings): State<Arc<Settings>>,
) -> Json<Value> {
Json(json!({
"status": "ok",
"port": config.server_port,
"data_path": config.data_path.to_string_lossy(),
"users": config.users.iter().map(|u| &u.slug).collect::<Vec<_>>(),
"whisper_url": config.whisper_url,
"ollama_url": config.ollama_url,
"whisper_url": settings.whisper.url,
"ollama_url": settings.ollama.url,
}))
}
server/src/routes/user_web.rs
+17 -10
View File
@@ -25,6 +25,7 @@ use crate::gazetteer::Gazetteer;
use crate::paths;
use crate::routes::case_actions::read_document;
use crate::routes::web::{RecordingView, scan_recordings};
use crate::settings::Settings;
use crate::transcribe::recovery as transcribe_recovery;
use crate::{BusyGuard, PipelineState};
@@ -429,7 +430,7 @@ impl PipelineState {
user_root: &Path,
slug: &str,
http_client: &reqwest::Client,
config: &Arc<Config>,
settings: &Arc<Settings>,
vocab: &Arc<Gazetteer>,
events_tx: &EventSender,
) {
@@ -456,7 +457,7 @@ impl PipelineState {
let user_root = user_root.to_path_buf();
let slug_owned = slug.to_owned();
let http_client = http_client.clone();
let config = Arc::clone(config);
let settings = Arc::clone(settings);
let vocab = Arc::clone(vocab);
let events_tx = events_tx.clone();
let busy = Arc::clone(&self.oneliner_heal_busy.0);
@@ -472,7 +473,7 @@ impl PipelineState {
&user_root,
&slug_owned,
&http_client,
&config,
&settings,
&vocab,
&events_tx,
&locks,
@@ -484,9 +485,11 @@ impl PipelineState {
}
}
#[allow(clippy::too_many_arguments)]
pub async fn handle_my_cases(
user: AuthenticatedWebUser,
State(config): State<Arc<Config>>,
State(settings): State<Arc<Settings>>,
State(pipeline): State<PipelineState>,
State(events_tx): State<EventSender>,
State(http_client): State<reqwest::Client>,
@@ -499,7 +502,7 @@ pub async fn handle_my_cases(
&user_root,
&user.slug,
&http_client,
&config,
&settings,
&vocab,
&events_tx,
)
@@ -508,7 +511,7 @@ pub async fn handle_my_cases(
// render. The common case is "nothing to do" and costs a handful of
// stat-calls per case; actual enqueues happen only when pre-conditions
// flip (new transcripts in, stale document, ...).
auto_trigger::try_enqueue_all_for_user(&user_root, &pipeline.analyze_tx, &config, &events_tx)
auto_trigger::try_enqueue_all_for_user(&user_root, &pipeline.analyze_tx, &settings, &events_tx)
.await;
// Lazy retention sweep: at most one disk scan per /web/cases visit,
@@ -531,6 +534,7 @@ pub async fn handle_my_cases(
busy,
show_closed,
retention.auto_delete_days,
settings.llm_configured(),
)
.await;
let total = cases.len();
@@ -579,6 +583,7 @@ pub async fn handle_my_cases(
pub async fn handle_case_page(
user: AuthenticatedWebUser,
State(config): State<Arc<Config>>,
State(settings): State<Arc<Settings>>,
State(pipeline): State<PipelineState>,
State(events_tx): State<EventSender>,
State(http_client): State<reqwest::Client>,
@@ -592,7 +597,7 @@ pub async fn handle_case_page(
&user_root,
&user.slug,
&http_client,
&config,
&settings,
&vocab,
&events_tx,
)
@@ -619,7 +624,7 @@ pub async fn handle_case_page(
// Auto-analysis trigger for deep-link navigation: same evaluation as
// the list view. Document render below still wins the race if the
// worker happens to finish synchronously.
auto_trigger::try_enqueue(&case_dir, &pipeline.analyze_tx, &config, &events_tx).await;
auto_trigger::try_enqueue(&case_dir, &pipeline.analyze_tx, &settings, &events_tx).await;
let case_id_str = case_id.to_string();
info!(slug = %user.slug, case_id = %case_id, "case page viewed");
@@ -634,7 +639,7 @@ pub async fn handle_case_page(
let (oneliner, oneliner_is_manual) = compute_oneliner_display(&case_dir, &recordings).await;
let a_busy = pipeline.analyze_busy.0.load(Ordering::Acquire);
let flags = compute_flags(&case_dir, &recordings, config.llm_configured(), a_busy).await;
let flags = compute_flags(&case_dir, &recordings, settings.llm_configured(), a_busy).await;
let recordings_count = recordings.len();
let transcribed_count = recordings
@@ -681,9 +686,11 @@ pub async fn handle_case_page(
///
/// Read-only sub-page showing all `.m4a` + transcript pairs for the case.
/// Useful for power-users/admins; not part of the day-to-day workflow.
#[allow(clippy::too_many_arguments)]
pub async fn handle_case_recordings(
user: AuthenticatedWebUser,
State(config): State<Arc<Config>>,
State(settings): State<Arc<Settings>>,
State(pipeline): State<PipelineState>,
State(events_tx): State<EventSender>,
State(http_client): State<reqwest::Client>,
@@ -696,7 +703,7 @@ pub async fn handle_case_recordings(
&user_root,
&user.slug,
&http_client,
&config,
&settings,
&vocab,
&events_tx,
)
@@ -810,9 +817,9 @@ async fn scan_user_cases(
worker_busy: bool,
include_closed: bool,
auto_delete_days: u32,
llm_configured: bool,
) -> Vec<UserCaseView> {
let user_root = config.data_path.join(slug);
let llm_configured = config.llm_configured();
let mut cases = Vec::new();
let now = OffsetDateTime::now_utc();
server/src/transcribe/recovery.rs
+5 -5
View File
@@ -4,11 +4,11 @@ use doctate_common::oneliners::OnelinerState;
use tracing::{info, warn};
use super::{TranscribeJob, TranscribeSender};
use crate::config::Config;
use crate::events::EventSender;
use crate::gazetteer::Gazetteer;
use crate::oneliner_locks::OnelinerLocks;
use crate::paths;
use crate::settings::Settings;
/// Walk `data_path/*/` and enqueue every recording pending transcription for
/// every user. Intended for startup; the same primitive backs the per-user
@@ -204,7 +204,7 @@ async fn has_any_transcript(case_dir: &Path) -> bool {
pub async fn regenerate_missing_oneliners(
data_path: &Path,
client: &reqwest::Client,
config: &Config,
settings: &Settings,
vocab: &Gazetteer,
events_tx: &EventSender,
locks: &OnelinerLocks,
@@ -215,7 +215,7 @@ pub async fn regenerate_missing_oneliners(
}
info!(count = cases.len(), "Regenerating missing oneliners");
for (case_dir, slug) in cases {
super::worker::update_oneliner(&case_dir, &slug, client, config, vocab, events_tx, locks)
super::worker::update_oneliner(&case_dir, &slug, client, settings, vocab, events_tx, locks)
.await;
}
}
@@ -228,7 +228,7 @@ pub async fn regenerate_missing_oneliners_for_user(
user_root: &Path,
slug: &str,
client: &reqwest::Client,
config: &Config,
settings: &Settings,
vocab: &Gazetteer,
events_tx: &EventSender,
locks: &OnelinerLocks,
@@ -238,7 +238,7 @@ pub async fn regenerate_missing_oneliners_for_user(
return;
}
for case_dir in cases {
super::worker::update_oneliner(&case_dir, slug, client, config, vocab, events_tx, locks)
super::worker::update_oneliner(&case_dir, slug, client, settings, vocab, events_tx, locks)
.await;
}
}
server/src/transcribe/worker.rs
+12 -9
View File
@@ -14,15 +14,18 @@ use crate::events::{self, CaseEventKind, EventSender};
use crate::gazetteer::Gazetteer;
use crate::oneliner_locks::OnelinerLocks;
use crate::paths;
use crate::settings::Settings;
use crate::{BusyGuard, WorkerBusy};
const ONELINER_TIMEOUT: Duration = Duration::from_secs(60);
/// Consume transcription jobs sequentially. Whisper holds the GPU, so parallelism
/// here would only cause contention. One job in flight at a time.
#[allow(clippy::too_many_arguments)]
pub async fn run(
mut rx: TranscribeReceiver,
config: Arc<Config>,
settings: Arc<Settings>,
client: reqwest::Client,
worker_busy: WorkerBusy,
vocab: Arc<Gazetteer>,
@@ -30,7 +33,7 @@ pub async fn run(
oneliner_locks: OnelinerLocks,
) {
info!("Transcription worker started");
let timeout = Duration::from_secs(config.whisper_timeout_seconds);
let timeout = Duration::from_secs(settings.whisper.timeout_seconds);
while let Some(job) = rx.recv().await {
let _guard = BusyGuard::new(worker_busy.clone());
@@ -44,7 +47,7 @@ pub async fn run(
// Look up per-user Whisper settings live from the shared config so that
// edits to users.toml (on next restart) reach in-flight jobs naturally.
// Unknown slug → empty settings (service falls back to language=de).
let settings: WhisperUserSettings = config
let user_whisper: WhisperUserSettings = config
.users
.iter()
.find(|u| u.slug == job.user_slug)
@@ -70,10 +73,10 @@ pub async fn run(
let text = match whisper::transcribe(
&client,
&config.whisper_url,
&settings.whisper.url,
remuxed.path(),
timeout,
&settings,
&user_whisper,
)
.await
{
@@ -139,7 +142,7 @@ pub async fn run(
case_dir,
&job.user_slug,
&client,
&config,
&settings,
&vocab,
&events_tx,
&oneliner_locks,
@@ -220,7 +223,7 @@ pub async fn update_oneliner(
case_dir: &Path,
user_slug: &str,
client: &reqwest::Client,
config: &Config,
settings: &Settings,
vocab: &Gazetteer,
events_tx: &EventSender,
locks: &OnelinerLocks,
@@ -302,9 +305,9 @@ pub async fn update_oneliner(
// during the call.
let state = match ollama::generate_oneliner(
client,
&config.ollama_url,
&config.ollama_model,
config.ollama_keep_alive,
&settings.ollama.url,
&settings.ollama.model,
settings.ollama.keep_alive,
&transcript,
ONELINER_TIMEOUT,
)