blomios/IdeA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(model-server): progression fine du téléchargement du modèle llamacpp (#54)

Browse Source 
Stretch B2/F2 de #54, par-dessus le MVP déjà mergé (B1/F1).

Backend : le port de téléchargement HF publie une progression débouncée
(bytes reçus / total, pourcentage) via le stream de statut du serveur
modèle, avec gestion du total inconnu (pas de faux %), du cache hit,
de l'annulation et du timeout.

Frontend : l'overlay plein-cellule de préparation du serveur affiche la
progression réelle (barre, %, octets, source) en mappant le fil de
statut, avec la règle « pas de faux % » quand le total est inconnu.

Tests : application + infrastructure (téléchargement débouncé, cancel,
timeout, cache hit, total inconnu) et vitest (overlay + formatage pur).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blomios
2026-07-14 10:37:14 +02:00
parent bd335c3a1c
commit 2183dfd291
15 changed files with 1368 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/application/src/error.rs
View File

@ -213,6 +213,7 @@ fn model_server_error_code(err: &ModelServerError) -> &'static str {
ModelServerError::Probe(_) => "probe",
ModelServerError::Process(_) => "process",
ModelServerError::Store(_) => "store",
ModelServerError::Cancelled => "cancelled",
ModelServerError::Timeout => "timeout",
}
}

194
crates/application/src/model_server.rs
View File

