diff --git a/.github/workflows/guard.yml b/.github/workflows/guard.yml index 17cd979..95ab7d8 100644 --- a/.github/workflows/guard.yml +++ b/.github/workflows/guard.yml @@ -17,7 +17,6 @@ concurrency: jobs: repo: name: repo - if: github.event_name != 'pull_request' runs-on: ubuntu-latest steps: - uses: actions/checkout@v6 diff --git a/.runseal/lib/guard/guard.ts b/.runseal/lib/guard/guard.ts index 3f9c086..0948df7 100644 --- a/.runseal/lib/guard/guard.ts +++ b/.runseal/lib/guard/guard.ts @@ -1,9 +1,9 @@ //! `runseal :guard` — the local validation suite the pre-commit hook runs. //! -//! Mirrors the CI `smoke (ubuntu-latest)` job (cargo fmt/clippy/test, all -//! `--locked` like CI) so "green locally" means "green in CI", plus the Deno -//! checks for the `.runseal/` wrappers that CI does not cover. Runnable on -//! demand as well as from the hook. +//! Mirrors the CI repo + Rust checks (flavor and cargo, all `--locked` like CI) +//! so "green locally" means "green in CI", plus the Deno checks for the +//! `.runseal/` wrappers that CI does not cover. Runnable on demand as well as +//! from the hook. import { run } from "@/lib/std/cmd.ts"; import { join } from "@/lib/std/fs.ts"; @@ -21,6 +21,11 @@ export async function guard(): Promise { const config = ".runseal/deno.json"; const steps: Step[] = [ + { + title: "flavor check", + command: "flavor", + args: ["check", "--root", ".", "--config", "flavor.toml"], + }, { title: "cargo fmt", command: "cargo", args: ["fmt", "--all", "--check"] }, { title: "cargo clippy", diff --git a/.runseal/lib/land/land.ts b/.runseal/lib/land/land.ts index 7e37f3e..d469942 100644 --- a/.runseal/lib/land/land.ts +++ b/.runseal/lib/land/land.ts @@ -72,10 +72,10 @@ export async function land(argv: string[]): Promise { } // 4. Wait for required checks — polled here, quietly, with a bounded timeout. - console.log(`waiting for required checks (timeout ${CHECK_TIMEOUT_MS / 1000}s) ...`); + console.log(`waiting for PR checks (timeout ${CHECK_TIMEOUT_MS / 1000}s) ...`); const outcome = await waitChecks(pr, repo); if (outcome !== "pass") { - return fail(`required checks ${outcome} — PR #${pr} left open`); + return fail(`PR checks ${outcome} — PR #${pr} left open`); } // 5. Squash merge. @@ -126,8 +126,8 @@ async function openPr(branch: string, repo: string): Promise { } /** - * Poll the PR's required checks until they all pass, one fails, or the timeout - * elapses. Parses `--json bucket` from stdout and ignores exit codes so pending + * Poll all PR checks until they pass, one fails, or the timeout elapses. Parses + * `--json bucket` from stdout and ignores exit codes so pending * states are handled uniformly; an empty set means checks have not registered * yet (just after creation). */ @@ -136,7 +136,7 @@ async function waitChecks(pr: number, repo: string): Promise<"pass" | "fail" | " while (Date.now() < deadline) { const result = await capture( "gh", - ["pr", "checks", String(pr), "--required", "--json", "bucket"], + ["pr", "checks", String(pr), "--json", "bucket"], { cwd: repo }, ); let buckets: string[] = []; diff --git a/.runseal/wrappers/guard.ts b/.runseal/wrappers/guard.ts index 08fdb5f..e0c61cd 100644 --- a/.runseal/wrappers/guard.ts +++ b/.runseal/wrappers/guard.ts @@ -1,5 +1,5 @@ -//! `runseal :guard` — run the local validation suite (mirrors CI smoke + Deno -//! checks). Thin entry point; logic lives in the guard module. +//! `runseal :guard` — run the local validation suite (mirrors CI repo/Rust + +//! Deno checks). Thin entry point; logic lives in the guard module. import { guard } from "@/lib/guard/guard.ts"; diff --git a/crates/santi-api/src/server.rs b/crates/santi-api/src/server.rs index 886ab14..beb908d 100644 --- a/crates/santi-api/src/server.rs +++ b/crates/santi-api/src/server.rs @@ -1,5 +1,6 @@ mod error; mod errors; +mod im; mod ingress; mod openapi; mod routes; diff --git a/crates/santi-api/src/server/im.rs b/crates/santi-api/src/server/im.rs new file mode 100644 index 0000000..783bb07 --- /dev/null +++ b/crates/santi-api/src/server/im.rs @@ -0,0 +1,64 @@ +use axum::{ + Json, + extract::{Path, Query, State}, +}; +use santi_core::{ + ImInboxEntry, ImSendRequest, ImSendResponse, IngestOutcome, SantiError, SantiService, +}; + +use super::ApiError; + +#[utoipa::path( + post, + path = "/api/v1/im/send", + request_body = ImSendRequest, + responses( + (status = 200, body = ImSendResponse), + (status = 423, body = SantiError), + (status = 404, body = SantiError), + (status = 500, body = SantiError) + ) +)] +pub(super) async fn send_im( + State(service): State, + Json(request): Json, +) -> Result, ApiError> { + let outcome = service + .im_send(&request.soul_id, &request.participant_id, &request.content) + .map_err(ApiError::from_service)?; + match outcome { + IngestOutcome::Accepted { receipt } => Ok(Json(ImSendResponse { + participant_id: request.participant_id, + receipt, + })), + IngestOutcome::Rejected { error } => Err(ApiError::from_santi(*error)), + } +} + +#[utoipa::path( + get, + path = "/api/v1/im/inbox/{participant_id}", + params( + ("participant_id" = String, Path), + ("since" = Option, Query) + ), + responses( + (status = 200, body = Vec), + (status = 500, body = SantiError) + ) +)] +pub(super) async fn poll_im( + State(service): State, + Path(participant_id): Path, + Query(params): Query, +) -> Result>, ApiError> { + service + .im_poll(&participant_id, params.since.unwrap_or(0)) + .map(Json) + .map_err(ApiError::from_service) +} + +#[derive(serde::Deserialize)] +pub(super) struct ImPollParams { + since: Option, +} diff --git a/crates/santi-api/src/server/openapi.rs b/crates/santi-api/src/server/openapi.rs index 7795bfd..ee2d4e9 100644 --- a/crates/santi-api/src/server/openapi.rs +++ b/crates/santi-api/src/server/openapi.rs @@ -30,8 +30,8 @@ use utoipa::OpenApi; super::errors::errors, super::sse::error_events, super::routes::runtime_snapshot, - super::routes::send_im, - super::routes::poll_im, + super::im::send_im, + super::im::poll_im, crate::bucket::get_bucket_object ), components(schemas( diff --git a/crates/santi-api/src/server/routes.rs b/crates/santi-api/src/server/routes.rs index e7b5423..4528e25 100644 --- a/crates/santi-api/src/server/routes.rs +++ b/crates/santi-api/src/server/routes.rs @@ -8,9 +8,9 @@ use axum::{ use santi_core::{ CompactExecRequest, CompactExecResponse, CompactQueryResponse, CreateSoulRequest, CreateStrandResponse, CreateWebhookRequest, DriveStrandResponse, ForkStrandResponse, - HealthResponse, ImInboxEntry, ImSendRequest, ImSendResponse, IngestOutcome, MaterialRequest, - SantiError, SantiService, SendStrandAcceptedResponse, SendStrandRequest, Soul, Strand, - StrandBudgetSnapshot, StrandDetail, StrandMaterial, StrandRuntimeSnapshot, WebhookSubscription, + HealthResponse, MaterialRequest, SantiError, SantiService, SendStrandAcceptedResponse, + SendStrandRequest, Soul, Strand, StrandBudgetSnapshot, StrandDetail, StrandMaterial, + StrandRuntimeSnapshot, WebhookSubscription, }; use tower_http::{ cors::{Any, CorsLayer}, @@ -50,8 +50,8 @@ pub(super) fn router(service: SantiService) -> Router { .route("/api/v1/strands/{strand_id}/runtime", get(runtime_snapshot)) // IM layer (orthogonal to the runtime; shares the server for cold-start): // send into a soul's IM conversation, poll a participant's passive inbox. - .route("/api/v1/im/send", post(send_im)) - .route("/api/v1/im/inbox/{participant_id}", get(poll_im)) + .route("/api/v1/im/send", post(super::im::send_im)) + .route("/api/v1/im/inbox/{participant_id}", get(super::im::poll_im)) .route( "/api/v1/bucket/{soul_id}/{strand_id}/{*key}", get(crate::bucket::get_bucket_object), @@ -439,67 +439,6 @@ pub(super) async fn strand_budget( .ok_or_else(|| ApiError::not_found("strand not found")) } -// ── IM layer routes ───────────────────────────────────────────────────────── -// The plain IM integrated into santi. `strand send`/the runtime stay source-less; -// the participant address is IM envelope only. Inbound reuses the runtime primitive -// (Text into an `im:` conversation strand). The reply comes back into -// the participant's passive inbox (written by the soul's offline `im reply` egress). - -#[utoipa::path( - post, - path = "/api/v1/im/send", - request_body = ImSendRequest, - responses( - (status = 200, body = ImSendResponse), - (status = 423, body = SantiError), - (status = 404, body = SantiError), - (status = 500, body = SantiError) - ) -)] -pub(super) async fn send_im( - State(service): State, - Json(request): Json, -) -> Result, ApiError> { - let outcome = service - .im_send(&request.soul_id, &request.participant_id, &request.content) - .map_err(ApiError::from_service)?; - match outcome { - IngestOutcome::Accepted { receipt } => Ok(Json(ImSendResponse { - participant_id: request.participant_id, - receipt, - })), - IngestOutcome::Rejected { error } => Err(ApiError::from_santi(*error)), - } -} - -#[utoipa::path( - get, - path = "/api/v1/im/inbox/{participant_id}", - params( - ("participant_id" = String, Path), - ("since" = Option, Query) - ), - responses( - (status = 200, body = Vec), - (status = 500, body = SantiError) - ) -)] -pub(super) async fn poll_im( - State(service): State, - Path(participant_id): Path, - Query(params): Query, -) -> Result>, ApiError> { - service - .im_poll(&participant_id, params.since.unwrap_or(0)) - .map(Json) - .map_err(ApiError::from_service) -} - -#[derive(serde::Deserialize)] -pub(super) struct ImPollParams { - since: Option, -} - pub(super) async fn openapi() -> Json { Json(super::openapi::document()) } diff --git a/crates/santi-api/src/upgrade/finalize.rs b/crates/santi-api/src/upgrade/finalize.rs index 2d93448..4c43af0 100644 --- a/crates/santi-api/src/upgrade/finalize.rs +++ b/crates/santi-api/src/upgrade/finalize.rs @@ -77,29 +77,7 @@ pub fn finalize_at( match &request.terminal { UpgradeTerminal::Upgraded { readiness } => { - store - .resolve_error_incident( - UPGRADE_INCIDENT_KEY, - "upgrade.succeeded", - json!({ - "attempt_id": request.attempt_id, - "artifact": bounded_detail(&request.deb), - "terminal": if matches!(readiness, super::UpgradeReadiness::Degraded) { - "upgraded_degraded" - } else { - "upgraded" - }, - "readiness": readiness, - }), - ) - .map_err(|error| { - Box::new(persistence_error( - &request.attempt_id, - &request.deb, - "upgrade.finalize.resolve_execution", - error, - )) - })?; + resolve_upgrade(&store, &request, *readiness)?; } UpgradeTerminal::RolledBack { failure } | UpgradeTerminal::Failed { failure } => { let terminal = if matches!(request.terminal, UpgradeTerminal::RolledBack { .. }) { @@ -125,6 +103,37 @@ pub fn finalize_at( finalize_handover(paths, &store, request, errors) } +fn resolve_upgrade( + store: &santi_core::SantiStore, + request: &UpgradeFinalizeRequest, + readiness: super::UpgradeReadiness, +) -> Result<(), Box> { + store + .resolve_error_incident( + UPGRADE_INCIDENT_KEY, + "upgrade.succeeded", + json!({ + "attempt_id": request.attempt_id, + "artifact": bounded_detail(&request.deb), + "terminal": if matches!(readiness, super::UpgradeReadiness::Degraded) { + "upgraded_degraded" + } else { + "upgraded" + }, + "readiness": readiness, + }), + ) + .map(|_| ()) + .map_err(|error| { + Box::new(persistence_error( + &request.attempt_id, + &request.deb, + "upgrade.finalize.resolve_execution", + error, + )) + }) +} + fn open_execution_failure( store: &santi_core::SantiStore, scope: &santi_core::ErrorScope, diff --git a/crates/santi-api/tests/server.rs b/crates/santi-api/tests/server.rs index 61b8c48..4ead754 100644 --- a/crates/santi-api/tests/server.rs +++ b/crates/santi-api/tests/server.rs @@ -144,7 +144,7 @@ async fn send_rejection_locks() { } #[tokio::test] -async fn accepted_drive_failure_degrades_health_until_explicit_recovery() { +async fn drive_failure_http_recovery() { let temp = tempfile::tempdir().expect("temp dir"); let database_path = temp.path().join("santi.sqlite"); let service = SantiService::open( diff --git a/crates/santi-api/tests/upgrade.rs b/crates/santi-api/tests/upgrade.rs index d91a8b3..81b8a23 100644 --- a/crates/santi-api/tests/upgrade.rs +++ b/crates/santi-api/tests/upgrade.rs @@ -185,7 +185,7 @@ fn success_orders_steps() { } #[test] -fn degraded_upgrade_does_not_roll_back() { +fn degraded_skips_rollback() { let mut host = FakeHost { probe_result: Ok(UpgradeReadiness::Degraded), ..Default::default() diff --git a/crates/santi-core/src/service/flow/ingest.rs b/crates/santi-core/src/service/flow/ingest.rs index e6adb25..64b773e 100644 --- a/crates/santi-core/src/service/flow/ingest.rs +++ b/crates/santi-core/src/service/flow/ingest.rs @@ -1,4 +1,4 @@ -use crate::store::{StartTurnOutcome, drive::DriveFailureInput}; +use crate::store::{DriveFailureInput, StartTurnOutcome}; use crate::{ DriveStrandResponse, DriveStrandState, ErrorScope, ErrorSource, InboxSource, IngestOutcome, MessageContent, MessageKind, SantiError, SantiStreamPayload, SendStrandAcceptedResponse, diff --git a/crates/santi-core/src/store.rs b/crates/santi-core/src/store.rs index 3c97ca9..6e797f3 100644 --- a/crates/santi-core/src/store.rs +++ b/crates/santi-core/src/store.rs @@ -12,7 +12,6 @@ mod assembly; pub(crate) mod budget; mod compact; mod db; -pub(crate) mod drive; mod errors; mod fork; mod im; @@ -23,6 +22,7 @@ mod turns; use db::*; pub use db::{read_schema_version, soul_memory_file}; +pub(crate) use errors::drive::DriveFailureInput; use rows::{actor_type_db, collect_rows, map_webhook_row, message_state_db}; /// The schema version this binary expects. On open, recognized runtime-schema diff --git a/crates/santi-core/src/store/budget.rs b/crates/santi-core/src/store/budget.rs index a7f19e8..efc0824 100644 --- a/crates/santi-core/src/store/budget.rs +++ b/crates/santi-core/src/store/budget.rs @@ -93,7 +93,7 @@ impl SantiStore { .map_err(|error| error.to_string())?; if let Some(error) = - super::drive::repeat_active_in_conn(&tx, strand_id, "ingest_active_guard")? + super::errors::drive::repeat_active_in_conn(&tx, strand_id, "ingest_active_guard")? { tx.commit().map_err(|error| error.to_string())?; return Ok(IngestOutcome::Rejected { @@ -300,7 +300,7 @@ impl SantiStore { params![turn_id, strand_id, trigger_type, trigger_ref, now], ) .map_err(|error| error.to_string())?; - super::drive::resolve_in_conn(&tx, strand_id, &turn_id, drained_messages.len())?; + super::errors::drive::resolve_in_conn(&tx, strand_id, &turn_id, drained_messages.len())?; tx.commit().map_err(|error| error.to_string())?; Ok(StartTurnOutcome::Started(StartedTurn { turn: turn_by_id(&conn, &turn_id)?.ok_or_else(|| "created turn missing".to_string())?, diff --git a/crates/santi-core/src/store/drive.rs b/crates/santi-core/src/store/errors/drive.rs similarity index 90% rename from crates/santi-core/src/store/drive.rs rename to crates/santi-core/src/store/errors/drive.rs index 25dcbd3..52b3e09 100644 --- a/crates/santi-core/src/store/drive.rs +++ b/crates/santi-core/src/store/errors/drive.rs @@ -1,7 +1,8 @@ use rusqlite::{Connection, params}; use serde_json::json; -use super::{SantiStore, errors}; +use super::{active_in_conn, open_incident_in_conn, resolve_in_conn as resolve_error_in_conn}; +use crate::store::SantiStore; use crate::{ErrorScope, ErrorSource, IncidentDraft, SantiError, catalog}; const DRIVE_DETAIL_BYTES: usize = 4096; @@ -17,16 +18,16 @@ pub(crate) fn drive_incident_key(strand_id: &str) -> String { format!("{}:strand:{strand_id}", catalog::STRAND_DRIVE_FAILED.code) } -pub(super) fn repeat_active_in_conn( +pub(in crate::store) fn repeat_active_in_conn( conn: &Connection, strand_id: &str, operation: &str, ) -> Result, String> { - if errors::active_in_conn(conn, &drive_incident_key(strand_id))?.is_none() { + if active_in_conn(conn, &drive_incident_key(strand_id))?.is_none() { return Ok(None); } let pending = pending_count(conn, strand_id)?; - errors::open_incident_in_conn( + open_incident_in_conn( conn, drive_draft( strand_id, @@ -42,13 +43,13 @@ pub(super) fn repeat_active_in_conn( .map(Some) } -pub(super) fn resolve_in_conn( +pub(in crate::store) fn resolve_in_conn( conn: &Connection, strand_id: &str, turn_id: &str, drained_count: usize, ) -> Result { - errors::resolve_in_conn( + resolve_error_in_conn( conn, &drive_incident_key(strand_id), "strand.drive_started", @@ -82,7 +83,7 @@ impl SantiStore { let mut conn = self.conn.lock().unwrap(); let tx = conn.transaction().map_err(|error| error.to_string())?; let pending = pending_count(&tx, strand_id)?; - let error = errors::open_incident_in_conn(&tx, drive_draft(strand_id, input, pending))?; + let error = open_incident_in_conn(&tx, drive_draft(strand_id, input, pending))?; tx.commit().map_err(|error| error.to_string())?; Ok(error) } diff --git a/crates/santi-core/src/store/errors/mod.rs b/crates/santi-core/src/store/errors/mod.rs index aabed77..a00d706 100644 --- a/crates/santi-core/src/store/errors/mod.rs +++ b/crates/santi-core/src/store/errors/mod.rs @@ -7,6 +7,8 @@ use santi_error::{ use super::SantiStore; use crate::timestamp_now; +pub(crate) mod drive; + const INCIDENT_COLUMNS: &str = r#" id, incident_key, code, status, category, severity, retry, exposure, scope_kind, scope_id, source_component, source_operation, diff --git a/crates/santi-core/tests/service/drive.rs b/crates/santi-core/tests/service/drive.rs index abc8de6..f310a53 100644 --- a/crates/santi-core/tests/service/drive.rs +++ b/crates/santi-core/tests/service/drive.rs @@ -1,221 +1,7 @@ -use super::support::*; -use rusqlite::Connection; - -#[tokio::test] -async fn drive_failure_recovers() { - let temp = tempfile::tempdir().expect("temp dir"); - let database_path = temp.path().join("santi.sqlite"); - let service = SantiService::open( - SantiServiceConfig { - database_path: database_path.display().to_string(), - runtime_root: temp.path().join("runtime").display().to_string(), - execution_root: temp.path().join("execution").display().to_string(), - bind_addr: Some("127.0.0.1:0".to_string()), - }, - Arc::new(FakeProvider::default()), - ) - .expect("open service"); - let strand = service.create_strand().expect("create strand").strand; - let conn = Connection::open(&database_path).expect("open sqlite"); - conn.execute_batch( - r#" - CREATE TRIGGER force_turn_insert_failure - BEFORE INSERT ON turns - BEGIN - SELECT RAISE(ABORT, 'forced turn insert failure'); - END; - "#, - ) - .expect("install failure trigger"); - - let accepted = service - .send_strand( - &strand.id, - SendStrandRequest { - content: vec![MessagePart::Text { - text: "accepted before drive failure".to_string(), - }], - }, - ) - .await - .expect("durable enqueue should remain accepted"); - assert!(accepted.turn.is_none()); - let warning = accepted.receipt.warning.as_deref().expect("drive warning"); - assert_eq!(warning.code, "runtime.strand.drive_failed"); - assert_eq!(warning.context["accepted_before_failure"], true); - assert_eq!(warning.context["inbox_id"], accepted.receipt.inbox_id); - assert_eq!(warning.context["recovery"]["resend"], false); - assert!(service.is_drive_degraded()); - - let runtime = service - .runtime_snapshot(&strand.id) - .expect("runtime") - .expect("strand"); - assert!(runtime.messages.is_empty()); - assert!(runtime.turns.is_empty()); - assert_eq!(runtime.errors.len(), 1); - let incident_id = runtime.errors[0].id.clone(); - assert_eq!(runtime.errors[0].status, santi_core::IncidentStatus::Active); - assert_eq!(runtime.errors[0].occurrence_count, 1); - assert_eq!(pending_count(&conn, &strand.id), 1); - - let rejected = service - .send_strand( - &strand.id, - SendStrandRequest { - content: vec![MessagePart::Text { - text: "must not enter inbox".to_string(), - }], - }, - ) - .await - .expect_err("active drive incident should quick-fail later writes"); - assert_eq!(rejected.code, "runtime.strand.drive_failed"); - assert_eq!(rejected.incident_id.as_deref(), Some(incident_id.as_str())); - assert_eq!(rejected.context["accepted_before_failure"], false); - assert_eq!(pending_count(&conn, &strand.id), 1); - - conn.execute_batch("DROP TRIGGER force_turn_insert_failure;") - .expect("remove failure trigger"); - let driven = service.drive_strand(&strand.id).expect("operator redrive"); - assert_eq!(driven.state, santi_core::DriveStrandState::Started); - let turn = driven.turn.expect("redrive turn"); - let runtime = wait_for_completed_turn(&service, &strand.id, &turn.id).await; - assert!(!service.is_drive_degraded()); - assert_eq!(pending_count(&conn, &strand.id), 0); - assert_eq!(runtime.errors.len(), 1); - assert_eq!(runtime.errors[0].id, incident_id); - assert_eq!( - runtime.errors[0].status, - santi_core::IncidentStatus::Resolved - ); - assert_eq!(runtime.errors[0].occurrence_count, 2); - assert_eq!(runtime.errors[0].revision, 2); - assert_eq!( - runtime.errors[0].resolved_by.as_deref(), - Some("strand.drive_started") - ); - assert!( - runtime - .messages - .iter() - .any(|message| message.content_text == "accepted before drive failure") - ); - assert!( - runtime - .messages - .iter() - .all(|message| { message.content_text != "must not enter inbox" }) - ); - let transitions: i64 = conn - .query_row("SELECT COUNT(*) FROM error_transitions", [], |row| { - row.get(0) - }) - .expect("transition count"); - assert_eq!(transitions, 2); -} - -#[tokio::test] -async fn cold_start_drive_failure_degrades_and_recovers() { - let temp = tempfile::tempdir().expect("temp dir"); - let database_path = temp.path().join("santi.sqlite"); - let config = SantiServiceConfig { - database_path: database_path.display().to_string(), - runtime_root: temp.path().join("runtime").display().to_string(), - execution_root: temp.path().join("execution").display().to_string(), - bind_addr: Some("127.0.0.1:0".to_string()), - }; - let service = SantiService::open(config.clone(), Arc::new(FakeProvider::default())) - .expect("open service"); - let strand = service.create_strand().expect("create strand").strand; - service.begin_shutdown(); - let accepted = service - .send_strand( - &strand.id, - SendStrandRequest { - content: vec![MessagePart::Text { - text: "pending across restart".to_string(), - }], - }, - ) - .await - .expect("shutdown should retain the durable request"); - assert!(accepted.receipt.warning.is_none()); - assert!(accepted.turn.is_none()); - drop(service); - - let conn = Connection::open(&database_path).expect("open sqlite"); - conn.execute_batch( - r#" - CREATE TRIGGER force_cold_start_turn_failure - BEFORE INSERT ON turns - BEGIN - SELECT RAISE(ABORT, 'forced cold-start turn failure'); - END; - "#, - ) - .expect("install failure trigger"); - - let restarted = SantiService::open(config, Arc::new(FakeProvider::default())) - .expect("open restarted service"); - restarted - .resume_pending() - .expect("strand-local drive failure should permit degraded startup"); - assert!(restarted.is_drive_degraded()); - assert_eq!(pending_count(&conn, &strand.id), 1); - let runtime = restarted - .runtime_snapshot(&strand.id) - .expect("runtime") - .expect("strand"); - assert_eq!(runtime.errors.len(), 1); - assert_eq!(runtime.errors[0].code, "runtime.strand.drive_failed"); - assert_eq!(runtime.errors[0].context["accepted_before_failure"], false); - assert_eq!(runtime.errors[0].context["pending_count"], 1); - assert_eq!(runtime.errors[0].source.operation, "cold_start_resume"); - - let rejected = restarted - .send_strand( - &strand.id, - SendStrandRequest { - content: vec![MessagePart::Text { - text: "must still quick-fail".to_string(), - }], - }, - ) - .await - .expect_err("active cold-start incident should gate writes"); - assert_eq!(rejected.code, "runtime.strand.drive_failed"); - assert_eq!(pending_count(&conn, &strand.id), 1); - - conn.execute_batch("DROP TRIGGER force_cold_start_turn_failure;") - .expect("remove failure trigger"); - let driven = restarted - .drive_strand(&strand.id) - .expect("operator redrive"); - let turn = driven.turn.expect("redrive turn"); - let runtime = wait_for_completed_turn(&restarted, &strand.id, &turn.id).await; - assert!(!restarted.is_drive_degraded()); - assert_eq!(pending_count(&conn, &strand.id), 0); - assert_eq!( - runtime.errors[0].status, - santi_core::IncidentStatus::Resolved - ); - assert!( - runtime - .messages - .iter() - .any(|message| message.content_text == "pending across restart") - ); -} +#[path = "drive/recovery.rs"] +mod recovery; -fn pending_count(conn: &Connection, strand_id: &str) -> i64 { - conn.query_row( - "SELECT COUNT(*) FROM strand_inbox WHERE strand_id = ?1", - [strand_id], - |row| row.get(0), - ) - .expect("pending count") -} +use super::support::*; #[tokio::test] async fn reminder_no_repoke() { diff --git a/crates/santi-core/tests/service/drive/recovery.rs b/crates/santi-core/tests/service/drive/recovery.rs new file mode 100644 index 0000000..9f5429a --- /dev/null +++ b/crates/santi-core/tests/service/drive/recovery.rs @@ -0,0 +1,218 @@ +use super::super::support::*; +use rusqlite::Connection; + +#[tokio::test] +async fn drive_failure_recovers() { + let temp = tempfile::tempdir().expect("temp dir"); + let database_path = temp.path().join("santi.sqlite"); + let service = SantiService::open( + SantiServiceConfig { + database_path: database_path.display().to_string(), + runtime_root: temp.path().join("runtime").display().to_string(), + execution_root: temp.path().join("execution").display().to_string(), + bind_addr: Some("127.0.0.1:0".to_string()), + }, + Arc::new(FakeProvider::default()), + ) + .expect("open service"); + let strand = service.create_strand().expect("create strand").strand; + let conn = Connection::open(&database_path).expect("open sqlite"); + conn.execute_batch( + r#" + CREATE TRIGGER force_turn_insert_failure + BEFORE INSERT ON turns + BEGIN + SELECT RAISE(ABORT, 'forced turn insert failure'); + END; + "#, + ) + .expect("install failure trigger"); + + let accepted = service + .send_strand( + &strand.id, + SendStrandRequest { + content: vec![MessagePart::Text { + text: "accepted before drive failure".to_string(), + }], + }, + ) + .await + .expect("durable enqueue should remain accepted"); + assert!(accepted.turn.is_none()); + let warning = accepted.receipt.warning.as_deref().expect("drive warning"); + assert_eq!(warning.code, "runtime.strand.drive_failed"); + assert_eq!(warning.context["accepted_before_failure"], true); + assert_eq!(warning.context["inbox_id"], accepted.receipt.inbox_id); + assert_eq!(warning.context["recovery"]["resend"], false); + assert!(service.is_drive_degraded()); + + let runtime = service + .runtime_snapshot(&strand.id) + .expect("runtime") + .expect("strand"); + assert!(runtime.messages.is_empty()); + assert!(runtime.turns.is_empty()); + assert_eq!(runtime.errors.len(), 1); + let incident_id = runtime.errors[0].id.clone(); + assert_eq!(runtime.errors[0].status, santi_core::IncidentStatus::Active); + assert_eq!(runtime.errors[0].occurrence_count, 1); + assert_eq!(pending_count(&conn, &strand.id), 1); + + let rejected = service + .send_strand( + &strand.id, + SendStrandRequest { + content: vec![MessagePart::Text { + text: "must not enter inbox".to_string(), + }], + }, + ) + .await + .expect_err("active drive incident should quick-fail later writes"); + assert_eq!(rejected.code, "runtime.strand.drive_failed"); + assert_eq!(rejected.incident_id.as_deref(), Some(incident_id.as_str())); + assert_eq!(rejected.context["accepted_before_failure"], false); + assert_eq!(pending_count(&conn, &strand.id), 1); + + conn.execute_batch("DROP TRIGGER force_turn_insert_failure;") + .expect("remove failure trigger"); + let driven = service.drive_strand(&strand.id).expect("operator redrive"); + assert_eq!(driven.state, santi_core::DriveStrandState::Started); + let turn = driven.turn.expect("redrive turn"); + let runtime = wait_for_completed_turn(&service, &strand.id, &turn.id).await; + assert!(!service.is_drive_degraded()); + assert_eq!(pending_count(&conn, &strand.id), 0); + assert_eq!(runtime.errors.len(), 1); + assert_eq!(runtime.errors[0].id, incident_id); + assert_eq!( + runtime.errors[0].status, + santi_core::IncidentStatus::Resolved + ); + assert_eq!(runtime.errors[0].occurrence_count, 2); + assert_eq!(runtime.errors[0].revision, 2); + assert_eq!( + runtime.errors[0].resolved_by.as_deref(), + Some("strand.drive_started") + ); + assert!( + runtime + .messages + .iter() + .any(|message| message.content_text == "accepted before drive failure") + ); + assert!( + runtime + .messages + .iter() + .all(|message| message.content_text != "must not enter inbox") + ); + let transitions: i64 = conn + .query_row("SELECT COUNT(*) FROM error_transitions", [], |row| { + row.get(0) + }) + .expect("transition count"); + assert_eq!(transitions, 2); +} + +#[tokio::test] +async fn cold_start_recovers() { + let temp = tempfile::tempdir().expect("temp dir"); + let database_path = temp.path().join("santi.sqlite"); + let config = SantiServiceConfig { + database_path: database_path.display().to_string(), + runtime_root: temp.path().join("runtime").display().to_string(), + execution_root: temp.path().join("execution").display().to_string(), + bind_addr: Some("127.0.0.1:0".to_string()), + }; + let service = SantiService::open(config.clone(), Arc::new(FakeProvider::default())) + .expect("open service"); + let strand = service.create_strand().expect("create strand").strand; + service.begin_shutdown(); + let accepted = service + .send_strand( + &strand.id, + SendStrandRequest { + content: vec![MessagePart::Text { + text: "pending across restart".to_string(), + }], + }, + ) + .await + .expect("shutdown should retain the durable request"); + assert!(accepted.receipt.warning.is_none()); + assert!(accepted.turn.is_none()); + drop(service); + + let conn = Connection::open(&database_path).expect("open sqlite"); + conn.execute_batch( + r#" + CREATE TRIGGER force_cold_start_turn_failure + BEFORE INSERT ON turns + BEGIN + SELECT RAISE(ABORT, 'forced cold-start turn failure'); + END; + "#, + ) + .expect("install failure trigger"); + + let restarted = SantiService::open(config, Arc::new(FakeProvider::default())) + .expect("open restarted service"); + restarted + .resume_pending() + .expect("strand-local drive failure should permit degraded startup"); + assert!(restarted.is_drive_degraded()); + assert_eq!(pending_count(&conn, &strand.id), 1); + let runtime = restarted + .runtime_snapshot(&strand.id) + .expect("runtime") + .expect("strand"); + assert_eq!(runtime.errors.len(), 1); + assert_eq!(runtime.errors[0].code, "runtime.strand.drive_failed"); + assert_eq!(runtime.errors[0].context["accepted_before_failure"], false); + assert_eq!(runtime.errors[0].context["pending_count"], 1); + assert_eq!(runtime.errors[0].source.operation, "cold_start_resume"); + + let rejected = restarted + .send_strand( + &strand.id, + SendStrandRequest { + content: vec![MessagePart::Text { + text: "must still quick-fail".to_string(), + }], + }, + ) + .await + .expect_err("active cold-start incident should gate writes"); + assert_eq!(rejected.code, "runtime.strand.drive_failed"); + assert_eq!(pending_count(&conn, &strand.id), 1); + + conn.execute_batch("DROP TRIGGER force_cold_start_turn_failure;") + .expect("remove failure trigger"); + let driven = restarted + .drive_strand(&strand.id) + .expect("operator redrive"); + let turn = driven.turn.expect("redrive turn"); + let runtime = wait_for_completed_turn(&restarted, &strand.id, &turn.id).await; + assert!(!restarted.is_drive_degraded()); + assert_eq!(pending_count(&conn, &strand.id), 0); + assert_eq!( + runtime.errors[0].status, + santi_core::IncidentStatus::Resolved + ); + assert!( + runtime + .messages + .iter() + .any(|message| message.content_text == "pending across restart") + ); +} + +fn pending_count(conn: &Connection, strand_id: &str) -> i64 { + conn.query_row( + "SELECT COUNT(*) FROM strand_inbox WHERE strand_id = ?1", + [strand_id], + |row| row.get(0), + ) + .expect("pending count") +} diff --git a/crates/santi/src/client.rs b/crates/santi/src/client.rs index 65875fc..c8003af 100644 --- a/crates/santi/src/client.rs +++ b/crates/santi/src/client.rs @@ -8,7 +8,12 @@ use crate::cli::{ ClientDefaults, Command, CompactCommand, ImCommand, StrandCommand, WatchFormat, split_send_args, }; use crate::text_source::read_summary_file; -use crate::watch::{next_sse_frame, render_watch_event, watch_until_idle}; +use crate::watch::{next_sse_frame, render_watch_event}; + +mod send; + +pub use send::send; +use send::{accepted_warning, accepted_warning_error}; const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); @@ -444,86 +449,3 @@ async fn im_inbox_high_water( .max() .unwrap_or(0)) } - -/// POST a send, then optionally `--watch` the stream until the strand is -/// idle again. Without `--watch` this is the prior fire-and-return behavior. -pub async fn send( - client: &reqwest::Client, - base: &str, - strand_id: &str, - body: serde_json::Value, - watch: bool, - watch_format: WatchFormat, -) -> Result<()> { - let url = format!("{base}/api/v1/strands/{strand_id}/send"); - // The send itself returns as soon as the message is enqueued (it does not - // wait for the turn), so it is an immediate request and gets the timeout; - // the optional `--watch` below follows the SSE stream, which does not. - let response = client - .post(&url) - .timeout(REQUEST_TIMEOUT) - .json(&body) - .send() - .await - .with_context(|| format!("POST {url}"))?; - let status = response.status(); - let text = response.text().await.context("read response body")?; - let accepted = serde_json::from_str::(&text).ok(); - if !watch { - match &accepted { - Some(value) => println!("{}", serde_json::to_string_pretty(value)?), - None => println!("{text}"), - } - } - if !status.is_success() { - if watch { - println!("{text}"); - } - anyhow::bail!("request failed with status {status}"); - } - if let Some(warning) = accepted_warning(accepted.as_ref()) { - if watch { - match &accepted { - Some(value) => println!("{}", serde_json::to_string_pretty(value)?), - None => println!("{text}"), - } - } - return Err(accepted_warning_error(warning)); - } - if !watch { - return Ok(()); - } - // Seed the in-flight set with the turn this send landed on (a fresh turn, or - // the running one it coalesced into), so a follow-on that handles our message - // is still awaited even if its `turn_started` arrives after the seed's end. - let seed_turn = accepted - .as_ref() - .and_then(|value| value.get("turn")) - .and_then(|turn| turn.get("id")) - .and_then(serde_json::Value::as_str) - .map(str::to_string); - watch_until_idle(client, base, strand_id, seed_turn, watch_format).await -} - -fn accepted_warning(value: Option<&serde_json::Value>) -> Option<&serde_json::Value> { - value? - .pointer("/receipt/warning") - .filter(|warning| !warning.is_null()) -} - -fn accepted_warning_error(warning: &serde_json::Value) -> anyhow::Error { - if let Some(command) = warning - .pointer("/context/recovery/command") - .and_then(serde_json::Value::as_str) - { - anyhow::anyhow!("message was accepted but not driven; do not resend it; run `{command}`") - } else { - let code = warning - .get("code") - .and_then(serde_json::Value::as_str) - .unwrap_or("unknown warning"); - anyhow::anyhow!( - "message was accepted but not driven; do not resend it; inspect and resolve `{code}`" - ) - } -} diff --git a/crates/santi/src/client/send.rs b/crates/santi/src/client/send.rs new file mode 100644 index 0000000..06f04c9 --- /dev/null +++ b/crates/santi/src/client/send.rs @@ -0,0 +1,83 @@ +use anyhow::{Context, Result}; + +use super::REQUEST_TIMEOUT; +use crate::cli::WatchFormat; +use crate::watch::watch_until_idle; + +/// POST a send, then optionally watch the stream until the strand is idle. +pub async fn send( + client: &reqwest::Client, + base: &str, + strand_id: &str, + body: serde_json::Value, + watch: bool, + watch_format: WatchFormat, +) -> Result<()> { + let url = format!("{base}/api/v1/strands/{strand_id}/send"); + let response = client + .post(&url) + .timeout(REQUEST_TIMEOUT) + .json(&body) + .send() + .await + .with_context(|| format!("POST {url}"))?; + let status = response.status(); + let text = response.text().await.context("read response body")?; + let accepted = serde_json::from_str::(&text).ok(); + if !watch { + print_response(accepted.as_ref(), &text)?; + } + if !status.is_success() { + if watch { + println!("{text}"); + } + anyhow::bail!("request failed with status {status}"); + } + if let Some(warning) = accepted_warning(accepted.as_ref()) { + if watch { + print_response(accepted.as_ref(), &text)?; + } + return Err(accepted_warning_error(warning)); + } + if !watch { + return Ok(()); + } + let seed_turn = accepted + .as_ref() + .and_then(|value| value.get("turn")) + .and_then(|turn| turn.get("id")) + .and_then(serde_json::Value::as_str) + .map(str::to_string); + watch_until_idle(client, base, strand_id, seed_turn, watch_format).await +} + +pub(super) fn accepted_warning(value: Option<&serde_json::Value>) -> Option<&serde_json::Value> { + value? + .pointer("/receipt/warning") + .filter(|warning| !warning.is_null()) +} + +pub(super) fn accepted_warning_error(warning: &serde_json::Value) -> anyhow::Error { + if let Some(command) = warning + .pointer("/context/recovery/command") + .and_then(serde_json::Value::as_str) + { + anyhow::anyhow!("message was accepted but not driven; do not resend it; run `{command}`") + } else { + let code = warning + .get("code") + .and_then(serde_json::Value::as_str) + .unwrap_or("unknown warning"); + anyhow::anyhow!( + "message was accepted but not driven; do not resend it; inspect and resolve `{code}`" + ) + } +} + +fn print_response(accepted: Option<&serde_json::Value>, text: &str) -> Result<()> { + match accepted { + Some(value) => println!("{}", serde_json::to_string_pretty(value)?), + None => println!("{text}"), + } + Ok(()) +} diff --git a/crates/santi/tests/send_watch.rs b/crates/santi/tests/send_watch.rs index 881be05..a312e55 100644 --- a/crates/santi/tests/send_watch.rs +++ b/crates/santi/tests/send_watch.rs @@ -194,7 +194,7 @@ async fn posts_once_watch_error() { } #[tokio::test] -async fn accepted_warning_does_not_watch_or_resend() { +async fn warning_stops_watch() { let server = spawn_server(FakeEventsResponse::AcceptedWarning).await; let client = reqwest::Client::new();