From ba4e942d2110416f7d2c326e84cad224a6f02895 Mon Sep 17 00:00:00 2001 From: "godon-robot[bot]" <252949883+godon-robot[bot]@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:48:49 +0000 Subject: [PATCH 1/2] test(rung10): the port opens - instruments read ported, verdict read carries the worker's stamp The parity bar's first module (test_instruments_read.py) found a real gap: the rust read_state_verdict carried the pre-rung-9 shape - three columns, no instruments pickup. Ported the python faithfully: the four-column read, the message-matched rollout-order guard (one retry, no ladder), the lenient stamp parse (torn reads absent), and the instruments key always present in the verdict's order. The port harness lands with it (tests/port/): FakeDb/FakeWindmill/ FakeCausal behind the gateway seams - the rust shape of what python's suite got from unittest.mock.patch. Co-authored-by: cherusk <10729954+cherusk@users.noreply.github.com> --- src/database.rs | 31 ++++- tests/port/instruments_read.rs | 68 ++++++++++ tests/port/main.rs | 9 ++ tests/port/support.rs | 226 +++++++++++++++++++++++++++++++++ 4 files changed, 332 insertions(+), 2 deletions(-) create mode 100644 tests/port/instruments_read.rs create mode 100644 tests/port/main.rs create mode 100644 tests/port/support.rs diff --git a/src/database.rs b/src/database.rs index c7245eb..19c4329 100644 --- a/src/database.rs +++ b/src/database.rs @@ -325,13 +325,30 @@ impl ArchiveDatabaseRepository { /// "not found in Raft group"). pub fn read_state_verdict(&self, systemtender_db_name: &str) -> DbResult> { let db_config = self.base_config.with_database(systemtender_db_name); - let rows = match self.db.execute_query( + // The instruments column is rollout-order guarded, message-matched + // (the worker's own beat-guard pattern): a state row born before + // the column - fresh tender, worker has not stamped yet - reads + // the three verdict columns alone, instruments None. + let attempted = match self.db.execute_query( &db_config, "SELECT finished_at IS NOT NULL, \ EXTRACT(EPOCH FROM (now() - updated_at)), \ - beat_interval_secs FROM systemtender_state LIMIT 1", + beat_interval_secs, instruments \ + FROM systemtender_state LIMIT 1", true, ) { + Err(e) if e.0.contains("instruments") && e.0.contains("does not exist") => { + self.db.execute_query( + &db_config, + "SELECT finished_at IS NOT NULL, \ + EXTRACT(EPOCH FROM (now() - updated_at)), \ + beat_interval_secs FROM systemtender_state LIMIT 1", + true, + ) + } + other => other, + }; + let rows = match attempted { Ok(r) => r, Err(e) => { let error_text = e.0.clone(); @@ -364,6 +381,16 @@ impl ArchiveDatabaseRepository { None => Value::Null, }, ); + // The worker's JSON stamp, parsed leniently - a torn + // payload reads as absent, never as a failed read. + let mut instruments = Value::Null; + if let Some(Some(cell)) = row.get(3) { + if !cell.is_empty() { + instruments = + serde_json::from_str::(cell).unwrap_or(Value::Null); + } + } + m.insert("instruments".to_string(), instruments); return Ok(Some(Value::Object(m))); } } diff --git a/tests/port/instruments_read.rs b/tests/port/instruments_read.rs new file mode 100644 index 0000000..c2445cb --- /dev/null +++ b/tests/port/instruments_read.rs @@ -0,0 +1,68 @@ +//! Ported from tests/unit/test_instruments_read.py - rung 9, the +//! instruments door's read side: the verdict read picks up the +//! worker's JSON stamp with a rollout-order guard (message-matched, +//! the worker's own beat-guard pattern); a torn payload reads as +//! absent, never as a failed read; the missing-table wordings still +//! carry no verdict. + +use crate::support::{srow, FakeDb}; +use godon_controller::database::{ArchiveDatabaseRepository, DbConfig}; +use serde_json::json; +use std::sync::Arc; + +fn repo_with(db: Arc) -> ArchiveDatabaseRepository { + ArchiveDatabaseRepository::with_gateway(DbConfig::archive_from_env(), db) +} + +#[test] +fn verdict_carries_the_stamp() { + let db = Arc::new(FakeDb::rows(vec![srow(&[ + "f", + "12.0", + "10.0", + r#"{"strategies_run": ["hold"], "wish": null, "seek_spend": null}"#, + ])])); + let repo = repo_with(db); + let verdict = repo.read_state_verdict("db").unwrap().unwrap(); + assert_eq!( + verdict["instruments"], + json!({"strategies_run": ["hold"], "wish": null, "seek_spend": null}) + ); + assert_eq!(verdict["finished"], json!(false)); +} + +#[test] +fn rollout_order_falls_back() { + // A row born before the column: the three verdict columns alone, + // instruments None - one retry, no ladder. + let db = Arc::new(FakeDb::from_fn(|_db, query| { + if query.contains("instruments") { + return Err(godon_controller::database::DbError( + r#"column "instruments" does not exist"#.to_string(), + )); + } + Ok(Some(vec![srow(&["f", "12.0", "10.0"])])) + })); + let repo = repo_with(db.clone()); + let verdict = repo.read_state_verdict("db").unwrap().unwrap(); + assert_eq!(verdict["instruments"], json!(null)); + assert_eq!(db.queries().len(), 2); +} + +#[test] +fn missing_table_still_no_verdict() { + let db = Arc::new(FakeDb::err( + r#"relation "systemtender_state" does not exist"#, + )); + let repo = repo_with(db); + assert!(repo.read_state_verdict("db").unwrap().is_none()); +} + +#[test] +fn torn_payload_reads_absent() { + let db = Arc::new(FakeDb::rows(vec![srow(&["f", "12.0", "10.0", "{torn json"])])); + let repo = repo_with(db); + let verdict = repo.read_state_verdict("db").unwrap().unwrap(); + assert_eq!(verdict["instruments"], json!(null)); + assert_eq!(verdict["finished"], json!(false)); +} diff --git a/tests/port/main.rs b/tests/port/main.rs new file mode 100644 index 0000000..125c070 --- /dev/null +++ b/tests/port/main.rs @@ -0,0 +1,9 @@ +//! The python unit suite, ported module by module - rung 10's parity +//! bar. One file here per python file under tests/unit/; the fakes +//! python got from unittest.mock.patch live in support, injected +//! through the DbGateway / WindmillGateway / CausalClient seams +//! instead of module patching. + +mod support; + +mod instruments_read; diff --git a/tests/port/support.rs b/tests/port/support.rs new file mode 100644 index 0000000..a254fb4 --- /dev/null +++ b/tests/port/support.rs @@ -0,0 +1,226 @@ +//! The shared fakes: what python's suite got from +//! `patch('controller.database.execute_query', ...)` and friends, +//! as gateways injected at construction. +//! +//! Rows are TEXT cells - the simple query protocol's shape - where +//! python's fakes handed typed tuples (psycopg2's decoded cells). A +//! python row `(False, 12.0, 'x')` ports as `srow(&["f", "12.0", +//! "x"])`. + +use godon_controller::database::{DbConfig, DbError, DbGateway, DbResult, TextRow}; +use godon_controller::wmill::{CausalClient, WindmillGateway}; +use serde_json::Value; +use std::collections::VecDeque; +use std::sync::{Arc, Mutex}; + +/// All-Some row of text cells. +pub fn srow(cells: &[&str]) -> TextRow { + cells.iter().map(|c| Some(c.to_string())).collect() +} + +/// Row with explicit NULLs: `row(&[Some("x"), None])`. +pub fn row(cells: &[Option<&str>]) -> TextRow { + cells.iter().map(|c| c.map(|s| s.to_string())).collect() +} + +pub type DbResponder = Box DbResult>> + Send + Sync>; + +/// The fake behind every ported test that patched execute_query. +/// Records (database, query) per call; answers from the responder. +pub struct FakeDb { + pub calls: Mutex>, + pub responder: DbResponder, +} + +impl FakeDb { + /// python's `side_effect=f(db_config, query)` - sees database name + /// and full query text. + pub fn from_fn( + f: impl Fn(&str, &str) -> DbResult>> + Send + Sync + 'static, + ) -> Self { + Self { + calls: Mutex::new(Vec::new()), + responder: Box::new(f), + } + } + + /// python's `return_value=[rows]` - every call answers the same. + pub fn rows(rows: Vec) -> Self { + Self::from_fn(move |_, _| Ok(Some(rows.clone()))) + } + + /// Answer nothing, successfully (the DDL/verb-write shape). + pub fn ok() -> Self { + Self::rows(Vec::new()) + } + + /// Always raise (python's `side_effect=Exception(...)`). + pub fn err(text: &str) -> Self { + Self::from_fn(move |_, _| Err(DbError(text.to_string()))) + } + + pub fn queries(&self) -> Vec { + self.calls + .lock() + .unwrap() + .iter() + .map(|(_, q)| q.clone()) + .collect() + } + + pub fn databases(&self) -> Vec { + self.calls + .lock() + .unwrap() + .iter() + .map(|(d, _)| d.clone()) + .collect() + } +} + +impl DbGateway for FakeDb { + fn execute_query( + &self, + db_config: &DbConfig, + query: &str, + _with_result: bool, + ) -> DbResult>> { + let db_name = db_config.database.clone().unwrap_or_default(); + self.calls + .lock() + .unwrap() + .push((db_name.clone(), query.to_string())); + (self.responder)(&db_name, query) + } + + fn execute_ddl_query(&self, db_config: &DbConfig, query: &str) -> DbResult<()> { + let db_name = db_config.database.clone().unwrap_or_default(); + self.calls + .lock() + .unwrap() + .push((db_name, query.to_string())); + Ok(()) + } +} + +fn pop(q: &Mutex>, default: T) -> T { + q.lock().unwrap().pop_front().unwrap_or(default) +} + +/// The fake behind python's `patch('...Windmill', return_value= +/// fake_client)`: per-call scripted outcomes (popped in order, +/// defaults when exhausted) plus a full record of what flew. +#[derive(Default)] +pub struct FakeWindmill { + pub cancel_results: Mutex>>, + pub job_results: Mutex, String>>>, + pub completed_results: Mutex, String>>>, + pub async_results: Mutex>>, + pub sync_results: Mutex>>, + pub cancelled: Mutex)>>, + pub dispatched: Mutex)>>, + pub sync_ran: Mutex>, +} + +impl FakeWindmill { + pub fn queue_async(&self, r: Result) -> &Self { + self.async_results.lock().unwrap().push_back(r); + self + } + pub fn queue_completed(&self, v: Option) -> &Self { + self.completed_results + .lock() + .unwrap() + .push_back(Ok(v)); + self + } + pub fn queue_job(&self, v: Option) -> &Self { + self.job_results.lock().unwrap().push_back(Ok(v)); + self + } +} + +impl WindmillGateway for FakeWindmill { + fn cancel_job(&self, job_id: &str, reason: Option<&str>) -> Result<(), String> { + self.cancelled + .lock() + .unwrap() + .push((job_id.to_string(), reason.map(|r| r.to_string()))); + pop(&self.cancel_results, Ok(())) + } + + fn get_job(&self, _job_id: &str) -> Result, String> { + pop(&self.job_results, Ok(None)) + } + + fn get_completed_maybe(&self, _job_id: &str) -> Result, String> { + pop(&self.completed_results, Ok(None)) + } + + fn run_script_by_path_async( + &self, + path: &str, + args: &Value, + tag: Option<&str>, + ) -> Result { + self.dispatched.lock().unwrap().push(( + path.to_string(), + args.clone(), + tag.map(|t| t.to_string()), + )); + pop(&self.async_results, Ok("job-fake".to_string())) + } + + fn run_script_by_path(&self, path: &str, args: &Value) -> Result { + self.sync_ran + .lock() + .unwrap() + .push((path.to_string(), args.clone())); + pop(&self.sync_results, Ok(Value::Null)) + } +} + +/// The fake behind python's urlopen/ask patches on the causal seam. +#[derive(Default)] +pub struct FakeCausal { + pub pages: Mutex>>, + pub asks: Mutex>>, + pub purge_results: Mutex>>, + pub asked: Mutex>, + pub purged: Mutex>, +} + +impl FakeCausal { + pub fn queue_page(&self, v: Option) -> &Self { + self.pages.lock().unwrap().push_back(v); + self + } + pub fn queue_ask(&self, r: Result) -> &Self { + self.asks.lock().unwrap().push_back(r); + self + } +} + +impl CausalClient for FakeCausal { + fn fetch_book_page(&self, _wish_id: &str) -> Option { + pop(&self.pages, None) + } + + fn ask_to_plan(&self, plan_request: &Value) -> Result { + self.asks.lock().unwrap().push(plan_request.clone()); + pop(&self.asks, Ok(Value::Null)) + } + + fn purge_curves(&self, systemtender_id: &str) -> Result { + self.purged.lock().unwrap().push(systemtender_id.to_string()); + pop(&self.purge_results, Ok(200)) + } +} + +/// Shared convenience: a config with a database name pinned, for +/// responders that branch on the db (python branched on db_config). +pub fn cfg_with_db(name: &str) -> DbConfig { + let mut c = DbConfig::archive_from_env(); + c.database = Some(name.to_string()); + c +} From 96f838d25f11140fcb1c9fd83efcc531783f332a Mon Sep 17 00:00:00 2001 From: "godon-robot[bot]" <252949883+godon-robot[bot]@users.noreply.github.com> Date: Fri, 16 Oct 2026 12:23:21 +0000 Subject: [PATCH 2/2] fix(rung10): port support - VecDeque pushes and the err lifetime Co-authored-by: cherusk <10729954+cherusk@users.noreply.github.com> --- tests/port/support.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/tests/port/support.rs b/tests/port/support.rs index a254fb4..1e0078a 100644 --- a/tests/port/support.rs +++ b/tests/port/support.rs @@ -56,7 +56,8 @@ impl FakeDb { /// Always raise (python's `side_effect=Exception(...)`). pub fn err(text: &str) -> Self { - Self::from_fn(move |_, _| Err(DbError(text.to_string()))) + let text = text.to_string(); + Self::from_fn(move |_, _| Err(DbError(text))) } pub fn queries(&self) -> Vec { @@ -163,7 +164,7 @@ impl WindmillGateway for FakeWindmill { args: &Value, tag: Option<&str>, ) -> Result { - self.dispatched.lock().unwrap().push(( + self.dispatched.lock().unwrap().push_back(( path.to_string(), args.clone(), tag.map(|t| t.to_string()), @@ -207,12 +208,12 @@ impl CausalClient for FakeCausal { } fn ask_to_plan(&self, plan_request: &Value) -> Result { - self.asks.lock().unwrap().push(plan_request.clone()); + self.asks.lock().unwrap().push_back(plan_request.clone()); pop(&self.asks, Ok(Value::Null)) } fn purge_curves(&self, systemtender_id: &str) -> Result { - self.purged.lock().unwrap().push(systemtender_id.to_string()); + self.purged.lock().unwrap().push_back(systemtender_id.to_string()); pop(&self.purge_results, Ok(200)) } }