@ -3,6 +3,7 @@
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use std::time::Instant as StdInstant;
use domain::events::DomainEvent;
use domain::model_server::{
@ -10,7 +11,8 @@ use domain::model_server::{
ModelSource,
};
use domain::ports::{
EventBus, FileSystem, ManagedProcess, ManagedProcessHandle, ModelServerError, ModelServerProbe,
EventBus, FileSystem, ManagedProcess, ManagedProcessHandle, ModelArtifactCancel,
ModelArtifactDownloader, ModelArtifactProgress, ModelServerError, ModelServerProbe,
ModelServerRegistry, ModelServerRuntime, ProcessStatus, ProfileStore, RemotePath,
};
use domain::{LocalModelServerId, StopPolicy};
@ -215,8 +217,10 @@ pub struct EnsureLocalModelServer {
events: Arc<dyn EventBus>,
active: Mutex<HashMap<LocalModelServerId, ActiveServer>>,
inflight: AsyncMutex<HashMap<LocalModelServerId, Arc<InflightEnsure>>>,
download_cancels: Mutex<HashMap<LocalModelServerId, ModelArtifactCancel>>,
readiness: ReadinessPolicy,
hf_download_deadline: Duration,
model_artifact_downloader: Option<Arc<dyn ModelArtifactDownloader>>,
}
impl EnsureLocalModelServer {
@ -240,8 +244,10 @@ impl EnsureLocalModelServer {
events,
active: Mutex::new(HashMap::new()),
inflight: AsyncMutex::new(HashMap::new()),
download_cancels: Mutex::new(HashMap::new()),
readiness: ReadinessPolicy::default(),
hf_download_deadline: DEFAULT_HF_DOWNLOAD_DEADLINE,
model_artifact_downloader: None,
}
}
@ -259,6 +265,16 @@ impl EnsureLocalModelServer {
self
}
/// Enables pre-resolution/download of Hugging Face artifacts.
#[must_use]
pub fn with_model_artifact_downloader(
mut self,
downloader: Arc<dyn ModelArtifactDownloader>,
) -> Self {
self.model_artifact_downloader = Some(downloader);
self
}
/// Ensures the server is reachable.
///
/// # Errors
@ -332,8 +348,13 @@ impl EnsureLocalModelServer {
self.ensure_model_path_accessible(&config).await?;
self.ensure_no_active_port_collision(&config).await?;
let spawn_config = match self.prepare_spawn_config(config.clone()).await {
Ok(config) => config,
Err(err) => return self.fail(config.id, err),
};
self.publish(config.id, ModelServerLifecycleStatus::Starting);
let spec = match self.runtime.build_spawn_spec(&config) {
let spec = match self.runtime.build_spawn_spec(&spawn_config) {
Ok(spec) => spec,
Err(err) => return self.fail(config.id, err),
};
@ -345,16 +366,71 @@ impl EnsureLocalModelServer {
config.id,
ActiveServer {
handle: handle.clone(),
port: config.endpoint.port,
stop_policy: config.stop_policy,
port: spawn_config.endpoint.port,
stop_policy: spawn_config.stop_policy,
},
);
let hf_source = hf_source(&config);
self.wait_for_started_server(&config, &handle, hf_source)
let hf_source = hf_source(&spawn_config);
self.wait_for_started_server(&spawn_config, &handle, hf_source)
.await
}
async fn prepare_spawn_config(
&self,
config: LocalModelServerConfig,
) -> Result<LocalModelServerConfig, ModelServerError> {
let Some(ModelSource::HuggingFace { repo }) = config.model.source.clone() else {
return Ok(config);
};
let Some(downloader) = self.model_artifact_downloader.as_ref() else {
return Ok(config);
};
let cancel = ModelArtifactCancel::new();
self.download_cancels
.lock()
.unwrap()
.insert(config.id, cancel.clone());
let events = Arc::clone(&self.events);
let source = repo.as_str().to_owned();
let server_id = config.id;
let debouncer = Arc::new(Mutex::new(DownloadProgressDebouncer::new(
server_id,
source.clone(),
events,
)));
let progress = {
let debouncer = Arc::clone(&debouncer);
Arc::new(move |progress: ModelArtifactProgress| {
debouncer.lock().unwrap().on_progress(progress);
}) as Arc<dyn Fn(ModelArtifactProgress) + Send + Sync>
};
let result = tokio::time::timeout(
self.hf_download_deadline,
downloader.resolve_hf_model(&repo, progress, cancel.clone()),
)
.await;
self.download_cancels.lock().unwrap().remove(&config.id);
let resolution = match result {
Ok(Ok(resolution)) => resolution,
Ok(Err(err)) => return Err(err),
Err(_) => {
cancel.cancel();
return Err(ModelServerError::Timeout);
}
};
let mut resolved = config;
resolved.model.source = Some(ModelSource::LocalPath {
path: resolution.path,
});
Ok(resolved)
}
async fn wait_for_started_server(
&self,
config: &LocalModelServerConfig,
@ -468,6 +544,9 @@ impl EnsureLocalModelServer {
/// # Errors
/// Returns the first process error after attempting every eligible stop.
pub async fn stop_on_app_exit(&self) -> Result<(), AppError> {
for cancel in self.download_cancels.lock().unwrap().values() {
cancel.cancel();
}
let entries: Vec<(LocalModelServerId, ActiveServer)> = self
.active
.lock()
@ -626,6 +705,109 @@ pub fn model_server_error_code(err: &ModelServerError) -> &'static str {
ModelServerError::Probe(_) => "probe",
ModelServerError::Process(_) => "process",
ModelServerError::Store(_) => "store",
ModelServerError::Cancelled => "cancelled",
ModelServerError::Timeout => "timeout",
}
}
struct DownloadProgressDebouncer {
server_id: LocalModelServerId,
fallback_source: String,
events: Arc<dyn EventBus>,
last_emit: Option<StdInstant>,
last_percent: Option<f32>,
window_start: StdInstant,
window_count: u32,
}
impl DownloadProgressDebouncer {
fn new(
server_id: LocalModelServerId,
fallback_source: String,
events: Arc<dyn EventBus>,
) -> Self {
let now = StdInstant::now();
Self {
server_id,
fallback_source,
events,
last_emit: None,
last_percent: None,
window_start: now,
window_count: 0,
}
}
fn on_progress(&mut self, progress: ModelArtifactProgress) {
let percent = progress_percent(&progress);
let final_progress = is_final_progress(&progress);
if !self.should_publish(percent, final_progress) {
return;
}
self.mark_published(percent);
self.events.publish(DomainEvent::ModelServerStatusChanged {
server_id: self.server_id,
status: ModelServerLifecycleStatus::Downloading {
downloaded_bytes: progress.downloaded_bytes,
total_bytes: progress.total_bytes,
percent,
source: progress
.source
.or_else(|| Some(self.fallback_source.clone())),
},
});
}
fn should_publish(&mut self, percent: Option<f32>, final_progress: bool) -> bool {
let now = StdInstant::now();
if now.duration_since(self.window_start) >= Duration::from_secs(1) {
self.window_start = now;
self.window_count = 0;
}
if self.last_emit.is_none() || final_progress {
return true;
}
if self.window_count >= 10 {
return false;
}
if self
.last_emit
.is_some_and(|last| now.duration_since(last) < Duration::from_millis(100))
{
return false;
}
if self
.last_emit
.is_some_and(|last| now.duration_since(last) >= Duration::from_millis(500))
{
return true;
}
match (self.last_percent, percent) {
(Some(last), Some(current)) => current - last >= 1.0,
(None, Some(_)) => true,
_ => false,
}
}
fn mark_published(&mut self, percent: Option<f32>) {
self.last_emit = Some(StdInstant::now());
self.last_percent = percent.or(self.last_percent);
self.window_count += 1;
}
}
fn progress_percent(progress: &ModelArtifactProgress) -> Option<f32> {
let downloaded = progress.downloaded_bytes?;
let total = progress.total_bytes?;
if total == 0 {
return None;
}
Some((downloaded as f32 / total as f32) * 100.0)
}
fn is_final_progress(progress: &ModelArtifactProgress) -> bool {
matches!(
(progress.downloaded_bytes, progress.total_bytes),
(Some(downloaded), Some(total)) if total > 0 && downloaded == total
)
}

353
crates/application/tests/model_server.rs
View File

@ -18,6 +18,7 @@ use domain::model_server::{
};
use domain::ports::{
DirEntry, EventBus, EventStream, FileSystem, FsError, ManagedProcess, ManagedProcessHandle,
ModelArtifactCancel, ModelArtifactDownloader, ModelArtifactProgress, ModelArtifactResolution,
ModelServerArgv, ModelServerError, ModelServerProbe, ModelServerRegistry, ModelServerRuntime,
ProcessStatus, ProfileStore, RemotePath, SpawnSpec, StoreError,
};
@ -280,6 +281,74 @@ impl ModelServerRuntime for FakeRuntime {
}
}
#[derive(Clone)]
enum FakeDownloadOutcome {
Resolve {
progress: Vec<ModelArtifactProgress>,
path: &'static str,
cache_hit: bool,
},
WaitForCancel,
Sleep(Duration),
}
struct FakeModelArtifactDownloader {
outcome: Mutex<FakeDownloadOutcome>,
}
impl FakeModelArtifactDownloader {
fn new(outcome: FakeDownloadOutcome) -> Self {
Self {
outcome: Mutex::new(outcome),
}
}
}
#[async_trait]
impl ModelArtifactDownloader for FakeModelArtifactDownloader {
async fn resolve_hf_model(
&self,
repo: &HfModelRef,
progress: Arc<dyn Fn(ModelArtifactProgress) + Send + Sync>,
cancel: ModelArtifactCancel,
) -> Result<ModelArtifactResolution, ModelServerError> {
let outcome = self.outcome.lock().unwrap().clone();
match outcome {
FakeDownloadOutcome::Resolve {
progress: events,
path,
cache_hit,
} => {
if cancel.is_cancelled() {
return Err(ModelServerError::Cancelled);
}
if !cache_hit {
for event in events {
progress(ModelArtifactProgress {
source: event.source.or_else(|| Some(repo.as_str().to_owned())),
..event
});
}
}
Ok(ModelArtifactResolution {
path: ModelPath::new(path).unwrap(),
cache_hit,
})
}
FakeDownloadOutcome::WaitForCancel => loop {
if cancel.is_cancelled() {
return Err(ModelServerError::Cancelled);
}
tokio::time::sleep(Duration::from_millis(1)).await;
},
FakeDownloadOutcome::Sleep(duration) => {
tokio::time::sleep(duration).await;
Err(ModelServerError::Probe("unexpected wake".to_owned()))
}
}
}
}
#[derive(Default)]
struct FakeFs {
existing: Mutex<Vec<String>>,
@ -345,6 +414,26 @@ fn ensure(
.with_hf_download_deadline(Duration::from_secs(5))
}
fn ensure_with_downloader(
registry: Arc<FakeRegistry>,
probe: Arc<FakeProbe>,
process: Arc<FakeProcess>,
fs: Arc<FakeFs>,
events: Arc<FakeEvents>,
downloader: Arc<FakeModelArtifactDownloader>,
) -> EnsureLocalModelServer {
ensure(registry, probe, process, fs, events)
.with_model_artifact_downloader(downloader as Arc<dyn ModelArtifactDownloader>)
}
fn progress(downloaded: Option<u64>, total: Option<u64>) -> ModelArtifactProgress {
ModelArtifactProgress {
downloaded_bytes: downloaded,
total_bytes: total,
source: None,
}
}
#[tokio::test]
async fn reachable_server_is_reused_without_spawn() {
let registry = Arc::new(FakeRegistry::default());
@ -738,6 +827,270 @@ async fn hf_download_phase_fails_when_process_exits() {
}));
}
#[tokio::test]
async fn hf_downloader_publishes_debounced_progress_and_spawns_resolved_model_path() {
let registry = Arc::new(FakeRegistry::default());
registry
.save(hf_config(
sid(14),
8094,
"Qwen/Qwen3-Coder-30B-A3B-Instruct-GGUF",
))
.await
.unwrap();
let process = Arc::new(FakeProcess::default());
let events = Arc::new(FakeEvents::default());
let downloader = Arc::new(FakeModelArtifactDownloader::new(
FakeDownloadOutcome::Resolve {
progress: vec![
progress(Some(10), Some(100)),
progress(Some(50), Some(100)),
progress(Some(100), Some(100)),
],
path: "/cache/model.gguf",
cache_hit: false,
},
));
let usecase = ensure_with_downloader(
Arc::clone(&registry),
Arc::new(FakeProbe::new(vec![
ModelServerStatus::Unreachable,
ModelServerStatus::ReadyStarted,
])),
Arc::clone(&process),
Arc::new(FakeFs::default()),
Arc::clone(&events),
downloader,
);
let out = usecase
.execute(EnsureLocalModelServerInput { server_id: sid(14) })
.await
.unwrap();
assert_eq!(out.ready.status, ModelServerStatus::ReadyStarted);
let statuses: Vec<ModelServerLifecycleStatus> = events
.0
.lock()
.unwrap()
.iter()
.filter_map(|event| match event {
DomainEvent::ModelServerStatusChanged { status, .. } => Some(status.clone()),
_ => None,
})
.collect();
assert!(matches!(
statuses.as_slice(),
[
ModelServerLifecycleStatus::Probing,
ModelServerLifecycleStatus::Downloading {
downloaded_bytes: Some(10),
total_bytes: Some(100),
percent: Some(10.0),
..
},
ModelServerLifecycleStatus::Downloading {
downloaded_bytes: Some(100),
total_bytes: Some(100),
percent: Some(100.0),
..
},
ModelServerLifecycleStatus::Starting,
ModelServerLifecycleStatus::Ready { reused: false },
]
));
let spawns = process.spawns.lock().unwrap();
assert_eq!(spawns.len(), 1);
assert_eq!(spawns[0].args[0..2], ["--model", "/cache/model.gguf"]);
assert!(!spawns[0].args.iter().any(|arg| arg == "-hf"));
}
#[tokio::test]
async fn hf_downloader_cache_hit_skips_downloading_events() {
let registry = Arc::new(FakeRegistry::default());
registry
.save(hf_config(sid(15), 8095, "Qwen/Qwen3-Coder:Q4_K_M"))
.await
.unwrap();
let process = Arc::new(FakeProcess::default());
let events = Arc::new(FakeEvents::default());
let downloader = Arc::new(FakeModelArtifactDownloader::new(
FakeDownloadOutcome::Resolve {
progress: Vec::new(),
path: "/cache/q4.gguf",
cache_hit: true,
},
));
let usecase = ensure_with_downloader(
Arc::clone(&registry),
Arc::new(FakeProbe::new(vec![
ModelServerStatus::Unreachable,
ModelServerStatus::ReadyStarted,
])),
Arc::clone(&process),
Arc::new(FakeFs::default()),
Arc::clone(&events),
downloader,
);
usecase
.execute(EnsureLocalModelServerInput { server_id: sid(15) })
.await
.unwrap();
let statuses: Vec<ModelServerLifecycleStatus> = events
.0
.lock()
.unwrap()
.iter()
.filter_map(|event| match event {
DomainEvent::ModelServerStatusChanged { status, .. } => Some(status.clone()),
_ => None,
})
.collect();
assert!(matches!(
statuses.as_slice(),
[
ModelServerLifecycleStatus::Probing,
ModelServerLifecycleStatus::Starting,
ModelServerLifecycleStatus::Ready { reused: false },
]
));
}
#[tokio::test]
async fn hf_downloader_unknown_total_publishes_no_percent() {
let registry = Arc::new(FakeRegistry::default());
registry
.save(hf_config(sid(16), 8096, "Qwen/Qwen3-Coder:Q4_K_M"))
.await
.unwrap();
let process = Arc::new(FakeProcess::default());
let events = Arc::new(FakeEvents::default());
let downloader = Arc::new(FakeModelArtifactDownloader::new(
FakeDownloadOutcome::Resolve {
progress: vec![progress(Some(10), None)],
path: "/cache/q4.gguf",
cache_hit: false,
},
));
let usecase = ensure_with_downloader(
Arc::clone(&registry),
Arc::new(FakeProbe::new(vec![
ModelServerStatus::Unreachable,
ModelServerStatus::ReadyStarted,
])),
Arc::clone(&process),
Arc::new(FakeFs::default()),
Arc::clone(&events),
downloader,
);
usecase
.execute(EnsureLocalModelServerInput { server_id: sid(16) })
.await
.unwrap();
assert!(events.0.lock().unwrap().iter().any(|event| {
matches!(
event,
DomainEvent::ModelServerStatusChanged {
status: ModelServerLifecycleStatus::Downloading {
downloaded_bytes: Some(10),
total_bytes: None,
percent: None,
..
},
..
}
)
}));
}
#[tokio::test]
async fn hf_downloader_cancel_fails_without_process_spawn() {
let registry = Arc::new(FakeRegistry::default());
registry
.save(hf_config(sid(17), 8097, "Qwen/Qwen3-Coder:Q4_K_M"))
.await
.unwrap();
let process = Arc::new(FakeProcess::default());
let events = Arc::new(FakeEvents::default());
let downloader = Arc::new(FakeModelArtifactDownloader::new(
FakeDownloadOutcome::WaitForCancel,
));
let usecase = Arc::new(ensure_with_downloader(
Arc::clone(&registry),
Arc::new(FakeProbe::new(vec![ModelServerStatus::Unreachable])),
Arc::clone(&process),
Arc::new(FakeFs::default()),
Arc::clone(&events),
downloader,
));
let task_usecase = Arc::clone(&usecase);
let task = tokio::spawn(async move {
task_usecase
.execute(EnsureLocalModelServerInput { server_id: sid(17) })
.await
});
tokio::time::sleep(Duration::from_millis(5)).await;
usecase.stop_on_app_exit().await.unwrap();
let err = task.await.unwrap().unwrap_err();
assert!(err.to_string().contains("cancelled"));
assert!(process.spawns.lock().unwrap().is_empty());
assert!(events.0.lock().unwrap().iter().any(|event| {
matches!(
event,
DomainEvent::ModelServerStatusChanged {
status: ModelServerLifecycleStatus::Failed { code, .. },
..
} if code == "cancelled"
)
}));
}
#[tokio::test]
async fn hf_downloader_timeout_fails_without_process_spawn() {
let registry = Arc::new(FakeRegistry::default());
registry
.save(hf_config(sid(18), 8098, "Qwen/Qwen3-Coder:Q4_K_M"))
.await
.unwrap();
let process = Arc::new(FakeProcess::default());
let events = Arc::new(FakeEvents::default());
let downloader = Arc::new(FakeModelArtifactDownloader::new(
FakeDownloadOutcome::Sleep(Duration::from_millis(50)),
));
let usecase = ensure_with_downloader(
Arc::clone(&registry),
Arc::new(FakeProbe::new(vec![ModelServerStatus::Unreachable])),
Arc::clone(&process),
Arc::new(FakeFs::default()),
Arc::clone(&events),
downloader,
)
.with_hf_download_deadline(Duration::from_millis(1));
let err = usecase
.execute(EnsureLocalModelServerInput { server_id: sid(18) })
.await
.unwrap_err();
assert!(err.to_string().contains("timeout"));
assert!(process.spawns.lock().unwrap().is_empty());
assert!(events.0.lock().unwrap().iter().any(|event| {
matches!(
event,
DomainEvent::ModelServerStatusChanged {
status: ModelServerLifecycleStatus::Failed { code, .. },
..
} if code == "timeout"
)
}));
}
#[tokio::test]
async fn missing_registry_entry_is_model_server_not_configured() {
let usecase = ensure(