From 091902039dde912f0af69da4f0ca5e9656984416 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 09:59:16 +0000 Subject: [PATCH] =?UTF-8?q?Lot=206=20:=20bail=20du=20moteur=20c=C3=B4t?= =?UTF-8?q?=C3=A9=20Rust=20(chat::lease)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit EngineLease prend le verrou de génération du Go, le renouvelle au tiers de sa durée en tâche de fond et signale au tour un stop passé par le Go ou la perte du bail (Reset, bascule, expiration). Rendu par release, ou en tâche de fond s'il est lâché. engine_stop arrête un détenteur Go (tâche, bench, optimiseur). Store::internal_post garde le statut HTTP pour distinguer les refus. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016tcGfJWefW5kUGoC2wL8cv --- rust/loki-rs/src/chat/lease.rs | 207 ++++++++++++++++++++++++++++ rust/loki-rs/src/chat/mod.rs | 4 +- rust/loki-rs/src/store.rs | 18 +++ rust/loki-rs/tests/common/fakego.rs | 53 +++++++ rust/loki-rs/tests/lot6.rs | 96 +++++++++++++ 5 files changed, 377 insertions(+), 1 deletion(-) create mode 100644 rust/loki-rs/src/chat/lease.rs diff --git a/rust/loki-rs/src/chat/lease.rs b/rust/loki-rs/src/chat/lease.rs new file mode 100644 index 0000000..0920aec --- /dev/null +++ b/rust/loki-rs/src/chat/lease.rs @@ -0,0 +1,207 @@ +//! Bail du moteur — le verrou de génération du Go (`conv.Generating`), pris +//! par une conversation tenue par Rust (`web_internal_lease.go`). +//! +//! Le verrou est partagé avec les tâches planifiées, le bench et l'optimiseur, +//! qui restent au Go : un tour Rust doit le tenir du début à la fin, sinon une +//! tâche partirait vers le même moteur en plein tour. +//! +//! [`EngineLease::acquire`] prend le bail et le renouvelle en tâche de fond +//! (au tiers de sa durée). Le Go le rompt (Reset, bascule) ou y signale un +//! stop : [`EngineLease::interrupted`] le dit au tour. Une façade tuée ne +//! renouvelle plus : le Go rend le verrou à l'expiration. [`EngineLease::release`] +//! le rend tout de suite ; un bail lâché sans `release` est rendu en tâche de +//! fond. + +use std::time::Duration; + +use hyper::StatusCode; +use serde_json::{Value, json}; +use tokio::sync::watch; +use tokio::task::JoinHandle; + +use crate::store::Store; + +/// Durée d'un bail par défaut (`extLeaseDefaultTTL`). +pub const LEASE_TTL: Duration = Duration::from_secs(30); + +const PATH_LEASE: &str = "/internal/v1/engine/lease"; +const PATH_RENEW: &str = "/internal/v1/engine/lease/renew"; +const PATH_RELEASE: &str = "/internal/v1/engine/lease/release"; +const PATH_STOP: &str = "/internal/v1/engine/stop"; + +/// Refus de prendre le bail. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum LeaseError { + /// Le moteur est tenu (tour, tâche planifiée, bench, optimiseur). `error` + /// est le message du Go (`busyReason`) ; la tâche est nommée s'il s'agit + /// d'une tâche planifiée. + Busy { + error: String, + task_id: Option, + task_name: Option, + }, + /// Go injoignable ou réponse inattendue. + Other(String), +} + +impl std::fmt::Display for LeaseError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + LeaseError::Busy { error, .. } => f.write_str(error), + LeaseError::Other(e) => write!(f, "bail du moteur : {e}"), + } + } +} + +/// État du bail vu du tour. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LeaseSignal { + Held, + /// Stop demandé côté Go : le tour s'arrête, le bail tient jusqu'à `release`. + Stop, + /// Bail rompu (Reset ou bascule côté Go, ou expiré) : le verrou n'est plus + /// à nous. + Lost, +} + +/// Le bail tenu ; rendu par [`Self::release`] ou à la destruction. +pub struct EngineLease { + id: String, + store: Store, + signal: watch::Receiver, + renew: JoinHandle<()>, + released: bool, +} + +impl EngineLease { + /// Prend le verrou de génération pour `ttl` (borné par le Go à 5 s – 5 min). + pub async fn acquire(store: Store, ttl: Duration) -> Result { + let body = json!({ "ttl_ms": ttl.as_millis() as u64 }); + let (status, v) = store + .internal_post(PATH_LEASE, &body) + .await + .map_err(LeaseError::Other)?; + match status { + StatusCode::OK => {} + StatusCode::CONFLICT => { + let s = |k: &str| v.get(k).and_then(Value::as_str).map(str::to_string); + return Err(LeaseError::Busy { + error: s("error").unwrap_or_else(|| "génération en cours".into()), + task_id: s("task_id"), + task_name: s("task_name"), + }); + } + s => return Err(LeaseError::Other(format!("{PATH_LEASE} : {s} {v}"))), + } + let id = v + .get("lease") + .and_then(Value::as_str) + .filter(|s| !s.is_empty()) + .ok_or_else(|| LeaseError::Other(format!("{PATH_LEASE} : réponse sans bail {v}")))? + .to_string(); + let granted = v + .get("ttl_ms") + .and_then(Value::as_u64) + .map(Duration::from_millis) + .unwrap_or(ttl); + let (tx, rx) = watch::channel(LeaseSignal::Held); + let renew = tokio::spawn(renew_loop(store.clone(), id.clone(), granted / 3, tx)); + Ok(EngineLease { + id, + store, + signal: rx, + renew, + released: false, + }) + } + + pub fn id(&self) -> &str { + &self.id + } + + /// État courant (dernier renouvellement). + pub fn signal(&self) -> LeaseSignal { + *self.signal.borrow() + } + + /// Se résout au premier stop ou à la perte du bail, et rend lequel. + pub async fn interrupted(&mut self) -> LeaseSignal { + match self.signal.wait_for(|s| *s != LeaseSignal::Held).await { + Ok(s) => *s, + // Tâche de renouvellement terminée sans signal : rien à attendre. + Err(_) => std::future::pending().await, + } + } + + /// Rend le verrou maintenant. + pub async fn release(mut self) { + self.released = true; + self.renew.abort(); + release(&self.store, &self.id).await; + } +} + +impl Drop for EngineLease { + fn drop(&mut self) { + self.renew.abort(); + if self.released { + return; + } + // Lâché sans release (tour annulé, panique) : rendu en tâche de fond. + // Hors runtime, l'expiration côté Go s'en charge. + if let Ok(rt) = tokio::runtime::Handle::try_current() { + let (store, id) = (self.store.clone(), std::mem::take(&mut self.id)); + rt.spawn(async move { release(&store, &id).await }); + } + } +} + +async fn release(store: &Store, id: &str) { + if let Err(e) = store + .internal_post(PATH_RELEASE, &json!({ "lease": id })) + .await + { + eprintln!("[lease] libération du bail échouée (le Go le rendra à l'expiration) : {e}"); + } +} + +/// Renouvelle jusqu'à la perte du bail. Une erreur réseau n'arrête pas la +/// boucle : le prochain essai peut passer avant l'expiration. +async fn renew_loop(store: Store, id: String, every: Duration, tx: watch::Sender) { + let mut tick = tokio::time::interval(every.max(Duration::from_millis(100))); + tick.tick().await; // le premier top est immédiat : le bail vient d'être pris + loop { + tick.tick().await; + match store + .internal_post(PATH_RENEW, &json!({ "lease": id })) + .await + { + Ok((StatusCode::OK, v)) => { + if v.get("stop").and_then(Value::as_bool) == Some(true) { + tx.send_if_modified(|s| { + let was = *s; + if was == LeaseSignal::Held { + *s = LeaseSignal::Stop; + } + was == LeaseSignal::Held + }); + } + } + Ok((StatusCode::GONE, _)) => { + tx.send_replace(LeaseSignal::Lost); + return; + } + Ok((s, v)) => eprintln!("[lease] renouvellement : {s} {v}"), + Err(e) => eprintln!("[lease] renouvellement : {e}"), + } + } +} + +/// Arrête le détenteur Go du moteur (tâche planifiée, bench, optimiseur) : +/// le bouton stop du chat, quand ce n'est pas un tour Rust qui tourne. +pub async fn engine_stop(store: &Store) -> Result<(), String> { + match store.internal_post(PATH_STOP, &json!({})).await? { + (StatusCode::NO_CONTENT, _) => Ok(()), + (s, v) => Err(format!("{PATH_STOP} : {s} {v}")), + } +} diff --git a/rust/loki-rs/src/chat/mod.rs b/rust/loki-rs/src/chat/mod.rs index e7af56a..3dc6b37 100644 --- a/rust/loki-rs/src/chat/mod.rs +++ b/rust/loki-rs/src/chat/mod.rs @@ -8,11 +8,13 @@ //! d'outils, statistiques) ; //! - [`compact`] : estimation du contexte et compactage de l'historique ; //! - [`journal`] : journal d'affichage (compactage, rejeu, pagination) ; -//! - [`persist`] : format en base, index des discussions, écrivain unique. +//! - [`persist`] : format en base, index des discussions, écrivain unique ; +//! - [`lease`] : bail du moteur (verrou de génération partagé avec le Go). pub mod compact; pub mod engine; pub mod journal; +pub mod lease; pub mod persist; pub mod stream; pub mod types; diff --git a/rust/loki-rs/src/store.rs b/rust/loki-rs/src/store.rs index 7cdd6f5..3697c0d 100644 --- a/rust/loki-rs/src/store.rs +++ b/rust/loki-rs/src/store.rs @@ -86,6 +86,24 @@ impl Store { serde_json::from_slice(&bytes).map_err(|e| e.to_string()) } + /// POST JSON à l'API interne, statut compris (refus métier à distinguer : + /// 409, 410…). Corps vide ou illisible → `Value::Null`. + pub async fn internal_post( + &self, + path: &str, + body: &serde_json::Value, + ) -> Result<(StatusCode, serde_json::Value), String> { + let (status, bytes) = self + .call( + Method::POST, + path, + serde_json::to_vec(body).unwrap_or_default(), + ) + .await?; + let v = serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null); + Ok((status, v)) + } + /// Scelle un blob avec la clé de la mémoire (gardée par le Go). Clair /// rendu tel quel si le chiffrement est coupé. pub async fn seal(&self, plain: Vec) -> Result, CryptError> { diff --git a/rust/loki-rs/tests/common/fakego.rs b/rust/loki-rs/tests/common/fakego.rs index b9c3b87..9876e97 100644 --- a/rust/loki-rs/tests/common/fakego.rs +++ b/rust/loki-rs/tests/common/fakego.rs @@ -32,6 +32,11 @@ pub struct FakeGo { pub oai_log: Mutex>, /// Mémoire verrouillée : seal/open répondent 423. pub locked: Mutex, + /// Bail du moteur : bail courant, stop signalé, détenteur Go, journal. + pub lease: Mutex>, + pub lease_stop: Mutex, + pub go_busy: Mutex>, + pub lease_log: Mutex>, /// Adresse d'écoute du faux Go (pour un client `Store` direct). pub backend: Mutex>, } @@ -118,6 +123,54 @@ async fn handle(st: Arc, req: Request) -> Response resp(200, body.strip_prefix(b"SCELLE:").unwrap_or(&body).to_vec()) } } + "/internal/v1/engine/lease" => { + let body = req.into_body().collect().await.unwrap().to_bytes(); + let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default(); + if let Some(why) = st.go_busy.lock().unwrap().clone() { + let out = serde_json::json!({"error": why, "task_id": "t1", "task_name": "veille"}); + return resp(409, out.to_string()); + } + let mut lease = st.lease.lock().unwrap(); + if lease.is_some() { + return resp(409, r#"{"error":"génération en cours"}"#); + } + let n = st.lease_log.lock().unwrap().len(); + let id = format!("bail{n}"); + *lease = Some(id.clone()); + st.lease_log + .lock() + .unwrap() + .push(format!("acquire {}", v["ttl_ms"])); + resp( + 200, + serde_json::json!({"lease": id, "ttl_ms": v["ttl_ms"]}).to_string(), + ) + } + "/internal/v1/engine/lease/renew" | "/internal/v1/engine/lease/release" => { + let body = req.into_body().collect().await.unwrap().to_bytes(); + let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default(); + let id = v["lease"].as_str().unwrap_or_default().to_string(); + let mut lease = st.lease.lock().unwrap(); + let held = lease.as_deref() == Some(id.as_str()); + if path.ends_with("release") { + st.lease_log.lock().unwrap().push(format!("release {id}")); + if held { + *lease = None; + } + return resp(204, ""); + } + st.lease_log.lock().unwrap().push(format!("renew {id}")); + if !held { + return resp(410, r#"{"error":"bail perdu"}"#); + } + let stop = *st.lease_stop.lock().unwrap(); + resp(200, serde_json::json!({"stop": stop}).to_string()) + } + "/internal/v1/engine/stop" => { + st.lease_log.lock().unwrap().push("stop".into()); + *st.go_busy.lock().unwrap() = None; + resp(204, "") + } "/internal/v1/tool/exec" => { let body = req.into_body().collect().await.unwrap().to_bytes(); let v: serde_json::Value = serde_json::from_slice(&body).unwrap(); diff --git a/rust/loki-rs/tests/lot6.rs b/rust/loki-rs/tests/lot6.rs index ec66183..e425f09 100644 --- a/rust/loki-rs/tests/lot6.rs +++ b/rust/loki-rs/tests/lot6.rs @@ -145,3 +145,99 @@ async fn index_illisible_jamais_reecrit() { assert_eq!(opened(&go, BK_CHAT, "conv:c1").unwrap(), b"{}"); assert_eq!(go.get(BK_CHAT, CK_INDEX).unwrap(), "SCELLE:pas du json"); } + +// ---- bail du moteur ---- + +use loki_rs::chat::lease::{EngineLease, LeaseError, LeaseSignal, engine_stop}; +use std::time::Duration; + +async fn until(mut f: impl FnMut() -> bool) { + tokio::time::timeout(Duration::from_secs(5), async { + while !f() { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("condition jamais atteinte"); +} + +#[tokio::test] +async fn bail_pris_renouvele_rendu() { + let (st, go, _stop, _t) = store().await; + // TTL de 300 ms : renouvelé toutes les 100 ms. + let lease = EngineLease::acquire(st.clone(), Duration::from_millis(300)) + .await + .unwrap(); + assert_eq!(lease.id(), "bail0"); + assert_eq!(go.lease_log.lock().unwrap()[0], "acquire 300"); + // Un second bail est refusé tant que le premier tient. + match EngineLease::acquire(st.clone(), Duration::from_secs(30)).await { + Err(LeaseError::Busy { error, .. }) => assert_eq!(error, "génération en cours"), + other => panic!("second bail : {:?}", other.map(|l| l.id().to_string())), + } + until(|| { + go.lease_log + .lock() + .unwrap() + .iter() + .any(|l| l == "renew bail0") + }) + .await; + assert_eq!(lease.signal(), LeaseSignal::Held); + lease.release().await; + assert_eq!(*go.lease.lock().unwrap(), None); + assert!( + go.lease_log + .lock() + .unwrap() + .contains(&"release bail0".to_string()) + ); +} + +#[tokio::test] +async fn bail_stop_puis_perdu() { + let (st, go, _stop, _t) = store().await; + let mut lease = EngineLease::acquire(st, Duration::from_millis(300)) + .await + .unwrap(); + *go.lease_stop.lock().unwrap() = true; + let got = tokio::time::timeout(Duration::from_secs(5), lease.interrupted()).await; + assert_eq!(got.unwrap(), LeaseSignal::Stop); + // Le Go rompt le bail (Reset) : perdu. + *go.lease.lock().unwrap() = None; + until(|| lease.signal() == LeaseSignal::Lost).await; + assert_eq!(lease.interrupted().await, LeaseSignal::Lost); +} + +#[tokio::test] +async fn bail_lache_sans_release_est_rendu() { + let (st, go, _stop, _t) = store().await; + let lease = EngineLease::acquire(st, Duration::from_secs(30)) + .await + .unwrap(); + drop(lease); + until(|| go.lease.lock().unwrap().is_none()).await; +} + +#[tokio::test] +async fn moteur_tenu_par_une_tache_go() { + let (st, go, _stop, _t) = store().await; + *go.go_busy.lock().unwrap() = Some("la tâche planifiée « veille » occupe le modèle".into()); + match EngineLease::acquire(st.clone(), Duration::from_secs(30)).await { + Err(LeaseError::Busy { + error, + task_id, + task_name, + }) => { + assert!(error.contains("veille")); + assert_eq!(task_id.as_deref(), Some("t1")); + assert_eq!(task_name.as_deref(), Some("veille")); + } + other => panic!("bail accordé : {:?}", other.map(|l| l.id().to_string())), + } + engine_stop(&st).await.unwrap(); + let lease = EngineLease::acquire(st, Duration::from_secs(30)) + .await + .unwrap(); + lease.release().await; +}