diff --git a/crates/nuncio-cli/src/runner.rs b/crates/nuncio-cli/src/runner.rs index a74ef3a..65571d5 100644 --- a/crates/nuncio-cli/src/runner.rs +++ b/crates/nuncio-cli/src/runner.rs @@ -2786,17 +2786,80 @@ impl HeadlessRunner { /// NOT auto-spawn the daemon. async fn handle_system_status(&self, json_mode: bool) -> String { match self.query_daemon_status().await { - Ok((engine_status, version)) => { + Ok(status) => { + let uptime_secs = status + .uptime + .as_ref() + .map(nuncio_proto::time::duration_to_secs) + .unwrap_or(0); if json_mode { + let accounts: Vec<_> = status + .account_sync_states + .iter() + .map(|a| { + json!({ + "account_id": a.account_id, + "state": sync_state_label(a.state), + "last_synced": a + .last_synced + .as_ref() + .map(nuncio_proto::time::timestamp_to_unix_secs), + "last_error": a.last_error, + }) + }) + .collect(); format_json(&json!({ - "engine_status": engine_status, - "version": version, + "engine_status": status.engine_status, + "version": status.version, + "ready": status.ready, + "db_healthy": status.db_healthy, + "uptime_secs": uptime_secs, + "accounts_loaded": status.accounts_loaded, + "unread_count": status.unread_count, + "outbox_depth": status.outbox_depth, + "last_error": status.last_error, + "accounts": accounts, })) } else { - format!( - "Nuncio daemon status: {engine_status} (nunciod v{version}, {})", - self.grpc_addr - ) + let mut out = format!( + "Nuncio daemon status: {} (nunciod v{}, {})\n", + status.engine_status, status.version, self.grpc_addr + ); + out.push_str(&format!( + " ready: {} db_healthy: {}\n", + status.ready, status.db_healthy + )); + out.push_str(&format!(" uptime: {uptime_secs}s\n")); + out.push_str(&format!( + " accounts_loaded: {} unread: {} outbox_depth: {}\n", + status.accounts_loaded, status.unread_count, status.outbox_depth + )); + out.push_str(&format!( + " last_error: {}", + status.last_error.as_deref().unwrap_or("") + )); + if status.account_sync_states.is_empty() { + out.push_str("\n accounts: "); + } else { + out.push_str("\n accounts:"); + for a in &status.account_sync_states { + out.push_str(&format!( + "\n {}: {}", + a.account_id, + sync_state_label(a.state) + )); + if let Some(ts) = a.last_synced.as_ref() { + out.push_str(&format!( + " (last synced: {})", + nuncio_proto::time::timestamp_to_unix_secs(ts) + )); + } + if let Some(err) = a.last_error.as_deref() { + out.push_str(&format!(" [error: {err}]")); + } + } + } + out } } Err(e) => { @@ -2810,10 +2873,10 @@ impl HeadlessRunner { } /// Reads the gRPC bearer token from the injected vault, connects to the - /// daemon over gRPC, and calls `GetStatus`. Returns `(engine_status, - /// version)` on success, or a human-readable error string describing + /// daemon over gRPC, and calls `GetStatus`. Returns the full live health + /// response on success, or a human-readable error string describing /// exactly what failed (vault, connection, or the RPC itself). - async fn query_daemon_status(&self) -> Result<(String, String), String> { + async fn query_daemon_status(&self) -> Result { let token_bytes = self .secrets .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) @@ -2835,7 +2898,21 @@ impl HeadlessRunner { })? .into_inner(); - Ok((response.engine_status, response.version)) + Ok(response) + } +} + +/// Maps a `nuncio.v1.SyncState` wire discriminant onto a short human label +/// for the CLI's `system status` rendering. An unrecognized value renders as +/// "unknown" rather than failing. +fn sync_state_label(state: i32) -> &'static str { + match nuncio_proto::v1::SyncState::try_from(state) + .unwrap_or(nuncio_proto::v1::SyncState::Unspecified) + { + nuncio_proto::v1::SyncState::Idle => "idle", + nuncio_proto::v1::SyncState::Syncing => "syncing", + nuncio_proto::v1::SyncState::Error => "error", + nuncio_proto::v1::SyncState::Unspecified => "unknown", } } @@ -3056,6 +3133,21 @@ mod tests { Ok(tonic::Response::new(GetStatusResponse { engine_status: "Ready".to_string(), version: "9.9.9".to_string(), + uptime: Some(nuncio_proto::time::duration_from_secs(5)), + accounts_loaded: 2, + unread_count: 3, + last_error: None, + outbox_depth: 1, + account_sync_states: vec![nuncio_proto::v1::AccountSyncState { + account_id: "acct-1".to_string(), + state: nuncio_proto::v1::SyncState::Idle as i32, + last_synced: Some(nuncio_proto::time::timestamp_from_unix_secs( + 1_700_000_000, + )), + last_error: None, + }], + ready: true, + db_healthy: true, })) } @@ -3106,6 +3198,12 @@ mod tests { assert!(out.contains(r#""status":"ok""#)); assert!(out.contains(r#""engine_status":"Ready""#)); assert!(out.contains(r#""version":"9.9.9""#)); + assert!(out.contains(r#""ready":true"#)); + assert!(out.contains(r#""db_healthy":true"#)); + assert!(out.contains(r#""accounts_loaded":2"#)); + assert!(out.contains(r#""unread_count":3"#)); + assert!(out.contains(r#""outbox_depth":1"#)); + assert!(out.contains(r#""account_id":"acct-1""#)); let text_out = runner .execute_command( diff --git a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin index f86c117..d555779 100644 Binary files a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin and b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin differ diff --git a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto index 0100fcd..41302bf 100644 --- a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto +++ b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto @@ -89,13 +89,64 @@ service System { // here later (e.g. requested detail level). message GetStatusRequest {} -// GetStatusResponse reports the current health and build identity of the -// running nunciod engine. +// Per-account synchronization phase, reported inside `AccountSyncState`. +// Values are append-only: never renumber or repurpose an existing value. +enum SyncState { + // Zero value; treated as IDLE by clients that do not recognize a value. + SYNC_STATE_UNSPECIFIED = 0; + // No sync is in flight for the account and its last attempt (if any) + // succeeded. + SYNC_STATE_IDLE = 1; + // A sync for the account is currently running. + SYNC_STATE_SYNCING = 2; + // No sync is in flight, but the account's most recent attempt failed; + // `last_error` carries the genuine failure detail. + SYNC_STATE_ERROR = 3; +} + +// AccountSyncState reports the live sync phase of a single configured +// account, plus the outcome of its most recent completed sync attempt. It is +// derived from the daemon's in-flight sync map and last-result tracking -- +// never fabricated. An account that has never synced since daemon start +// reports `IDLE` with no `last_synced`/`last_error`. +message AccountSyncState { + // The account this state describes. + string account_id = 1; + // Current sync phase. + SyncState state = 2; + // When the account last synced SUCCESSFULLY, if it has since daemon start. + // Absent when the account has not completed a successful sync yet. + optional google.protobuf.Timestamp last_synced = 3; + // Present only when the most recent attempt failed (`state == ERROR`): + // the genuine failure detail. + optional string last_error = 4; +} + +// GetStatusResponse reports the current health, readiness, and build identity +// of the running nunciod engine. Every numeric/health field is backed by live +// daemon state read when the RPC is served -- never a placeholder. message GetStatusResponse { - // Human-readable engine status (e.g. "ok", "degraded"). + // Human-readable engine status (e.g. "Idle", "Syncing", "ShuttingDown"). string engine_status = 1; // The nunciod build/crate version serving this response. string version = 2; + // How long the daemon process has been serving since startup. + google.protobuf.Duration uptime = 3; + // Number of configured accounts currently loaded from the store. + uint64 accounts_loaded = 4; + // Total unread messages across every synced folder. + uint64 unread_count = 5; + // Last engine-level error recorded on the event bus, if any. + optional string last_error = 6; + // Number of pending outbound mutations queued in the outbox. + uint64 outbox_depth = 7; + // Live per-account sync state, one entry per configured account. + repeated AccountSyncState account_sync_states = 8; + // Overall readiness: true when the daemon is serving normally (database + // healthy and not shutting down). A client can gate work on this. + bool ready = 9; + // Whether a cheap `SELECT 1` database liveness probe just succeeded. + bool db_healthy = 10; } // SubscribeRequest currently carries no parameters. Fields may be added diff --git a/crates/nuncio-proto/src/lib.rs b/crates/nuncio-proto/src/lib.rs index 4ccefe7..8720315 100644 --- a/crates/nuncio-proto/src/lib.rs +++ b/crates/nuncio-proto/src/lib.rs @@ -56,10 +56,23 @@ mod tests { let response = GetStatusResponse { engine_status: "ok".to_string(), version: "0.1.0".to_string(), + uptime: Some(super::time::duration_from_secs(42)), + accounts_loaded: 2, + unread_count: 7, + last_error: None, + outbox_depth: 1, + account_sync_states: Vec::new(), + ready: true, + db_healthy: true, }; assert_eq!(response.engine_status, "ok"); assert_eq!(response.version, "0.1.0"); + assert_eq!(response.accounts_loaded, 2); + assert_eq!(response.unread_count, 7); + assert_eq!(response.outbox_depth, 1); + assert!(response.ready); + assert!(response.db_healthy); } #[test] diff --git a/crates/nuncio-proto/src/time.rs b/crates/nuncio-proto/src/time.rs index 6702ddd..380499f 100644 --- a/crates/nuncio-proto/src/time.rs +++ b/crates/nuncio-proto/src/time.rs @@ -62,6 +62,17 @@ pub fn duration_from_secs(secs: u64) -> Duration { } } +/// Converts a `std::time::Duration` (e.g. a process-uptime measurement taken +/// with `Instant::elapsed`) into a wire `Duration`, preserving sub-second +/// precision. Saturates rather than overflowing on an implausibly large span. +#[must_use] +pub fn duration_from_std(d: std::time::Duration) -> Duration { + Duration { + seconds: i64::try_from(d.as_secs()).unwrap_or(i64::MAX), + nanos: d.subsec_nanos() as i32, + } +} + /// Extracts a whole-second duration from a `Duration`, truncating any /// sub-second `nanos` component. A negative `seconds` (never produced by /// this crate's own mappers) reads back as zero rather than panicking. diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index a7095d8..c22ba55 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -438,6 +438,39 @@ impl DatabaseEngine { } } + /// Cheap liveness probe: runs `SELECT 1` against the pool and returns + /// `Ok(())` only if it genuinely round-trips. Unlike `check_integrity` + /// (a full `PRAGMA quick_check`), this is a constant-time readiness ping + /// suitable for a health endpoint served on every status request. + pub async fn ping(&self) -> Result<(), DatabaseError> { + let _: (i64,) = sqlx::query_as("SELECT 1") + .fetch_one(&self.pool) + .await + .map_err(DatabaseError::Query)?; + Ok(()) + } + + /// Total unread messages across every folder, read live from the store. + pub async fn count_unread_messages(&self) -> Result { + let (count,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM messages WHERE read_flag = 0") + .fetch_one(&self.pool) + .await + .map_err(DatabaseError::Query)?; + Ok(count.max(0) as u64) + } + + /// Number of outbound mutations still queued in the outbox (status + /// `pending`) -- the outbox depth reported by the daemon's health surface. + pub async fn count_pending_mutations(&self) -> Result { + let (count,): (i64,) = sqlx::query_as( + "SELECT COUNT(*) FROM pending_remote_mutations WHERE status = 'pending'", + ) + .fetch_one(&self.pool) + .await + .map_err(DatabaseError::Query)?; + Ok(count.max(0) as u64) + } + /// Cryptographic hash-chain audit ledger verification (`verify_chain_integrity()`) /// detecting log tampering or corrupted `filter_execution_logs`, using the ledger /// HMAC key provisioned for this engine. diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 00659cf..8d387bf 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -7,10 +7,12 @@ use crate::lifecycle::ShutdownSignal; use crate::pagination::{self, CursorField}; -use crate::sync_dispatcher::{sync_all_configured, ProductionAccountSyncer, SyncDispatcher}; +use crate::sync_dispatcher::{ + sync_all_configured, AccountSyncPhase, ProductionAccountSyncer, SyncDispatcher, +}; use nuncio_cal::CalendarBackend; use nuncio_contacts::ContactsBackend; -use nuncio_core::{CoreCommand, CoreEvent, EventBus}; +use nuncio_core::{CoreCommand, CoreEvent, EngineStatus, EventBus}; use nuncio_filter::{FilterEngine, NsqlParser, NsqlValidator, ValidationOptions}; use nuncio_mail::{ImapEngine, MailBackend, MessageSender, SmtpTransportEngine}; use nuncio_proto::errors; @@ -26,14 +28,15 @@ use nuncio_proto::v1::mail_server::{Mail, MailServer}; use nuncio_proto::v1::system_server::{System, SystemServer}; use nuncio_proto::v1::ErrorReason; use nuncio_proto::v1::{ - export_request, AccountConfig as AccountConfigProto, AddAccountRequest, AddAccountResponse, - Attachment as AttachmentProto, AuditRecord as AuditRecordProto, BatchFilterProgress, - CalendarEvent as CalendarEventProto, CalendarSyncRequest, CalendarSyncResponse, - Contact as ContactProto, ContactEmail as ContactEmailProto, ContactPhone as ContactPhoneProto, - ContactsSyncRequest, ContactsSyncResponse, CreateContactRequest, CreateContactResponse, - CreateRuleRequest, CreateRuleResponse, DatabaseRecovered, DavTransport as DavTransportProto, - DeleteRuleRequest, DeleteRuleResponse, Event, EventError, ExportFormat as ExportFormatProto, - ExportRequest, ExportResponse, ExportRulesRequest, ExportRulesResponse, FilterExecuted, + export_request, AccountConfig as AccountConfigProto, AccountSyncState as AccountSyncStateProto, + AddAccountRequest, AddAccountResponse, Attachment as AttachmentProto, + AuditRecord as AuditRecordProto, BatchFilterProgress, CalendarEvent as CalendarEventProto, + CalendarSyncRequest, CalendarSyncResponse, Contact as ContactProto, + ContactEmail as ContactEmailProto, ContactPhone as ContactPhoneProto, ContactsSyncRequest, + ContactsSyncResponse, CreateContactRequest, CreateContactResponse, CreateRuleRequest, + CreateRuleResponse, DatabaseRecovered, DavTransport as DavTransportProto, DeleteRuleRequest, + DeleteRuleResponse, Event, EventError, ExportFormat as ExportFormatProto, ExportRequest, + ExportResponse, ExportRulesRequest, ExportRulesResponse, FilterExecuted, FilterExecutionLog as FilterExecutionLogProto, FilterRule as FilterRuleProto, Folder as FolderProto, GetContactRequest, GetContactResponse, GetEventRequest, GetEventResponse, GetExecutionLogsRequest, GetExecutionLogsResponse, GetMessageRequest, @@ -47,18 +50,18 @@ use nuncio_proto::v1::{ PreviewRuleRequest, PreviewRuleResponse, RemoveAccountRequest, RemoveAccountResponse, RuleExportFormat as RuleExportFormatProto, SearchMessagesRequest, SearchMessagesResponse, SendMessageRequest, SendMessageResponse, ShuttingDown, SubscribeRequest, SyncCompleted, - SyncRequest, SyncResponse, SyncStarted, TestAccountConnectionRequest, - TestAccountConnectionResponse, TlsMode as TlsModeProto, TriageProgress, TriageRequest, - UpdateAccountRequest, UpdateAccountResponse, UpdateAvailable, UpdateRuleRequest, - UpdateRuleResponse, ValidateRuleRequest, ValidateRuleResponse, VerifyChainRequest, - VerifyChainResponse, + SyncRequest, SyncResponse, SyncStarted, SyncState as SyncStateProto, + TestAccountConnectionRequest, TestAccountConnectionResponse, TlsMode as TlsModeProto, + TriageProgress, TriageRequest, UpdateAccountRequest, UpdateAccountResponse, UpdateAvailable, + UpdateRuleRequest, UpdateRuleResponse, ValidateRuleRequest, ValidateRuleResponse, + VerifyChainRequest, VerifyChainResponse, }; use nuncio_store::db::DatabaseEngine; use nuncio_store::search::SearchEngine; use nuncio_store::vault::SecretManager; use std::pin::Pin; use std::sync::Arc; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use subtle::ConstantTimeEq; use thiserror::Error; use tokio::net::TcpListener; @@ -216,10 +219,36 @@ fn map_core_event(event: CoreEvent) -> Option { Some(Event { kind: Some(kind) }) } +/// Maps a dispatcher [`AccountSyncPhase`] onto its wire `nuncio.v1.SyncState`. +fn map_sync_phase_to_proto(phase: AccountSyncPhase) -> SyncStateProto { + match phase { + AccountSyncPhase::Idle => SyncStateProto::Idle, + AccountSyncPhase::Syncing => SyncStateProto::Syncing, + AccountSyncPhase::Error => SyncStateProto::Error, + } +} + +/// Converts a wall-clock [`SystemTime`] (an account's last successful sync +/// instant) into a wire `Timestamp` at second granularity. A pre-epoch time +/// (never produced here) maps to `None` rather than a negative instant. +fn system_time_to_timestamp(t: SystemTime) -> Option { + t.duration_since(UNIX_EPOCH) + .ok() + .and_then(|d| i64::try_from(d.as_secs()).ok()) + .map(nuncio_proto::time::timestamp_from_unix_secs) +} + /// `nuncio.v1.System` gRPC service implementation backed by the daemon's live -/// [`EventBus`] state. +/// [`EventBus`] state, message store, and sync dispatcher. struct SystemGrpcService { event_bus: Arc, + db: Arc, + sync_dispatcher: Arc, + /// Captured once when the service is constructed at daemon startup; + /// `GetStatus` reports the elapsed time since as process uptime. An + /// `Instant` (monotonic) rather than a wall clock, so a system time + /// adjustment can never make uptime jump or go negative. + started_at: Instant, } #[tonic::async_trait] @@ -229,9 +258,68 @@ impl System for SystemGrpcService { _request: Request, ) -> Result, Status> { let state = self.event_bus.current_state(); + + // Cheap `SELECT 1` liveness probe; the sole source of `db_healthy`. + let db_healthy = self.db.ping().await.is_ok(); + + // Enumerate configured accounts live from the store: their count is + // `accounts_loaded`, and each one's live sync phase comes from the + // shared dispatcher's in-flight/last-result tracking. A read failure + // degrades to an empty set (already reflected by `db_healthy`) rather + // than fabricating a count. + let accounts = match self.db.list_accounts().await { + Ok(accounts) => accounts, + Err(e) => { + tracing::warn!("GetStatus: failed to list accounts for health surface: {e}"); + Vec::new() + } + }; + let accounts_loaded = accounts.len() as u64; + let account_sync_states = accounts + .iter() + .map(|account| { + let status = self.sync_dispatcher.sync_status(&account.id); + AccountSyncStateProto { + account_id: account.id.clone(), + state: map_sync_phase_to_proto(status.phase) as i32, + last_synced: status.last_success_at.and_then(system_time_to_timestamp), + last_error: status.last_error, + } + }) + .collect(); + + let unread_count = match self.db.count_unread_messages().await { + Ok(count) => count, + Err(e) => { + tracing::warn!("GetStatus: failed to count unread messages: {e}"); + 0 + } + }; + let outbox_depth = match self.db.count_pending_mutations().await { + Ok(depth) => depth, + Err(e) => { + tracing::warn!("GetStatus: failed to count pending outbox mutations: {e}"); + 0 + } + }; + + // Ready to serve normally: the database answered its liveness probe + // and the engine is not tearing down. + let ready = db_healthy && state.status != EngineStatus::ShuttingDown; + Ok(Response::new(GetStatusResponse { engine_status: format!("{:?}", state.status), version: env!("CARGO_PKG_VERSION").to_string(), + uptime: Some(nuncio_proto::time::duration_from_std( + self.started_at.elapsed(), + )), + accounts_loaded, + unread_count, + last_error: state.last_error, + outbox_depth, + account_sync_states, + ready, + db_healthy, })) } @@ -2890,6 +2978,9 @@ pub async fn serve_on_listener_with_overrides_and_shutdown( let system_service = SystemGrpcService { event_bus: event_bus.clone(), + db: db.clone(), + sync_dispatcher: sync_dispatcher.clone(), + started_at: Instant::now(), }; let system_interceptor = BearerAuthInterceptor::new(token.clone()); let system_svc = InterceptedService::new( @@ -3521,6 +3612,91 @@ mod tests { .into_inner(); assert_eq!(response.engine_status, "Syncing"); assert_eq!(response.version, env!("CARGO_PKG_VERSION")); + // Live health surface against a fresh in-memory daemon: the database + // answered its liveness probe, uptime is genuinely elapsing, and with + // no accounts configured the counts are honestly zero. + assert!(response.db_healthy, "ephemeral db must answer SELECT 1"); + assert!(response.ready, "healthy, non-shutting-down daemon is ready"); + let uptime = response.uptime.expect("uptime is always populated"); + assert!( + uptime.seconds > 0 || uptime.nanos > 0, + "uptime must be strictly positive, got {uptime:?}" + ); + assert_eq!(response.accounts_loaded, 0); + assert_eq!(response.unread_count, 0); + assert_eq!(response.outbox_depth, 0); + assert!(response.account_sync_states.is_empty()); + } + + #[tokio::test] + async fn get_status_reports_live_accounts_and_outbox_depth() { + // Back the health numbers with real persisted state: two configured + // accounts and one queued outbox mutation must be reflected exactly. + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("connect ephemeral test db"); + let db = Arc::new(db); + + for id in ["acct-h1", "acct-h2"] { + let account = nuncio_core::AccountConfig { + id: id.to_string(), + name: "Health".to_string(), + email_address: format!("{id}@nuncio.mx"), + keyring_secret_key: format!("nuncio/{id}"), + sync_interval_secs: 60, + transport: nuncio_core::Transport::Jmap(nuncio_core::JmapTransport { + endpoint_host: "jmap.nuncio.mx".to_string(), + }), + }; + db.save_account(&account).await.expect("save account"); + } + + db.save_pending_mutation(&nuncio_filter::PendingRemoteMutation { + id: "mut-1".to_string(), + rule_id: "rule-1".to_string(), + message_id: "msg-1".to_string(), + mutation_type: "MOVE".to_string(), + payload: "{}".to_string(), + status: "pending".to_string(), + retry_count: 0, + created_at: 0, + }) + .await + .expect("queue an outbox mutation"); + + let event_bus = Arc::new(EventBus::new()); + let secrets = Arc::new(SecretManager::mock()); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + let (addr, _handle) = + spawn_test_server_with(event_bus, db, filter_engine, secrets, "correct-token").await; + + let mut client = SystemClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + let mut request = Request::new(GetStatusRequest {}); + request.metadata_mut().insert( + "authorization", + "Bearer correct-token" + .parse() + .expect("valid ascii metadata value"), + ); + let response = client + .get_status(request) + .await + .expect("valid token is accepted") + .into_inner(); + + assert!(response.db_healthy); + assert_eq!(response.accounts_loaded, 2); + assert_eq!(response.outbox_depth, 1); + // Each configured account has a per-account sync state; with no sync + // run yet every one is IDLE with no last-synced/last-error. + assert_eq!(response.account_sync_states.len(), 2); + for state in &response.account_sync_states { + assert_eq!(state.state, SyncStateProto::Idle as i32); + assert!(state.last_synced.is_none()); + assert!(state.last_error.is_none()); + } } #[tokio::test] diff --git a/crates/nunciod/src/sync_dispatcher.rs b/crates/nunciod/src/sync_dispatcher.rs index 8e46819..4526da0 100644 --- a/crates/nunciod/src/sync_dispatcher.rs +++ b/crates/nunciod/src/sync_dispatcher.rs @@ -45,7 +45,7 @@ use nuncio_store::db::DatabaseEngine; use nuncio_store::vault::SecretManager; use std::collections::HashMap; use std::sync::{Arc, Mutex, PoisonError}; -use std::time::Duration; +use std::time::{Duration, SystemTime}; use thiserror::Error; use tokio::sync::{watch, Semaphore}; use tracing::Instrument; @@ -96,6 +96,42 @@ pub enum SyncDispatchError { Sync(Arc), } +/// Live sync phase of a single account, derived from the dispatcher's +/// in-flight map and last-result tracking for the daemon health surface. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AccountSyncPhase { + /// No sync in flight; the last attempt (if any) succeeded. + Idle, + /// A sync for the account is currently running. + Syncing, + /// No sync in flight, but the most recent attempt failed. + Error, +} + +/// A snapshot of one account's sync state, combining its current phase with +/// the outcome of its most recent completed attempt. `last_success_at` +/// survives a subsequent failure so a transient error does not erase the +/// last known-good sync time. +#[derive(Debug, Clone)] +pub struct AccountSyncStatus { + /// Current phase. + pub phase: AccountSyncPhase, + /// When the account last completed a SUCCESSFUL sync, if ever since start. + pub last_success_at: Option, + /// The most recent attempt's failure detail, present only when the last + /// attempt failed. + pub last_error: Option, +} + +/// The recorded outcome of an account's most recent completed sync attempt. +/// A success clears `last_error` and stamps `last_success_at`; a failure sets +/// `last_error` while leaving any earlier `last_success_at` intact. +#[derive(Default, Clone)] +struct AccountSyncRecord { + last_success_at: Option, + last_error: Option, +} + /// Injectable seam for the actual work of syncing one account. Production /// uses [`ProductionAccountSyncer`]; tests inject an in-memory mock. #[async_trait] @@ -179,6 +215,10 @@ struct Inner { /// that publishes the sync's outcome once. A second request for the same /// account clones the receiver and joins the in-flight sync. inflight: Mutex>>>, + /// The most recent completed outcome per account, retained after the + /// in-flight entry is cleared so the health surface can report each + /// account's last sync time and last error. + last_results: Mutex>, per_account_timeout: Duration, shutdown: ShutdownSignal, } @@ -246,6 +286,7 @@ impl SyncDispatcher { syncer, semaphore: Arc::new(Semaphore::new(concurrency)), inflight: Mutex::new(HashMap::new()), + last_results: Mutex::new(HashMap::new()), per_account_timeout, shutdown, }), @@ -298,6 +339,25 @@ impl SyncDispatcher { tokio::spawn( async move { let outcome = inner.run_guarded(&id).await; + // Record this attempt's outcome for the health surface + // before clearing the in-flight entry: a success stamps + // the last-synced time and clears any prior error; a + // failure records the error while preserving the last + // known-good sync time. + { + let mut results = inner + .last_results + .lock() + .unwrap_or_else(PoisonError::into_inner); + let record = results.entry(id.clone()).or_default(); + match &outcome { + Ok(_) => { + record.last_success_at = Some(SystemTime::now()); + record.last_error = None; + } + Err(e) => record.last_error = Some(e.to_string()), + } + } // Remove the in-flight entry before publishing so a // request arriving after completion starts a fresh sync // rather than joining an already-finished one. @@ -325,6 +385,41 @@ impl SyncDispatcher { published.unwrap_or(Err(SyncDispatchError::Cancelled)) } + /// Report the live sync state of a single account for the daemon health + /// surface: `Syncing` while a sync is in flight, otherwise `Error` if the + /// most recent attempt failed, else `Idle`. `last_success_at`/`last_error` + /// carry the genuine outcome of the last completed attempt (never + /// fabricated); an account that has not synced since daemon start reports + /// `Idle` with neither set. + pub fn sync_status(&self, account_id: &str) -> AccountSyncStatus { + let syncing = self + .inner + .inflight + .lock() + .unwrap_or_else(PoisonError::into_inner) + .contains_key(account_id); + let record = self + .inner + .last_results + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get(account_id) + .cloned() + .unwrap_or_default(); + let phase = if syncing { + AccountSyncPhase::Syncing + } else if record.last_error.is_some() { + AccountSyncPhase::Error + } else { + AccountSyncPhase::Idle + }; + AccountSyncStatus { + phase, + last_success_at: record.last_success_at, + last_error: record.last_error, + } + } + /// Dispatch syncs for several accounts concurrently, each through the same /// bounded/coalesced/timed per-account path, and return the total number /// of messages synced across the accounts that succeeded. Per-account @@ -598,6 +693,67 @@ mod tests { ); } + #[tokio::test] + async fn sync_status_reports_idle_for_an_unknown_account() { + let syncer = Arc::new(MockSyncer::new()); + let dispatcher = dispatcher_with(syncer, 2, Duration::from_secs(30)); + + let status = dispatcher.sync_status("never-seen"); + assert_eq!(status.phase, AccountSyncPhase::Idle); + assert!(status.last_success_at.is_none()); + assert!(status.last_error.is_none()); + } + + #[tokio::test] + async fn sync_status_reports_syncing_while_a_sync_is_in_flight() { + let syncer = Arc::new(MockSyncer::new().blocking("acct-live")); + let dispatcher = dispatcher_with(syncer.clone(), 2, Duration::from_secs(30)); + + let d = dispatcher.clone(); + let handle = tokio::spawn(async move { d.sync_account("acct-live").await }); + + // Wait until the mock has actually entered the sync (it blocks there). + while syncer.call_count() == 0 { + sleep(Duration::from_millis(5)).await; + } + assert_eq!( + dispatcher.sync_status("acct-live").phase, + AccountSyncPhase::Syncing + ); + + syncer.gate.notify_one(); + let _ = handle.await; + } + + #[tokio::test] + async fn sync_status_reports_idle_with_last_synced_after_success() { + let syncer = Arc::new(MockSyncer::new()); + let dispatcher = dispatcher_with(syncer, 2, Duration::from_secs(30)); + + assert!(matches!(dispatcher.sync_account("acct-ok").await, Ok(1))); + + let status = dispatcher.sync_status("acct-ok"); + assert_eq!(status.phase, AccountSyncPhase::Idle); + assert!(status.last_success_at.is_some()); + assert!(status.last_error.is_none()); + } + + #[tokio::test] + async fn sync_status_reports_error_after_a_failed_attempt() { + let syncer = Arc::new(MockSyncer::new().failing("acct-bad")); + let dispatcher = dispatcher_with(syncer, 2, Duration::from_secs(30)); + + assert!(matches!( + dispatcher.sync_account("acct-bad").await, + Err(SyncDispatchError::Sync(_)) + )); + + let status = dispatcher.sync_status("acct-bad"); + assert_eq!(status.phase, AccountSyncPhase::Error); + assert!(status.last_error.is_some()); + assert!(status.last_success_at.is_none()); + } + #[tokio::test] async fn rejects_new_work_once_shutdown_is_requested() { let event_bus = Arc::new(EventBus::new());