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; +}