diff --git a/CHANGELOG.md b/CHANGELOG.md index 1fe8d79a1..c25b622b2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,7 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Upgrading -This release adds migrations v044 to v050. Run `awa migrate` as the migrator +This release adds migrations v044 to v051. Run `awa migrate` as the migrator role; the [0.6→0.7 upgrade guide](docs/upgrade-0.6-to-0.7.md) has the details for each step. @@ -73,6 +73,22 @@ for each step. `awa.migration_lock_key()` expose the runner's advisory lock key. - Python prints a diagnostic to stderr if native shutdown has stalled for five seconds. +- **The admin overview answers "is it keeping up?" (v051, + [#543](https://github.com/hardbyte/awa/pull/543)).** The maintenance leader + writes one row per queue per minute to `awa.queue_history` (depth by state, + oldest wait, and claimed/completed/failed/retried/rate-limited counts) and + one fleet row to `awa.fleet_history` (live and healthy instances, leader, + running versions), keeping 48 hours by default. Runtimes publish the + per-queue outcome counters this needs in their existing snapshots. The + overview, Queues page and queue detail classify each queue as not being + worked, falling behind, waiting long, paused, draining, steady or idle, + say what is holding waiting work back (no workers, rate limit, worker + capacity), show measured completion and arrival rates and an hour of + history, and mark rollouts, leader changes and lost instances on the chart. + `GET /api/health/queues` returns the same verdicts for alerting + (`?fail_on=critical` or `warning` answers 503 at that level). + Configure retention with `ClientBuilder::queue_history_retention` or + Python's `queue_history_retention_hours`; zero stops sampling. ### Changed diff --git a/README.md b/README.md index 5c364f6e4..9c03d19ab 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ Awa (Māori: river) fills the gap between Postgres event queues that are too narrow for real job-queue behavior and language-specific job frameworks (River, Oban, Sidekiq) that couple you to one ecosystem. If you run Rust or Python (or both) on Postgres and want priorities, cron, DLQ, and transactional enqueue without Redis or RabbitMQ, Awa is built for you. -![AWA Web UI — Jobs (dark mode)](https://raw.githubusercontent.com/hardbyte/awa/main/docs/images/awa-ui-dark.png) +![Awa admin overview during an incident: one queue is not being worked, another is falling behind its rate limit, with recurring failures and fleet changes alongside](https://raw.githubusercontent.com/hardbyte/awa/main/docs/images/awa-ui-overview-incident-dark.png) ## Core Concepts @@ -49,7 +49,7 @@ Awa (Māori: river) fills the gap between Postgres event queues that are too nar - **Operator descriptors** — code-declared queue and job-kind names/descriptions with stale/drift visibility in the UI. - **Postgres-only** — one dependency you already have; no Redis, no RabbitMQ, no separate scheduler. -![AWA Web UI — Queue detail (dark mode)](https://raw.githubusercontent.com/hardbyte/awa/main/docs/images/awa-ui-queue-detail-dark.png) +![Awa admin overview during recovery: three queues draining their backlog with an estimate of when all will be clear, after a hotfix rollout shown in the fleet changes](https://raw.githubusercontent.com/hardbyte/awa/main/docs/images/awa-ui-overview-recovery-dark.png) ## Correctness diff --git a/awa-model/migrations/v051_queue_history.sql b/awa-model/migrations/v051_queue_history.sql new file mode 100644 index 000000000..909fb944d --- /dev/null +++ b/awa-model/migrations/v051_queue_history.sql @@ -0,0 +1,66 @@ +-- v051: per-minute queue and fleet history for the admin overview +-- +-- Job tables record current state, not how a backlog moved, so the admin UI +-- could not tell a queue that is draining from one falling behind. The +-- maintenance leader now writes one row per queue per minute (depth by state, +-- oldest available wait, and claim/completion/failure deltas differenced from +-- the per-queue counters runtimes publish in their snapshots) plus one fleet +-- row per minute (instance health, leader, running versions). The UI reads +-- these instead of sampling in each `awa serve` process. +-- +-- Both tables are additive and written only by the elected maintenance +-- leader, never by triggers or the claim/finalize path, so ADR-045 does not +-- apply and enqueue/claim cost is unchanged. Rows older than the configured +-- retention (default 48 hours) are deleted by the same leader. +-- +-- N-1 binaries: never read or write these tables, and a 0.7.x leader simply +-- does not populate them; the UI then reports no history. Counter columns are +-- NULL for a minute when any live runtime serving the queue predates the +-- counters, so partial sums are never presented as totals. +-- +-- Locks and time: two CREATE TABLE IF NOT EXISTS on new relations; no +-- existing table is touched. Effectively instant. +-- Runtime roles need SELECT, INSERT and DELETE on both tables; the default +-- privileges in docs/security/database-roles.md grant that. +-- Tracking: https://github.com/hardbyte/awa/pull/543 + +CREATE TABLE IF NOT EXISTS awa.queue_history ( + queue TEXT NOT NULL, + bucket TIMESTAMPTZ NOT NULL, + available BIGINT NOT NULL, + running BIGINT NOT NULL, + scheduled BIGINT NOT NULL, + retryable BIGINT NOT NULL, + failed BIGINT NOT NULL, + waiting_external BIGINT NOT NULL, + lag_seconds DOUBLE PRECISION, + paused BOOLEAN NOT NULL DEFAULT FALSE, + -- Deltas since the previous sample. NULL when unknown: the first sample + -- after a leader change, or a live runtime without counters. + claimed BIGINT, + completed BIGINT, + failed_attempts BIGINT, + retried BIGINT, + rate_limited BIGINT, + claim_wait_ms BIGINT, + -- Live runtimes serving the queue and their summed in-flight count. + live_workers INT NOT NULL DEFAULT 0, + capacity BIGINT, + PRIMARY KEY (queue, bucket) +); + +CREATE INDEX IF NOT EXISTS idx_awa_queue_history_bucket + ON awa.queue_history (bucket); + +CREATE TABLE IF NOT EXISTS awa.fleet_history ( + bucket TIMESTAMPTZ PRIMARY KEY, + live_instances INT NOT NULL, + healthy_instances INT NOT NULL, + stale_instances INT NOT NULL, + leader_instance_id UUID, + versions TEXT[] NOT NULL DEFAULT ARRAY[]::TEXT[] +); + +INSERT INTO awa.schema_version (version, description) +VALUES (51, 'Per-minute queue and fleet history') +ON CONFLICT (version) DO NOTHING; diff --git a/awa-model/src/admin.rs b/awa-model/src/admin.rs index 8eaa2ecd4..745120869 100644 --- a/awa-model/src/admin.rs +++ b/awa-model/src/admin.rs @@ -1691,6 +1691,36 @@ pub struct QueueRuntimeSnapshot { pub in_flight: u32, pub overflow_held: Option, pub config: QueueRuntimeConfigSnapshot, + /// Cumulative outcome counters since this instance started (#543). + /// `None` from runtimes that predate them, so readers can tell "not + /// reported" from zero. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub counters: Option, +} + +/// Monotonic per-queue counters a runtime publishes with each snapshot. +/// +/// They reset when the instance restarts; consumers difference consecutive +/// readings of the same `instance_id` and treat a decrease as a reset. +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct QueueOutcomeCounters { + /// Jobs this instance claimed from the queue. + pub claimed: u64, + pub completed: u64, + /// Attempts that ended the job as failed (terminal). + pub failed: u64, + /// Attempts that failed and were scheduled for another try. + pub retried: u64, + /// Dispatch polls skipped because the queue's rate limit had no tokens. + pub rate_limited: u64, + /// Sum over claimed jobs of the time between becoming runnable and + /// being claimed, in milliseconds. Divide by `claimed` for a mean. + pub claim_wait_ms: u64, + /// Attempts that parked waiting for an external callback. Whatever + /// resolves the callback may run in another process, so the eventual + /// outcome is not counted here. + #[serde(default)] + pub parked: u64, } #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)] diff --git a/awa-model/src/lib.rs b/awa-model/src/lib.rs index d52c6b4b0..bef9ced8e 100644 --- a/awa-model/src/lib.rs +++ b/awa-model/src/lib.rs @@ -13,6 +13,7 @@ pub mod job; pub mod kind; pub mod migrations; pub mod partitioned_queue; +pub mod queue_history; pub mod queue_storage; pub mod reschedule; pub mod storage; diff --git a/awa-model/src/migrations.rs b/awa-model/src/migrations.rs index 6d13a6db5..3f62c71a0 100644 --- a/awa-model/src/migrations.rs +++ b/awa-model/src/migrations.rs @@ -5,7 +5,7 @@ use sqlx::{Connection, PgPool}; use tracing::{info, warn}; /// Current schema version. -pub const CURRENT_VERSION: i32 = 50; +pub const CURRENT_VERSION: i32 = 51; /// Migrations that require an exclusive (no-live-runtime) upgrade window. /// @@ -290,6 +290,11 @@ const MIGRATIONS: &[(i32, &str, &[&str])] = &[ "Drop the redundant lane index on ready_entries partitions", &[V23_UP, V50_UP], ), + ( + 51, + "Per-minute queue and fleet history", + &[SCHEMA_DISCOVERY_UP, V51_UP], + ), ]; const V1_UP: &str = include_str!("../migrations/v001_canonical_schema.sql"); @@ -348,6 +353,7 @@ const V47_UP: &str = include_str!("../migrations/v047_replica_identity.sql"); const V48_UP: &str = include_str!("../migrations/v048_claim_cursor_catch_up.sql"); const V49_UP: &str = include_str!("../migrations/v049_prepared_lane_sequences.sql"); const V50_UP: &str = include_str!("../migrations/v050_drop_ready_lane_shard_index.sql"); +const V51_UP: &str = include_str!("../migrations/v051_queue_history.sql"); /// Old version numbers from pre-0.4 releases that used V3/V4/V5 numbering. /// Also tolerates the unreleased inline-V6 branch numbering used during review. diff --git a/awa-model/src/queue_history.rs b/awa-model/src/queue_history.rs new file mode 100644 index 000000000..fd133704c --- /dev/null +++ b/awa-model/src/queue_history.rs @@ -0,0 +1,760 @@ +//! Per-minute queue and fleet history (v051, #543). +//! +//! The elected maintenance leader samples queue depth and differences the +//! cumulative per-queue counters runtimes publish in their snapshots +//! ([`crate::admin::QueueOutcomeCounters`]) into per-minute deltas. The admin +//! UI reads the result to judge whether each queue is keeping up. +//! +//! Counter deltas are `None` ("unknown") rather than a guess whenever the +//! leader cannot account for every live runtime serving a queue: on its first +//! sample, for a runtime it has not seen before that started earlier, for +//! runtimes that predate the counters, and when the interval since the last +//! sample is not about a minute. Completed, failed and retried counts are also +//! unknown wherever a callback may have resolved a parked job, since that can +//! happen in any process. A runtime that dies between samples loses at most +//! its last snapshot interval of counts. + +use std::collections::{BTreeMap, BTreeSet, HashMap}; + +use chrono::{DateTime, TimeDelta, Utc}; +use serde::Serialize; +use sqlx::PgPool; +use uuid::Uuid; + +use crate::admin::{QueueOutcomeCounters, QueueOverview, QueueRuntimeMode, RuntimeInstance}; +use crate::error::AwaError; + +/// Counter deltas for one queue over one sample interval. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)] +pub struct QueueFlow { + pub claimed: i64, + pub completed: i64, + pub failed_attempts: i64, + pub retried: i64, + pub rate_limited: i64, + pub claim_wait_ms: i64, + pub parked: i64, +} + +/// How far a sample interval may stray from a minute and still be stored as +/// one. Workers publish counters on their own snapshot cadence, so some jitter +/// is inherent; an interval well off a minute (the first one after taking +/// leadership, or a delayed sample) would misstate a per-minute rate. +const MINUTE_TOLERANCE: TimeDelta = TimeDelta::seconds(15); + +impl QueueFlow { + fn between(previous: Option<&QueueOutcomeCounters>, current: &QueueOutcomeCounters) -> Self { + let delta = |now: u64, before: u64| i64::try_from(now - before).unwrap_or(i64::MAX); + match previous { + // Any counter going backwards means the runtime restarted (or its + // counters did); everything it reports now is new since then. + Some(prev) + if current.claimed >= prev.claimed + && current.completed >= prev.completed + && current.failed >= prev.failed + && current.retried >= prev.retried + && current.rate_limited >= prev.rate_limited + && current.claim_wait_ms >= prev.claim_wait_ms + && current.parked >= prev.parked => + { + Self { + claimed: delta(current.claimed, prev.claimed), + completed: delta(current.completed, prev.completed), + failed_attempts: delta(current.failed, prev.failed), + retried: delta(current.retried, prev.retried), + rate_limited: delta(current.rate_limited, prev.rate_limited), + claim_wait_ms: delta(current.claim_wait_ms, prev.claim_wait_ms), + parked: delta(current.parked, prev.parked), + } + } + _ => Self::between(Some(&QueueOutcomeCounters::default()), current), + } + } + + fn add(&mut self, other: Self) { + self.claimed = self.claimed.saturating_add(other.claimed); + self.completed = self.completed.saturating_add(other.completed); + self.failed_attempts = self.failed_attempts.saturating_add(other.failed_attempts); + self.retried = self.retried.saturating_add(other.retried); + self.rate_limited = self.rate_limited.saturating_add(other.rate_limited); + self.claim_wait_ms = self.claim_wait_ms.saturating_add(other.claim_wait_ms); + self.parked = self.parked.saturating_add(other.parked); + } +} + +/// What the leader remembers between samples. Lost on leader change, which +/// makes the next sample's counter deltas unknown. +#[derive(Debug, Default)] +pub struct CounterBaseline { + sampled_at: Option>, + readings: HashMap<(Uuid, String), QueueOutcomeCounters>, + waiting_external: HashMap, +} + +/// Per-queue runtime coverage for one sample. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct QueueCoverage { + pub flow: Option, + pub live_workers: i32, + /// Summed `max_workers` across live hard-reserved runtimes. `None` when + /// any live runtime runs the queue in weighted mode, where capacity is + /// shared and cannot be attributed to one queue. + pub capacity: Option, +} + +impl CounterBaseline { + /// Difference the live runtimes' counters against the previous sample and + /// advance the baseline to `now`. + pub fn advance( + &mut self, + now: DateTime, + instances: &[RuntimeInstance], + ) -> HashMap { + let mut coverage: HashMap = HashMap::new(); + let mut unknown: BTreeSet = BTreeSet::new(); + let about_a_minute = self.sampled_at.is_none_or(|sampled_at| { + ((now - sampled_at) - TimeDelta::minutes(1)).abs() <= MINUTE_TOLERANCE + }); + let mut readings = HashMap::new(); + + for instance in instances.iter().filter(|instance| !instance.stale) { + for snapshot in &instance.queues { + let entry = + coverage + .entry(snapshot.queue.clone()) + .or_insert_with(|| QueueCoverage { + flow: Some(QueueFlow::default()), + live_workers: 0, + capacity: Some(0), + }); + entry.live_workers += 1; + entry.capacity = match (entry.capacity, snapshot.config.mode) { + (Some(total), QueueRuntimeMode::HardReserved) => { + Some(total + i64::from(snapshot.config.max_workers.unwrap_or(0))) + } + _ => None, + }; + + let Some(current) = snapshot.counters else { + unknown.insert(snapshot.queue.clone()); + continue; + }; + let key = (instance.instance_id, snapshot.queue.clone()); + let previous = self.readings.get(&key); + let started_since_last_sample = self + .sampled_at + .is_some_and(|sampled_at| instance.started_at >= sampled_at); + if !about_a_minute || (previous.is_none() && !started_since_last_sample) { + // Counts from before our baseline: we cannot tell how many + // fall in this interval. + unknown.insert(snapshot.queue.clone()); + } else if let Some(flow) = entry.flow.as_mut() { + flow.add(QueueFlow::between(previous, ¤t)); + } + readings.insert(key, current); + } + } + for queue in unknown { + if let Some(entry) = coverage.get_mut(&queue) { + entry.flow = None; + } + } + self.sampled_at = Some(now); + self.readings = readings; + coverage + } + + /// Queues where a callback may have completed, failed or retried a job + /// since the last sample: a job was parked during the interval, or one was + /// already waiting at its start. Callbacks resolve from any process (an + /// HTTP receiver, an admin call), which no runtime counts. + pub fn callback_queues( + &mut self, + queues: &[QueueOverview], + coverage: &HashMap, + ) -> BTreeSet { + let waiting: HashMap = queues + .iter() + .map(|queue| (queue.queue.clone(), queue.waiting_external)) + .collect(); + let affected = queues + .iter() + .filter(|queue| { + // With no earlier reading, assume what is waiting now was + // waiting all interval. + let waited = self + .waiting_external + .get(&queue.queue) + .copied() + .unwrap_or(queue.waiting_external) + > 0; + let parked = coverage + .get(&queue.queue) + .and_then(|c| c.flow) + .is_some_and(|flow| flow.parked > 0); + waited || parked + }) + .map(|queue| queue.queue.clone()) + .collect(); + self.waiting_external = waiting; + affected + } +} + +/// One row of `awa.queue_history`. +#[derive(Debug, Clone, PartialEq, Serialize, sqlx::FromRow)] +pub struct QueueHistoryRow { + pub queue: String, + pub bucket: DateTime, + pub available: i64, + pub running: i64, + pub scheduled: i64, + pub retryable: i64, + pub failed: i64, + pub waiting_external: i64, + pub lag_seconds: Option, + pub paused: bool, + pub claimed: Option, + pub completed: Option, + pub failed_attempts: Option, + pub retried: Option, + pub rate_limited: Option, + pub claim_wait_ms: Option, + pub live_workers: i32, + pub capacity: Option, +} + +/// One row of `awa.fleet_history`. +#[derive(Debug, Clone, PartialEq, Serialize, sqlx::FromRow)] +pub struct FleetHistoryRow { + pub bucket: DateTime, + pub live_instances: i32, + pub healthy_instances: i32, + pub stale_instances: i32, + pub leader_instance_id: Option, + pub versions: Vec, +} + +/// Build the rows for one sample from current queue depth and runtime state. +pub fn build_sample( + bucket: DateTime, + queues: &[QueueOverview], + coverage: &HashMap, + callback_queues: &BTreeSet, + instances: &[RuntimeInstance], +) -> (Vec, FleetHistoryRow) { + let rows = queues + .iter() + .map(|queue| { + let covered = coverage.get(&queue.queue); + let flow = covered.and_then(|c| c.flow); + QueueHistoryRow { + queue: queue.queue.clone(), + bucket, + available: queue.available, + running: queue.running, + scheduled: queue.scheduled, + retryable: queue.retryable, + failed: queue.failed, + waiting_external: queue.waiting_external, + lag_seconds: queue.lag_seconds, + paused: queue.paused, + claimed: flow.map(|f| f.claimed), + completed: flow.map(|f| f.completed), + failed_attempts: flow.map(|f| f.failed_attempts), + retried: flow.map(|f| f.retried), + rate_limited: flow.map(|f| f.rate_limited), + claim_wait_ms: flow.map(|f| f.claim_wait_ms), + // A queue nobody serves has no unknown counts: nothing ran. + live_workers: covered.map_or(0, |c| c.live_workers), + capacity: covered.and_then(|c| c.capacity), + } + }) + .map(|mut row| { + if row.live_workers == 0 { + row.claimed = Some(0); + row.completed = Some(0); + row.failed_attempts = Some(0); + row.retried = Some(0); + row.rate_limited = Some(0); + row.claim_wait_ms = Some(0); + } + if callback_queues.contains(&row.queue) { + row.completed = None; + row.failed_attempts = None; + row.retried = None; + } + row + }) + .collect(); + + let live: Vec<&RuntimeInstance> = instances.iter().filter(|i| !i.stale).collect(); + let versions: BTreeSet = live.iter().map(|i| i.version.clone()).collect(); + let fleet = FleetHistoryRow { + bucket, + live_instances: live.len() as i32, + healthy_instances: live.iter().filter(|i| i.healthy).count() as i32, + stale_instances: (instances.len() - live.len()) as i32, + leader_instance_id: live.iter().find(|i| i.leader).map(|i| i.instance_id), + versions: versions.into_iter().collect(), + }; + (rows, fleet) +} + +/// Whether v051 is applied. Lets callers report "no history" on an older +/// schema instead of failing. +pub async fn history_installed(pool: &PgPool) -> Result { + let installed: bool = sqlx::query_scalar( + "SELECT to_regclass('awa.queue_history') IS NOT NULL \ + AND to_regclass('awa.fleet_history') IS NOT NULL", + ) + .fetch_one(pool) + .await?; + Ok(installed) +} + +/// Write one sample. A bucket that already has rows (a second leader in the +/// same minute after failover) keeps the first write. +pub async fn record_sample( + pool: &PgPool, + rows: &[QueueHistoryRow], + fleet: &FleetHistoryRow, +) -> Result<(), AwaError> { + let mut tx = pool.begin().await?; + if !rows.is_empty() { + let column = + |f: fn(&QueueHistoryRow) -> Option| rows.iter().map(f).collect::>(); + sqlx::query( + r#" + INSERT INTO awa.queue_history ( + queue, bucket, available, running, scheduled, retryable, failed, + waiting_external, lag_seconds, paused, claimed, completed, + failed_attempts, retried, rate_limited, claim_wait_ms, + live_workers, capacity + ) + SELECT * FROM UNNEST( + $1::text[], $2::timestamptz[], $3::bigint[], $4::bigint[], $5::bigint[], + $6::bigint[], $7::bigint[], $8::bigint[], $9::float8[], $10::bool[], + $11::bigint[], $12::bigint[], $13::bigint[], $14::bigint[], $15::bigint[], + $16::bigint[], $17::int[], $18::bigint[] + ) + ON CONFLICT (queue, bucket) DO NOTHING + "#, + ) + .bind(rows.iter().map(|r| r.queue.clone()).collect::>()) + .bind(rows.iter().map(|r| r.bucket).collect::>()) + .bind(rows.iter().map(|r| r.available).collect::>()) + .bind(rows.iter().map(|r| r.running).collect::>()) + .bind(rows.iter().map(|r| r.scheduled).collect::>()) + .bind(rows.iter().map(|r| r.retryable).collect::>()) + .bind(rows.iter().map(|r| r.failed).collect::>()) + .bind(rows.iter().map(|r| r.waiting_external).collect::>()) + .bind(rows.iter().map(|r| r.lag_seconds).collect::>()) + .bind(rows.iter().map(|r| r.paused).collect::>()) + .bind(column(|r| r.claimed)) + .bind(column(|r| r.completed)) + .bind(column(|r| r.failed_attempts)) + .bind(column(|r| r.retried)) + .bind(column(|r| r.rate_limited)) + .bind(column(|r| r.claim_wait_ms)) + .bind(rows.iter().map(|r| r.live_workers).collect::>()) + .bind(column(|r| r.capacity)) + .execute(&mut *tx) + .await?; + } + sqlx::query( + r#" + INSERT INTO awa.fleet_history ( + bucket, live_instances, healthy_instances, stale_instances, + leader_instance_id, versions + ) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (bucket) DO NOTHING + "#, + ) + .bind(fleet.bucket) + .bind(fleet.live_instances) + .bind(fleet.healthy_instances) + .bind(fleet.stale_instances) + .bind(fleet.leader_instance_id) + .bind(&fleet.versions) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) +} + +/// Delete history older than `retention`. Returns rows removed. +pub async fn prune_history(pool: &PgPool, retention: std::time::Duration) -> Result { + let secs = i64::try_from(retention.as_secs()).unwrap_or(i64::MAX); + let queues = sqlx::query( + "DELETE FROM awa.queue_history WHERE bucket < now() - make_interval(secs => $1)", + ) + .bind(secs as f64) + .execute(pool) + .await? + .rows_affected(); + let fleet = sqlx::query( + "DELETE FROM awa.fleet_history WHERE bucket < now() - make_interval(secs => $1)", + ) + .bind(secs as f64) + .execute(pool) + .await? + .rows_affected(); + Ok(queues + fleet) +} + +/// Queue rows since `since`, oldest first, grouped by queue. +pub async fn queue_history_since( + pool: &PgPool, + since: DateTime, +) -> Result>, AwaError> { + let rows: Vec = sqlx::query_as( + r#" + SELECT queue, bucket, available, running, scheduled, retryable, failed, + waiting_external, lag_seconds, paused, claimed, completed, + failed_attempts, retried, rate_limited, claim_wait_ms, + live_workers, capacity + FROM awa.queue_history + WHERE bucket >= $1 + ORDER BY queue, bucket + "#, + ) + .bind(since) + .fetch_all(pool) + .await?; + let mut grouped: BTreeMap> = BTreeMap::new(); + for row in rows { + grouped.entry(row.queue.clone()).or_default().push(row); + } + Ok(grouped) +} + +/// Fleet rows since `since`, oldest first. +pub async fn fleet_history_since( + pool: &PgPool, + since: DateTime, +) -> Result, AwaError> { + Ok(sqlx::query_as( + r#" + SELECT bucket, live_instances, healthy_instances, stale_instances, + leader_instance_id, versions + FROM awa.fleet_history + WHERE bucket >= $1 + ORDER BY bucket + "#, + ) + .bind(since) + .fetch_all(pool) + .await?) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::admin::{QueueRuntimeConfigSnapshot, QueueRuntimeSnapshot, StorageCapability}; + + fn at(seconds: i64) -> DateTime { + DateTime::from_timestamp(1_800_000_000 + seconds, 0).unwrap() + } + + fn counters(claimed: u64, completed: u64) -> QueueOutcomeCounters { + QueueOutcomeCounters { + claimed, + completed, + claim_wait_ms: claimed * 10, + ..Default::default() + } + } + + fn instance( + id: u128, + started_at: DateTime, + queues: Vec<(&str, Option, QueueRuntimeMode)>, + ) -> RuntimeInstance { + RuntimeInstance { + instance_id: Uuid::from_u128(id), + hostname: None, + pid: 1, + version: "0.7.0".into(), + storage_capability: StorageCapability::QueueStorage, + transition_role: Default::default(), + started_at, + last_seen_at: started_at, + snapshot_interval_ms: 10_000, + stale: false, + healthy: true, + postgres_connected: true, + poll_loop_alive: true, + heartbeat_alive: true, + maintenance_alive: true, + shutting_down: false, + leader: id == 1, + global_max_workers: None, + queues: queues + .into_iter() + .map(|(queue, counters, mode)| QueueRuntimeSnapshot { + queue: queue.into(), + in_flight: 0, + overflow_held: None, + config: QueueRuntimeConfigSnapshot { + mode, + max_workers: Some(8), + min_workers: None, + weight: None, + global_max_workers: None, + poll_interval_ms: 200, + deadline_duration_secs: 300, + priority_aging_interval_secs: 60, + claimers: None, + claim_batch_size: None, + dlq_enabled: None, + rate_limit: None, + }, + counters, + }) + .collect(), + } + } + + const HARD: QueueRuntimeMode = QueueRuntimeMode::HardReserved; + + #[test] + fn first_sample_after_leader_start_is_unknown() { + let mut baseline = CounterBaseline::default(); + let coverage = baseline.advance( + at(60), + &[instance( + 1, + at(0), + vec![("email", Some(counters(5, 5)), HARD)], + )], + ); + assert_eq!(coverage["email"].flow, None); + assert_eq!(coverage["email"].live_workers, 1); + assert_eq!(coverage["email"].capacity, Some(8)); + } + + #[test] + fn later_samples_difference_and_sum_across_instances() { + let mut baseline = CounterBaseline::default(); + baseline.advance( + at(60), + &[ + instance(1, at(0), vec![("email", Some(counters(5, 5)), HARD)]), + instance(2, at(0), vec![("email", Some(counters(1, 0)), HARD)]), + ], + ); + let coverage = baseline.advance( + at(120), + &[ + instance(1, at(0), vec![("email", Some(counters(15, 14)), HARD)]), + instance(2, at(0), vec![("email", Some(counters(4, 4)), HARD)]), + ], + ); + let flow = coverage["email"].flow.unwrap(); + assert_eq!(flow.claimed, 13); + assert_eq!(flow.completed, 13); + assert_eq!(flow.claim_wait_ms, 130); + assert_eq!(coverage["email"].capacity, Some(16)); + } + + #[test] + fn a_restart_counts_from_zero() { + let mut baseline = CounterBaseline::default(); + baseline.advance( + at(60), + &[instance( + 1, + at(0), + vec![("email", Some(counters(50, 50)), HARD)], + )], + ); + let coverage = baseline.advance( + at(120), + &[instance( + 1, + at(0), + vec![("email", Some(counters(3, 2)), HARD)], + )], + ); + assert_eq!(coverage["email"].flow.unwrap().claimed, 3); + } + + #[test] + fn new_instances_count_only_if_they_started_after_the_baseline() { + let mut baseline = CounterBaseline::default(); + baseline.advance( + at(60), + &[instance( + 1, + at(0), + vec![("email", Some(counters(1, 1)), HARD)], + )], + ); + let fresh = baseline.advance( + at(120), + &[ + instance(1, at(0), vec![("email", Some(counters(2, 2)), HARD)]), + instance(2, at(90), vec![("email", Some(counters(7, 7)), HARD)]), + ], + ); + assert_eq!(fresh["email"].flow.unwrap().completed, 8); + + // An instance the leader first sees, but which predates the baseline, + // makes the minute unknown rather than double counting. + let unseen = baseline.advance( + at(180), + &[ + instance(1, at(0), vec![("email", Some(counters(3, 3)), HARD)]), + instance(2, at(90), vec![("email", Some(counters(7, 7)), HARD)]), + instance(3, at(10), vec![("email", Some(counters(900, 900)), HARD)]), + ], + ); + assert_eq!(unseen["email"].flow, None); + } + + #[test] + fn runtimes_without_counters_make_the_queue_unknown() { + let mut baseline = CounterBaseline::default(); + baseline.advance( + at(60), + &[instance( + 1, + at(0), + vec![("email", Some(counters(1, 1)), HARD)], + )], + ); + let coverage = baseline.advance( + at(120), + &[ + instance(1, at(0), vec![("email", Some(counters(2, 2)), HARD)]), + instance(2, at(0), vec![("email", None, HARD)]), + ], + ); + assert_eq!(coverage["email"].flow, None); + assert_eq!(coverage["email"].live_workers, 2); + } + + #[test] + fn weighted_queues_have_no_attributable_capacity() { + let mut baseline = CounterBaseline::default(); + let coverage = baseline.advance( + at(60), + &[instance( + 1, + at(0), + vec![("email", None, QueueRuntimeMode::Weighted)], + )], + ); + assert_eq!(coverage["email"].capacity, None); + } + + fn overview(name: &str, available: i64, waiting_external: i64) -> QueueOverview { + QueueOverview { + queue: name.into(), + display_name: None, + description: None, + owner: None, + docs_url: None, + tags: vec![], + extra: serde_json::Value::Null, + descriptor_last_seen_at: None, + descriptor_stale: false, + descriptor_mismatch: false, + total_queued: available + waiting_external, + scheduled: 0, + available, + retryable: 0, + running: 0, + failed: 0, + waiting_external, + completed_last_hour: 0, + lag_seconds: Some(120.0), + paused: false, + } + } + + #[test] + fn intervals_well_off_a_minute_are_unknown() { + let mut baseline = CounterBaseline::default(); + let worker = |claimed| { + [instance( + 1, + at(0), + vec![("email", Some(counters(claimed, claimed)), HARD)], + )] + }; + baseline.advance(at(57), &worker(10)); + // Leadership taken at :57, next sample just after the minute: three + // seconds of counts must not be stored as a minute's worth. + assert_eq!(baseline.advance(at(62), &worker(11))["email"].flow, None); + let on_time = baseline.advance(at(121), &worker(71)); + assert_eq!(on_time["email"].flow.unwrap().claimed, 60); + // A sample delayed well past the minute is unknown too. + assert_eq!(baseline.advance(at(230), &worker(171))["email"].flow, None); + } + + #[test] + fn callbacks_make_outcomes_unknown_but_not_claims() { + let mut baseline = CounterBaseline::default(); + let worker = |claimed, parked| { + [instance( + 1, + at(0), + vec![( + "hooks", + Some(QueueOutcomeCounters { + parked, + ..counters(claimed, 0) + }), + HARD, + )], + )] + }; + let quiet = [overview("hooks", 0, 0), overview("plain", 3, 0)]; + let coverage = baseline.advance(at(60), &worker(5, 0)); + assert!(baseline.callback_queues(&quiet, &coverage).is_empty()); + + // Parked during the interval: whatever resolves it is not counted. + let coverage = baseline.advance(at(120), &worker(9, 4)); + let parked = baseline.callback_queues(&[overview("hooks", 0, 4)], &coverage); + assert!(parked.contains("hooks")); + let (rows, _) = build_sample(at(120), &[overview("hooks", 0, 4)], &coverage, &parked, &[]); + assert_eq!(rows[0].claimed, Some(4)); + assert_eq!(rows[0].completed, None); + assert_eq!(rows[0].failed_attempts, None); + assert_eq!(rows[0].retried, None); + + // Nothing parked this minute, but jobs were waiting at its start. + let coverage = baseline.advance(at(180), &worker(9, 4)); + let resolved = baseline.callback_queues(&[overview("hooks", 0, 0)], &coverage); + assert!(resolved.contains("hooks")); + + // Once nothing waited at the start and nothing parked, counts return. + let coverage = baseline.advance(at(240), &worker(9, 4)); + assert!(baseline + .callback_queues(&[overview("hooks", 0, 0)], &coverage) + .is_empty()); + } + + #[test] + fn unserved_callback_queues_are_unknown_not_zero() { + // No worker serves the queue, yet callbacks can still finish its jobs. + let queue = overview("receiver_only", 0, 2); + let callback = BTreeSet::from(["receiver_only".to_string()]); + let (rows, _) = build_sample(at(60), &[queue], &HashMap::new(), &callback, &[]); + assert_eq!(rows[0].claimed, Some(0)); + assert_eq!(rows[0].completed, None); + } + + #[test] + fn unserved_queues_record_zero_flow() { + let queue = overview("orphan", 4, 0); + let (rows, fleet) = build_sample(at(60), &[queue], &HashMap::new(), &BTreeSet::new(), &[]); + assert_eq!(rows[0].live_workers, 0); + assert_eq!(rows[0].claimed, Some(0)); + assert_eq!(fleet.live_instances, 0); + assert!(fleet.versions.is_empty()); + } +} diff --git a/awa-python/python/awa/_awa.pyi b/awa-python/python/awa/_awa.pyi index d873bc476..9a5f2589c 100644 --- a/awa-python/python/awa/_awa.pyi +++ b/awa-python/python/awa/_awa.pyi @@ -573,6 +573,7 @@ class Client: completed_retention_hours: float | None = None, failed_retention_hours: float | None = None, descriptor_retention_days: float | None = None, + queue_history_retention_hours: float | None = None, cleanup_batch_size: int | None = None, leader_election_interval_ms: int | None = None, heartbeat_interval_ms: int | None = None, diff --git a/awa-python/python/awa/client.py b/awa-python/python/awa/client.py index 371342b7f..ffe5313ef 100644 --- a/awa-python/python/awa/client.py +++ b/awa-python/python/awa/client.py @@ -870,6 +870,7 @@ async def start( completed_retention_hours: float | None = None, failed_retention_hours: float | None = None, descriptor_retention_days: float | None = None, + queue_history_retention_hours: float | None = None, cleanup_batch_size: int | None = None, leader_election_interval_ms: int | None = None, heartbeat_interval_ms: int | None = None, @@ -903,6 +904,10 @@ async def start( ``awa.job_kind_descriptors``. Defaults to 30 days; pass ``0`` to disable retention (useful if you manage the catalog externally). + ``queue_history_retention_hours`` controls how long the maintenance + leader keeps the per-minute queue and fleet history behind the admin + UI's trends. Defaults to 48 hours; pass ``0`` to stop sampling. + Workers are queue-storage-capable by default, but during a storage transition ``storage_transition_role`` controls how this runtime participates: @@ -936,6 +941,7 @@ async def start( completed_retention_hours=completed_retention_hours, failed_retention_hours=failed_retention_hours, descriptor_retention_days=descriptor_retention_days, + queue_history_retention_hours=queue_history_retention_hours, cleanup_batch_size=cleanup_batch_size, leader_election_interval_ms=leader_election_interval_ms, heartbeat_interval_ms=heartbeat_interval_ms, diff --git a/awa-python/src/client.rs b/awa-python/src/client.rs index 1719e3366..1d1382b50 100644 --- a/awa-python/src/client.rs +++ b/awa-python/src/client.rs @@ -1772,7 +1772,7 @@ impl PyClient { }) } - #[pyo3(signature = (queues=None, *, poll_interval_ms=200, global_max_workers=None, completed_retention_hours=None, failed_retention_hours=None, descriptor_retention_days=None, cleanup_batch_size=None, leader_election_interval_ms=None, heartbeat_interval_ms=None, promote_interval_ms=None, heartbeat_rescue_interval_ms=None, heartbeat_staleness_ms=None, deadline_rescue_interval_ms=None, callback_rescue_interval_ms=None, terminal_count_rollup_interval_ms=None, queue_storage_schema=None, queue_storage_queue_slot_count=16, queue_storage_lease_slot_count=8, queue_storage_claim_slot_count=8, queue_storage_queue_stripe_count=1, queue_storage_queue_rotate_interval_ms=1000, queue_storage_lease_rotate_interval_ms=1000, queue_storage_claim_rotate_interval_ms=None, queue_storage_rotate_min_slot_age_ms=None, queue_storage_rotate_target_slot_bytes=None, queue_storage_rotate_min_slot_rows=None, queue_storage_rotate_max_slot_age_ms=None, storage_transition_role=None))] + #[pyo3(signature = (queues=None, *, poll_interval_ms=200, global_max_workers=None, completed_retention_hours=None, failed_retention_hours=None, descriptor_retention_days=None, queue_history_retention_hours=None, cleanup_batch_size=None, leader_election_interval_ms=None, heartbeat_interval_ms=None, promote_interval_ms=None, heartbeat_rescue_interval_ms=None, heartbeat_staleness_ms=None, deadline_rescue_interval_ms=None, callback_rescue_interval_ms=None, terminal_count_rollup_interval_ms=None, queue_storage_schema=None, queue_storage_queue_slot_count=16, queue_storage_lease_slot_count=8, queue_storage_claim_slot_count=8, queue_storage_queue_stripe_count=1, queue_storage_queue_rotate_interval_ms=1000, queue_storage_lease_rotate_interval_ms=1000, queue_storage_claim_rotate_interval_ms=None, queue_storage_rotate_min_slot_age_ms=None, queue_storage_rotate_target_slot_bytes=None, queue_storage_rotate_min_slot_rows=None, queue_storage_rotate_max_slot_age_ms=None, storage_transition_role=None))] #[allow(clippy::too_many_arguments)] fn start<'py>( &self, @@ -1783,6 +1783,7 @@ impl PyClient { completed_retention_hours: Option, failed_retention_hours: Option, descriptor_retention_days: Option, + queue_history_retention_hours: Option, cleanup_batch_size: Option, leader_election_interval_ms: Option, heartbeat_interval_ms: Option, @@ -1877,6 +1878,14 @@ impl PyClient { } builder = builder.descriptor_retention(Duration::from_secs_f64(days * 86400.0)); } + if let Some(hours) = queue_history_retention_hours { + if !hours.is_finite() || hours < 0.0 { + return Err(pyo3::exceptions::PyValueError::new_err( + "queue_history_retention_hours must be a non-negative finite number (0 disables)", + )); + } + builder = builder.queue_history_retention(Duration::from_secs_f64(hours * 3600.0)); + } if let Some(batch_size) = cleanup_batch_size { if batch_size <= 0 { return Err(pyo3::exceptions::PyValueError::new_err( diff --git a/awa-python/tests/test_start_config.py b/awa-python/tests/test_start_config.py index eb404d444..12482c0ef 100644 --- a/awa-python/tests/test_start_config.py +++ b/awa-python/tests/test_start_config.py @@ -294,11 +294,24 @@ async def handle(job): [(queue, 10)], completed_retention_hours=1.0, failed_retention_hours=168.0, + queue_history_retention_hours=24.0, cleanup_batch_size=500, ) await client.shutdown() +@pytest.mark.asyncio +async def test_negative_queue_history_retention_raises(client): + queue = "cfg_history" + + @client.task(ConfigTestJob, queue=queue) + async def handle(job): + return None + + with pytest.raises(ValueError, match="queue_history_retention_hours"): + await client.start([(queue, 1)], queue_history_retention_hours=-1.0) + + @pytest.mark.asyncio async def test_maintenance_interval_kwargs_accepted(client): """start() accepts all maintenance interval kwargs including heartbeat_staleness_ms.""" diff --git a/awa-ui/frontend/e2e/dashboard.spec.ts b/awa-ui/frontend/e2e/dashboard.spec.ts index 7676db74c..d4758f623 100644 --- a/awa-ui/frontend/e2e/dashboard.spec.ts +++ b/awa-ui/frontend/e2e/dashboard.spec.ts @@ -1,76 +1,91 @@ -import { test, expect } from "@playwright/test"; +import { test, expect, type Page } from "@playwright/test"; + +/** The overview lists only queues needing attention until expanded; seeded + * queues may or may not qualify depending on their age, so expand first. */ +async function showAllQueues(page: Page) { + await expect(page.getByTestId("verdict")).not.toHaveText("Checking queues…"); + const expand = page.getByRole("button", { + name: /^Show \d+ queues? keeping up$/, + }); + if (await expand.isVisible()) await expand.click(); + return page.getByRole("table", { + name: /^(All queues by risk|Queues needing attention)$/, + }); +} test.describe("Operations overview", () => { - test("overview separates current work and last-hour completions", async ({ - page, - }) => { + test("answers whether the system is keeping up", async ({ page }) => { await page.goto("/"); - await expect( - page.getByRole("heading", { name: "Work in motion" }), - ).toBeVisible(); - await expect( - page.getByRole("heading", { name: "Current work" }), - ).toBeVisible(); - await expect( - page.getByText("Completed · last 60 minutes", { exact: true }), - ).toBeVisible(); - await expect(page.locator(".work-counter")).toHaveCount(6); + await expect(page.getByRole("heading", { name: "Overview" })).toBeVisible(); + await expect(page.getByTestId("verdict")).not.toHaveText( + "Checking queues…", + ); + await expect(page.getByText("Waiting across all queues")).toBeVisible(); + // The e2e server runs no workers, so no leader has recorded history; + // the page says why trends are missing instead of inventing them. + await expect(page.getByText(/No trends( yet)?:/)).toBeVisible(); }); - test("queue pressure includes available wait and state drilldowns", async ({ + test("queue table includes trend columns and state drilldowns", async ({ page, }) => { await page.goto("/"); - const table = page.getByRole("table", { name: "Queue pressure" }); + const table = await showAllQueues(page); await expect(table).toBeVisible(); for (const header of [ "Queue", - "Available", - "Oldest available", - "Running", - "Retryable / failed", + "Status", + "Waiting", + "Last hour", + "Trend", + "Oldest wait", + "Limited by", ]) { await expect( table.getByRole("columnheader", { name: header, exact: true }), ).toBeVisible(); } - const row = table.getByRole("row", { name: /e2e_test/ }); + const row = table.getByRole("row", { name: /E2E Queue/ }); await row.locator('a[href*="state=available"]').click(); await expect(page).toHaveURL(/state=available/); await expect(page).toHaveURL(/e2e_test/); }); - for (const state of ["available", "retryable", "failed"]) { - test(`${state} counter links to its job state`, async ({ page }) => { + for (const [label, state] of [ + ["Retrying", "retryable"], + ["Failed", "failed"], + ["Running", "running"], + ] as const) { + test(`${label} counter links to its job state`, async ({ page }) => { await page.goto("/"); - await page.locator(`.work-counter[href*="state=${state}"]`).click(); + await page + .getByRole("navigation", { name: "Job states" }) + .getByRole("link", { name: new RegExp(`^${label}`) }) + .click(); await expect(page).toHaveURL(new RegExp(`state=${state}`)); await expect(page.getByRole("heading", { name: "Jobs" })).toBeVisible(); }); } - test("worker summary reports health or explicit empty state", async ({ + test("worker summary reports problems or an explicit healthy state", async ({ page, }) => { await page.goto("/"); - const section = page.getByRole("region", { name: "Worker health" }); + await expect(page.getByRole("link", { name: "All workers" })).toBeVisible(); await expect( - section.getByRole("link", { name: "All workers" }), + page + .getByText( + /Workers needing attention|reporting instances are healthy|No runtime snapshots/, + ) + .or(page.getByRole("list", { name: "Workers needing attention" })), ).toBeVisible(); - await expect(section).toContainText( - /healthy.*live.*stale.*leader|No runtime snapshots/, - ); }); - test("failed job drilldown opens the retained failure", async ({ page }) => { + test("failure group opens the latest retained failure", async ({ page }) => { await page.goto("/"); - const section = page.getByRole("region", { - name: "Failed jobs to inspect", - }); - await expect( - section.getByText(/Latest 5 retained failed jobs/), - ).toBeVisible(); - await section.locator(".failure-preview-row").first().click(); + const groups = page.getByRole("list", { name: "Failure groups" }); + await expect(groups).toBeVisible(); + await groups.getByRole("link").first().click(); await expect(page).toHaveURL(/\/jobs\/\d+/); await expect(page.getByRole("heading", { name: /Job #\d+/ })).toBeVisible(); }); diff --git a/awa-ui/frontend/e2e/navigation.spec.ts b/awa-ui/frontend/e2e/navigation.spec.ts index 753b5da45..01ecc19e5 100644 --- a/awa-ui/frontend/e2e/navigation.spec.ts +++ b/awa-ui/frontend/e2e/navigation.spec.ts @@ -40,7 +40,7 @@ test.describe("Shell navigation", () => { await nav.getByRole("link", { name: "Overview" }).click(); await expect(page).toHaveURL(/\/$/); await expect( - page.getByRole("heading", { name: "Work in motion" }) + page.getByRole("heading", { name: "Overview" }) ).toBeVisible(); }); @@ -55,7 +55,7 @@ test.describe("Shell navigation", () => { await logoLink.click(); await expect(page).toHaveURL(/\/$/); await expect( - page.getByRole("heading", { name: "Work in motion" }) + page.getByRole("heading", { name: "Overview" }) ).toBeVisible(); }); diff --git a/awa-ui/frontend/e2e/operations-fixture.ts b/awa-ui/frontend/e2e/operations-fixture.ts index 32b95966d..c7ae6da6b 100644 --- a/awa-ui/frontend/e2e/operations-fixture.ts +++ b/awa-ui/frontend/e2e/operations-fixture.ts @@ -1,23 +1,206 @@ import type { Page } from "@playwright/test"; -import type { JobRow, QueueStats, RuntimeInstance } from "../src/lib/api"; +import type { + JobRow, + QueueAssessment, + QueueHealthKind, + QueueStats, + RuntimeInstance, +} from "../src/lib/api"; +import { formatDuration, formatRate } from "../src/lib/operations"; + +/** + * Three moments in one synthetic fleet: + * + * - `normal`: every queue keeps up; nothing needs a human. + * - `incident`: a 0.7.1 rollout left image-processing with no worker + * claiming it, email-delivery is falling behind its rate limit, + * invoice-export was paused, and search-index drains a burst. + * - `recovery`: a 0.7.2 hotfix is rolled out, the email rate limit raised + * and invoice-export resumed; the backlogs drain. + */ +export type Phase = "normal" | "incident" | "recovery"; +type Scenario = Phase | "empty" | "error" | "loading"; + +const NAMES = [ + "email-delivery", + "invoice-export", + "webhook-dispatch", + "image-processing", + "search-index", + "daily-reports", +]; +const EMAIL = 0; +const INVOICE = 1; +const WEBHOOK = 2; +const IMAGE = 3; +const SEARCH = 4; + +/** What a queue looks like now, and how its waiting count got there. */ +interface QueueStory { + available: number; + running: number; + retryable: number; + failed: number; + lag: number; + paused?: boolean; + /** Waiting jobs `minutesAgo` minutes ago. */ + depth: (minutesAgo: number) => number; + /** The server's verdict while work is waiting (otherwise steady/idle). */ + verdict?: { + health: QueueHealthKind; + perMinute: number; + limitedBy?: QueueAssessment["limited_by"]; + reason?: string; + }; + claimed: number; + arrivals: number; + claimWait: number; + failing?: number; + rateLimited?: boolean; +} + +function story(phase: Phase, index: number): QueueStory { + const quiet = 180 + index * 7; + const steady = (available: number): QueueStory => ({ + available, + running: 40 + index, + retryable: phase === "normal" ? 2 : 10, + failed: 0, + lag: phase === "normal" ? 1.2 : 2.4, + depth: (m) => available + ((m * 7 + index) % 5) * 4, + claimed: 2140, + arrivals: 2140, + claimWait: 1.2, + }); + if (phase === "normal") { + if (index === SEARCH) return { ...steady(0), depth: () => 0 }; + if (index === EMAIL) + return { ...steady(1840), retryable: 12, failed: 3, lag: 2.1 }; + return steady(quiet); + } + if (phase === "incident") { + if (index === EMAIL) + return { + ...steady(184200), + retryable: 1840, + failed: 236, + lag: 458, + depth: (m) => (m <= 20 ? 184200 - m * 2500 : 134200 + (m % 3) * 300), + verdict: { + health: "behind", + perMinute: 2500, + limitedBy: "rate_limit", + }, + arrivals: 4640, + claimWait: 410, + failing: 4, + rateLimited: true, + }; + if (index === INVOICE) + return { + ...steady(2117), + lag: 135, + paused: true, + depth: (m) => (m <= 12 ? 2117 : 40 + (m % 4) * 5), + }; + if (index === IMAGE) + return { + ...steady(2151), + running: 0, + lag: 95, + depth: (m) => (m <= 2 ? 2151 - m * 900 : 30 + (m % 2) * 10), + verdict: { + health: "stalled", + perMinute: 415, + limitedBy: "nothing", + reason: "Nothing running or claimed; oldest has waited 1m 35s", + }, + claimed: 0, + arrivals: 415, + }; + if (index === SEARCH) + return { + ...steady(12168), + depth: (m) => (m <= 15 ? 12168 + m * 2000 : 4000), + verdict: { health: "draining", perMinute: -2000 }, + arrivals: 140, + }; + return steady(2100 + index * 17); + } + // recovery + if (index === EMAIL) + return { + ...steady(62000), + retryable: 310, + failed: 236, + lag: 210, + depth: (m) => + m <= 14 + ? 62000 + m * 8700 + : m <= 34 + ? 184200 - (m - 14) * 2500 + : 134200 + (m % 3) * 300, + verdict: { health: "draining", perMinute: -8700 }, + claimed: 10840, + arrivals: 2140, + claimWait: 190, + }; + if (index === INVOICE) + return { + ...steady(640), + lag: 20, + depth: (m) => (m <= 6 ? 640 + m * 246 : m <= 26 ? 2117 : 40), + verdict: { health: "draining", perMinute: -246 }, + }; + if (index === IMAGE) + return { + ...steady(380), + lag: 40, + depth: (m) => (m <= 10 ? 380 + m * 177 : m <= 22 ? 2151 : 30), + verdict: { health: "draining", perMinute: -177 }, + }; + if (index === SEARCH) return { ...steady(0), depth: () => 0 }; + return steady(2100 + index * 17); +} + +const RANK: Record = { + stalled: 0, + behind: 1, + slow: 2, + paused: 3, + draining: 4, + unknown: 5, + steady: 6, + idle: 7, +}; + +const LABEL: Record = { + stalled: "Not being worked", + behind: "Falling behind", + slow: "Long wait", + paused: "Paused", + draining: "Draining", + unknown: "No trend yet", + steady: "Steady", + idle: "Idle", +}; + +function plural(count: number, noun: string): string { + return `${count} ${noun}${count === 1 ? "" : "s"}`; +} export async function mockOperations( page: Page, - scenario: "busy" | "empty" | "error" | "loading" = "busy", + scenario: Scenario = "incident", readOnly = false, ) { + const phase: Phase = + scenario === "normal" || scenario === "recovery" ? scenario : "incident"; const now = Date.now(); const ago = (seconds: number) => new Date(now - seconds * 1000).toISOString(); - const queues: QueueStats[] = Array.from({ length: 24 }, (_, index) => ({ - queue: - [ - "email-delivery", - "invoice-export", - "webhook-dispatch", - "image-processing", - "search-index", - "daily-reports", - ][index] ?? `tenant-sync-${index}`, + const stories = Array.from({ length: 24 }, (_, index) => story(phase, index)); + const queues: QueueStats[] = stories.map((s, index) => ({ + queue: NAMES[index] ?? `tenant-sync-${index}`, display_name: null, description: "Synthetic queue for local operator validation", owner: "Platform", @@ -28,20 +211,21 @@ export async function mockOperations( descriptor_stale: false, descriptor_mismatch: false, total_queued: 0, - available: index === 0 ? 184200 : 2100 + index * 17, - running: 40 + index, - retryable: index === 0 ? 1840 : 10, - failed: index === 0 ? 236 : 0, + available: s.available, + running: s.running, + retryable: s.retryable, + failed: s.failed, scheduled: 850 + index, - waiting_external: index === 2 ? 14400 : 0, - completed_last_hour: 128400 + index * 211, - lag_seconds: index === 0 ? 458 : index === 1 ? 135 : 2.4, - paused: index === 1, - })).map((q) => ({ - ...q, - total_queued: - q.available + q.running + q.retryable + q.scheduled + q.waiting_external, + waiting_external: + index === WEBHOOK ? (phase === "normal" ? 1400 : 14400) : 0, + completed_last_hour: s.claimed * 60, + lag_seconds: s.lag, + paused: s.paused ?? false, })); + for (const q of queues) + q.total_queued = + q.available + q.running + q.retryable + q.scheduled + q.waiting_external; + const config = { mode: "hard_reserved", max_workers: 64, @@ -54,72 +238,294 @@ export async function mockOperations( dlq_enabled: true, rate_limit: null, }; + const version = { normal: "0.7.1", incident: "0.7.1", recovery: "0.7.2" }[ + phase + ]; const instances: RuntimeInstance[] = Array.from( { length: 48 }, - (_, index) => ({ - instance_id: `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`, - hostname: `worker-eu-${String(index + 1).padStart(2, "0")}`, - pid: 1300 + index, - version: "0.7.0", - storage_capability: "queue_storage", - transition_role: "auto", - started_at: ago(86400), - last_seen_at: ago(index === 0 ? 170 : 4), - snapshot_interval_ms: 10000, - stale: index === 0, - healthy: index > 1, - postgres_connected: true, - poll_loop_alive: index !== 1, - heartbeat_alive: true, - maintenance_alive: true, - shutting_down: false, - leader: index === 2, - global_max_workers: 64, - queues: queues - .slice(0, 6) - .map((q) => ({ + (_, index) => { + const troubled = phase === "incident"; + return { + instance_id: `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`, + hostname: `worker-eu-${String(index + 1).padStart(2, "0")}`, + pid: 1300 + index, + version, + storage_capability: "queue_storage", + transition_role: "auto", + started_at: ago(phase === "recovery" ? 600 : 86400), + last_seen_at: ago(troubled && index === 0 ? 170 : 4), + snapshot_interval_ms: 10000, + stale: troubled && index === 0, + healthy: !troubled || index > 1, + postgres_connected: true, + poll_loop_alive: !troubled || index !== 1, + heartbeat_alive: true, + maintenance_alive: true, + shutting_down: false, + leader: index === 2, + global_max_workers: 64, + queues: queues.slice(0, 6).map((q) => ({ queue: q.queue, in_flight: 8, overflow_held: 0, config: config as RuntimeInstance["queues"][number]["config"], })), - }), + }; + }, ); - const jobs: JobRow[] = Array.from({ length: 60 }, (_, index) => ({ - id: 984213 - index, - queue: "email-delivery", - kind: "SendTransactionalEmail", - args: { template: "receipt", synthetic: true }, - state: "failed", - priority: 2, - original_priority: 2, - attempt: 5, - max_attempts: 5, - run_at: ago(900), - heartbeat_at: null, - deadline_at: null, - attempted_at: ago(120), - finalized_at: ago(index * 12 + 8), - created_at: ago(1000), - errors: [ - { - error: "Provider request timed out after 30 seconds", - attempt: 5, - at: ago(8), - }, + const live = instances.filter((i) => !i.stale); + + // Per-minute history the way the leader records it: an hour of buckets, + // a leader that started 50 minutes ago, and a three-minute gap where + // sampling failed. + const BUCKETS = 60; + const minuteStart = Math.floor(now / 60_000) * 60_000; + const bucketAt = (bucket: number) => + new Date(minuteStart - (BUCKETS - 1 - bucket) * 60_000).toISOString(); + const sampled = (bucket: number) => + bucket >= 10 && ![33, 34, 35].includes(bucket); + const series = (value: (bucket: number) => T) => + Array.from({ length: BUCKETS }, (_, bucket) => + sampled(bucket) ? value(bucket) : null, + ); + const events = { + normal: [], + incident: [ + [41, "version_change", "Versions 0.7.0 → 0.7.0, 0.7.1"], + [44, "version_change", "Versions 0.7.0, 0.7.1 → 0.7.1"], + [47, "instances_lost", "1 instance stopped reporting (48 → 47 live)"], ], - metadata: {}, - tags: ["synthetic"], - progress: null, - callback_id: null, - callback_timeout_at: null, - callback_filter: null, - callback_on_complete: null, - callback_on_fail: null, - callback_transform: null, - queue_descriptor: null, - kind_descriptor: null, - })); + recovery: [ + [27, "version_change", "Versions 0.7.0 → 0.7.0, 0.7.1"], + [30, "version_change", "Versions 0.7.0, 0.7.1 → 0.7.1"], + [32, "instances_lost", "1 instance stopped reporting (48 → 47 live)"], + [48, "version_change", "Versions 0.7.1 → 0.7.1, 0.7.2"], + [50, "leader_change", "Leader 3f2a91c4 → 8d07b1e2"], + [51, "version_change", "Versions 0.7.1, 0.7.2 → 0.7.2"], + ], + }[phase] as [number, string, string][]; + const queueHistory = { + installed: true, + bucket_secs: 60, + minutes: BUCKETS, + observed_since: bucketAt(10), + buckets: Array.from({ length: BUCKETS }, (_, bucket) => bucketAt(bucket)), + queues: Object.fromEntries( + queues.map((q, index) => { + const s = stories[index]!; + return [ + q.queue, + { + available: series((b) => Math.max(0, s.depth(BUCKETS - 1 - b))), + running: series(() => q.running), + lag_seconds: series(() => q.lag_seconds), + claimed: series(() => s.claimed), + completed: series(() => s.claimed), + failed_attempts: series(() => s.failing ?? 0), + rate_limited: series(() => (s.rateLimited ? 30 : 0)), + }, + ]; + }), + ), + fleet: { + live_instances: series(() => live.length), + healthy_instances: series(() => live.filter((i) => i.healthy).length), + }, + events: events.map(([bucket, kind, detail]) => ({ + at: bucketAt(bucket), + kind, + detail, + })), + }; + + // The server classifies queues (awa-ui/src/assessment.rs); the fixture + // reproduces its verdicts for each story from the live queue state, so + // pause, resume and drain change the report as they would for real. + const assess = (q: QueueStats, index: number): QueueAssessment => { + const s = stories[index]!; + const base = { + queue: q.queue, + flow: { + claimed_per_minute: s.claimed, + completed_per_minute: s.claimed, + failed_per_minute: s.failing ?? 0, + retried_per_minute: 0, + arrivals_per_minute: s.arrivals, + mean_claim_wait_seconds: s.claimWait, + rate_limited: s.rateLimited ?? false, + }, + live_workers: live.length, + in_flight: q.running, + capacity: 64 * live.length, + clear_seconds: null as number | null, + trend: { per_minute: 0, change: 0, minutes: 5 }, + limited_by: null as QueueAssessment["limited_by"], + }; + const make = ( + health: QueueHealthKind, + reason: string, + extra: Partial = {}, + ): QueueAssessment => ({ + ...base, + health, + label: LABEL[health], + reason, + needs_attention: + health === "stalled" || + health === "behind" || + health === "slow" || + (health === "paused" && q.available > 0), + rank: health === "paused" && q.available === 0 ? 7 : RANK[health], + ...extra, + }); + if (q.paused) + return q.available + ? make("paused", `${q.available} waiting while paused`, { + limited_by: "paused", + }) + : make("paused", "Nothing waiting"); + if (q.available === 0) return make("idle", "Nothing waiting"); + const verdict = s.verdict; + if (!verdict) return make("steady", "Backlog holding level"); + const trend = { + per_minute: verdict.perMinute, + change: verdict.perMinute * 5, + minutes: 5, + }; + if (verdict.health === "draining") { + const clear = (q.available / -verdict.perMinute) * 60; + return make( + "draining", + `Clears in about ${formatDuration(clear)} at this pace`, + { trend, clear_seconds: clear }, + ); + } + return make( + verdict.health, + verdict.reason ?? + `Backlog ${formatRate(verdict.perMinute, true)} over 5m`, + { trend, limited_by: verdict.limitedBy ?? null }, + ); + }; + const healthReport = () => { + const assessed = queues + .map((q, index) => assess(q, index)) + .sort( + (a, b) => + a.rank - b.rank || + queues.find((q) => q.queue === b.queue)!.available - + queues.find((q) => q.queue === a.queue)!.available || + a.queue.localeCompare(b.queue), + ); + const count = (h: QueueHealthKind) => + assessed.filter((a) => a.health === h); + const stalled = count("stalled"); + const behind = count("behind").length; + const paused = assessed.filter( + (a) => a.health === "paused" && a.needs_attention, + ).length; + const recovering = count("draining").filter( + (a) => (a.clear_seconds ?? 0) >= 120, + ); + const others = [ + behind && `${plural(behind, "queue")} falling behind`, + paused && `${plural(paused, "paused queue")} holding work`, + ].filter(Boolean); + const longest = Math.max(...recovering.map((a) => a.clear_seconds ?? 0)); + const [status, title, detail] = stalled.length + ? [ + "critical", + `${plural(stalled.length, "queue")} not being worked`, + [ + `Waiting work nobody is claiming: ${stalled.map((a) => a.queue).join(", ")}.`, + ...others, + ].join(" · "), + ] + : behind + ? ["warning", `${plural(behind, "queue")} falling behind`, ""] + : paused + ? [ + "ok", + `${plural(paused, "paused queue")} holding work`, + "Every running queue is keeping up. Resume paused queues when ready.", + ] + : recovering.length + ? [ + "ok", + "Recovering", + `${plural(recovering.length, "queue")} draining a backlog (${recovering + .slice(0, 3) + .map((a) => a.queue) + .join( + ", ", + )}); all clear in about ${formatDuration(longest)} at this pace.`, + ] + : [ + "ok", + "Keeping up", + `${plural(assessed.length, "queue")} steady, draining, or idle.`, + ]; + return { + status, + title, + detail, + generated_at: new Date(now).toISOString(), + history_available: true, + history_since: bucketAt(10), + queues: assessed, + }; + }; + + const failureCount = phase === "normal" ? 3 : 60; + const jobs: JobRow[] = Array.from({ length: failureCount }, (_, index) => { + const image = phase !== "normal" && index % 7 === 3; + // After recovery the failures stopped about 14 minutes ago. + const finalized = + phase === "normal" + ? 1200 + index * 380 + : index * 12 + (phase === "recovery" ? 840 : 8); + return { + id: 984213 - index, + queue: image ? "image-processing" : "email-delivery", + kind: image ? "ResizeImage" : "SendTransactionalEmail", + args: { template: "receipt", synthetic: true }, + state: "failed", + priority: 2, + original_priority: 2, + attempt: 5, + max_attempts: 5, + run_at: ago(finalized + 900), + heartbeat_at: null, + deadline_at: null, + attempted_at: ago(finalized + 100), + finalized_at: ago(finalized), + created_at: ago(finalized + 1000), + errors: [ + { + error: + phase === "normal" + ? "Mailbox unavailable (550 5.1.1)" + : image + ? `Out of memory decoding upload ${7_310_000 + index}` + : "Provider request timed out after 30 seconds", + attempt: 5, + at: ago(finalized), + }, + ], + metadata: {}, + tags: ["synthetic"], + progress: null, + callback_id: null, + callback_timeout_at: null, + callback_filter: null, + callback_on_complete: null, + callback_on_fail: null, + callback_transform: null, + queue_descriptor: null, + kind_descriptor: null, + }; + }); + const mutations: string[] = []; let failReads = scenario === "error"; await page.route("**/api/**", async (route) => { @@ -160,36 +566,54 @@ export async function mockOperations( await new Promise((resolve) => setTimeout(resolve, 2500)); if ( failReads && - ["/stats", "/queues", "/runtime", "/jobs", "/dlq/depth"].includes(path) + [ + "/stats", + "/queues", + "/stats/queue-history", + "/health/queues", + "/runtime", + "/jobs", + "/dlq/depth", + ].includes(path) ) return route.fulfill({ status: 503, json: { error: "Synthetic API unavailable" }, }); const empty = scenario === "empty"; + const sum = (pick: (q: QueueStats) => number) => + queues.reduce((total, q) => total + pick(q), 0); let data: unknown = []; if (path === "/stats") data = empty ? {} : { - available: queues.reduce((s, q) => s + q.available, 0), - running: 1236, - retryable: 2070, - failed: 236, - scheduled: 20676, - waiting_external: 14400, + available: sum((q) => q.available), + running: sum((q) => q.running), + retryable: sum((q) => q.retryable), + failed: sum((q) => q.failed), + scheduled: sum((q) => q.scheduled), + waiting_external: sum((q) => q.waiting_external), completed: 14820000, }; else if (path === "/queues") data = empty ? [] : queues; + else if (path === "/health/queues") + data = empty + ? { ...healthReport(), status: "ok", title: "Keeping up", queues: [] } + : healthReport(); + else if (path === "/stats/queue-history") + data = empty + ? { ...queueHistory, observed_since: null, queues: {}, events: [] } + : queueHistory; else if (path === "/queues/runtime") data = empty ? [] : queues.map((q) => ({ queue: q.queue, - instance_count: 48, - live_instances: 47, - stale_instances: 1, - healthy_instances: 46, + instance_count: instances.length, + live_instances: live.length, + stale_instances: instances.length - live.length, + healthy_instances: live.filter((i) => i.healthy).length, total_in_flight: q.running, overflow_held_total: 0, config_mismatch: false, @@ -199,17 +623,24 @@ export async function mockOperations( data = queues.find((q) => q.queue === decodeURIComponent(path.slice(8))); else if (path === "/runtime") data = { - total_instances: empty ? 0 : 48, - live_instances: empty ? 0 : 47, - stale_instances: empty ? 0 : 1, - healthy_instances: empty ? 0 : 46, + total_instances: empty ? 0 : instances.length, + live_instances: empty ? 0 : live.length, + stale_instances: empty ? 0 : instances.length - live.length, + healthy_instances: empty ? 0 : live.filter((i) => i.healthy).length, leader_instances: empty ? 0 : 1, instances: empty ? [] : instances, }; else if (path === "/dlq/depth") data = { - total: empty ? 0 : 1240, - by_queue: empty ? [] : [{ queue: "email-delivery", count: 1240 }], + total: empty ? 0 : phase === "normal" ? 12 : 1240, + by_queue: empty + ? [] + : [ + { + queue: "email-delivery", + count: phase === "normal" ? 12 : 1240, + }, + ], }; else if (path === "/jobs") data = empty diff --git a/awa-ui/frontend/e2e/operations.spec.ts b/awa-ui/frontend/e2e/operations.spec.ts index feb1ac992..74006ee43 100644 --- a/awa-ui/frontend/e2e/operations.spec.ts +++ b/awa-ui/frontend/e2e/operations.spec.ts @@ -3,7 +3,7 @@ import { mockOperations } from "./operations-fixture"; const artifacts = process.env.AWA_REVIEW_ARTIFACT_DIR ?? "test-results/operations-review"; -test("busy overview communicates current work and opens targeted drilldowns", async ({ +test("incident overview leads with the problem and opens targeted drilldowns", async ({ page, }) => { const fixture = await mockOperations(page); @@ -11,22 +11,49 @@ test("busy overview communicates current work and opens targeted drilldowns", as page.on("pageerror", (error) => errors.push(error.message)); await page.setViewportSize({ width: 1440, height: 1050 }); await page.goto("/"); + await expect(page.getByRole("heading", { name: "Overview" })).toBeVisible(); + await expect(page.getByTestId("verdict")).toHaveText( + "1 queue not being worked", + ); + await expect(page.getByText(/1 queue falling behind/)).toBeVisible(); + const attention = page + .getByRole("table", { name: "Queues needing attention" }) + .locator("tbody tr"); + // Three queues need attention; the draining one stays in view. + await expect(attention).toHaveCount(4); + await expect(attention.nth(0)).toContainText("image-processing"); + await expect(attention.nth(0)).toContainText("Not being worked"); + await expect(attention.nth(1)).toContainText("email-delivery"); + await expect(attention.nth(1)).toContainText("Falling behind"); + await expect(attention.nth(1)).toContainText("+2,500/min"); + await expect(attention.nth(1)).toContainText("7m 38s"); + await expect(attention.nth(1)).toContainText("Rate limit"); + await expect(attention.nth(0)).toContainText("No visible limit"); + // Fleet changes are listed and marked on the total-backlog chart. await expect( - page.getByRole("heading", { name: "Work in motion" }), - ).toBeVisible(); - await expect( - page.locator(".work-counter").filter({ hasText: "Available" }), - ).toContainText("237,192"); + page.getByRole("list", { name: "Fleet changes" }).getByRole("listitem"), + ).toHaveCount(3); + await expect(page.locator("[data-marker]")).toHaveCount(3); + await expect(page.getByText(/arriving .*\/min · last 5m/)).toBeVisible(); + await expect(attention.nth(2)).toContainText("Paused"); + await expect(attention.nth(3)).toContainText("search-index"); + await expect(attention.nth(3)).toContainText("Draining"); + await page.getByRole("button", { name: "Show 20 queues keeping up" }).click(); + const everything = page + .getByRole("table", { name: "All queues by risk" }) + .locator("tbody tr"); + await expect(everything).toHaveCount(24); + await expect(everything.nth(3)).toContainText("search-index"); + await expect(everything.nth(3)).toContainText("Draining"); await expect( - page.locator(".work-counter").filter({ hasText: "Retryable" }), - ).toContainText("2,070"); - await expect(page.locator(".worker-preview-row")).toHaveCount(4); - await expect(page.locator(".worker-preview-row").first()).toContainText( - "health unknown", - ); - await expect(page.locator(".pressure-table tbody tr").first()).toContainText( - "7m 38s", - ); + page.getByRole("navigation", { name: "Job states" }), + ).toContainText("Retrying2,070"); + const workers = page.getByRole("list", { name: "Workers needing attention" }); + await expect(workers.getByRole("listitem")).toHaveCount(2); + await expect(workers).toContainText("health unknown"); + const failures = page.getByRole("list", { name: "Failure groups" }); + await expect(failures.getByRole("listitem")).toHaveCount(2); + await expect(failures.getByRole("listitem").first()).toContainText("×43"); await page.screenshot({ path: `${artifacts}/overview-desktop.png`, fullPage: true, @@ -43,9 +70,7 @@ test("busy overview communicates current work and opens targeted drilldowns", as await page.getByRole("link", { name: "worker-eu-01" }).click(); await expect(page).toHaveURL(/runtime\/00000000/); await page.goBack(); - await expect( - page.getByRole("heading", { name: "Work in motion" }), - ).toBeVisible(); + await expect(page.getByRole("heading", { name: "Overview" })).toBeVisible(); } expect(fixture.mutations).toEqual([]); expect(errors).toEqual([]); @@ -57,7 +82,9 @@ test("queue filters and core controls use only the synthetic API", async ({ const fixture = await mockOperations(page); await page.goto("/queues"); await page.getByRole("searchbox", { name: "Find a queue" }).fill("invoice"); - await expect(page.locator(".pressure-table tbody tr")).toHaveCount(1); + const rows = page.getByRole("table", { name: "Queues" }).locator("tbody tr"); + await expect(rows).toHaveCount(1); + await expect(rows).toContainText("max 64"); await page.getByRole("button", { name: "Resume", exact: true }).click(); await expect( page.getByRole("button", { name: "Pause", exact: true }), @@ -74,16 +101,24 @@ test("queue filters and core controls use only the synthetic API", async ({ expect(fixture.mutations).toHaveLength(2); await page.getByRole("button", { name: "Drain…" }).click(); await page.getByRole("button", { name: "Drain queue", exact: true }).click(); - await expect(page.locator(".pressure-table tbody tr")).toContainText( - "0 scheduled", - ); + await expect(rows.locator("td").nth(1)).toHaveText("0"); + await expect(page).toHaveURL(/q=invoice/); expect(fixture.mutations).toContain("POST /queues/invoice-export/drain"); await page.getByRole("searchbox").fill("not-a-real-queue"); await expect(page.getByText("No queues match these filters.")).toBeVisible(); await page.getByRole("searchbox").fill(""); - await page.getByRole("combobox", { name: "Sort by" }).selectOption("failed"); - await expect(page.locator(".pressure-table tbody tr").first()).toContainText( - "email-delivery", + await page + .getByRole("combobox", { name: "Sort queues" }) + .selectOption("failed"); + await expect(rows.first()).toContainText("email-delivery"); + // The drain above emptied the paused queue, so only the stalled and the + // falling-behind queue still need attention. + await page.getByText("Needs attention only").click(); + await expect(rows).toHaveCount(2); + await page.reload(); + await expect(rows).toHaveCount(2); + await expect(page.getByRole("combobox", { name: "Sort queues" })).toHaveValue( + "failed", ); await page.setViewportSize({ width: 1440, height: 1050 }); await page.screenshot({ @@ -108,7 +143,9 @@ test("mobile navigation, controls, job drilldown and layout at narrow sizes", as await mockOperations(page); await page.setViewportSize({ width: 390, height: 844 }); await page.goto("/"); - await expect(page.locator(".work-counter").first()).toContainText("237,192"); + await expect(page.getByTestId("verdict")).toHaveText( + "1 queue not being worked", + ); await page.screenshot({ path: `${artifacts}/overview-mobile.png`, fullPage: true, @@ -124,9 +161,7 @@ test("mobile navigation, controls, job drilldown and layout at narrow sizes", as await expect(page.getByRole("dialog")).toHaveCount(0); await page.getByRole("button", { name: "Toggle Sidebar" }).click(); await page.getByRole("link", { name: "Overview", exact: true }).click(); - await expect( - page.getByRole("heading", { name: "Work in motion" }), - ).toBeVisible(); + await expect(page.getByRole("heading", { name: "Overview" })).toBeVisible(); await expect(page.getByRole("dialog")).toHaveCount(0); } await page.goto("/queues"); @@ -162,7 +197,7 @@ test("mobile navigation, controls, job drilldown and layout at narrow sizes", as await page.goto(route); const heading = route === "/" - ? "Work in motion" + ? "Overview" : route === "/queues" ? "Queues" : route === "/jobs" @@ -192,8 +227,8 @@ test("mobile navigation, controls, job drilldown and layout at narrow sizes", as page.getByRole("link", { name: "email-delivery", exact: true }), ).toBeVisible(); if (route === "/") - await expect(page.locator(".work-counter").first()).toContainText( - "237,192", + await expect(page.getByTestId("verdict")).toHaveText( + "1 queue not being worked", ); await page.evaluate(() => document.fonts.ready); if ( @@ -233,7 +268,9 @@ for (const scenario of ["empty", "error", "loading"] as const) { ).toBeVisible(); if (scenario === "error") { await expect(page.getByRole("alert")).toBeVisible({ timeout: 15000 }); - await expect(page.locator(".work-counter").first()).toContainText("—"); + await expect( + page.getByRole("navigation", { name: "Job states" }), + ).toContainText("Running—"); } if (scenario === "loading") await expect(page.getByText("Loading snapshots…")).toBeVisible(); @@ -249,13 +286,14 @@ test("failed refresh retains known values and labels them out of date", async ({ }) => { const fixture = await mockOperations(page); await page.goto("/"); - await expect(page.locator(".work-counter").first()).toContainText("237,192"); + const verdict = page.getByTestId("verdict"); + await expect(verdict).toHaveText("1 queue not being worked"); fixture.setReadFailure(true); await page.getByRole("button", { name: "Refresh data", exact: true }).click(); await expect(page.getByText(/Refresh failed ·/)).toBeVisible({ timeout: 15000, }); - await expect(page.locator(".work-counter").first()).toContainText("237,192"); + await expect(verdict).toHaveText("1 queue not being worked"); await page.screenshot({ path: `${artifacts}/stale-desktop.png`, fullPage: true, @@ -268,7 +306,7 @@ test("failed refresh retains known values and labels them out of date", async ({ test("read-only controls remain disabled and dark mode stays legible", async ({ page, }) => { - const fixture = await mockOperations(page, "busy", true); + const fixture = await mockOperations(page, "incident", true); await page.addInitScript(() => localStorage.setItem("theme", "dark")); await page.setViewportSize({ width: 1440, height: 1050 }); await page.goto("/"); @@ -310,7 +348,7 @@ test("mobile drawer closes promptly with both motion preferences", async ({ ); await expect( page.getByRole("heading", { - name: name === "Overview" ? "Work in motion" : name, + name, exact: true, }), ).toBeVisible(); @@ -323,3 +361,75 @@ test("mobile drawer closes promptly with both motion preferences", async ({ ).toBeFocused(); } }); + +test("normal operations, an incident, and the recovery each read at a glance", async ({ + page, +}) => { + await page.setViewportSize({ width: 1440, height: 1000 }); + + await mockOperations(page, "normal"); + await page.goto("/"); + await expect(page.getByTestId("verdict")).toHaveText("Keeping up"); + await expect( + page.getByRole("table", { name: "Queues needing attention" }), + ).toHaveCount(0); + await expect( + page.getByRole("button", { name: "Show 24 queues keeping up" }), + ).toBeVisible(); + await expect( + page.getByText("All 48 reporting instances are healthy."), + ).toBeVisible(); + await expect(page.getByRole("list", { name: "Fleet changes" })).toHaveCount( + 0, + ); + await page.screenshot({ path: `${artifacts}/phase-normal-desktop.png` }); + + await page.unrouteAll({ behavior: "ignoreErrors" }); + await mockOperations(page, "incident"); + await page.goto("/"); + await expect(page.getByTestId("verdict")).toHaveText( + "1 queue not being worked", + ); + await expect( + page + .getByRole("table", { name: "Queues needing attention" }) + .locator("tbody tr") + .first(), + ).toContainText("image-processing"); + await page.screenshot({ path: `${artifacts}/phase-incident-desktop.png` }); + await page.goto("/queues/email-delivery"); + await expect(page.getByText("Limited by: Rate limit")).toBeVisible(); + await page.screenshot({ + path: `${artifacts}/phase-incident-queue-desktop.png`, + }); + + await page.unrouteAll({ behavior: "ignoreErrors" }); + await mockOperations(page, "recovery"); + await page.goto("/"); + await expect(page.getByTestId("verdict")).toHaveText("Recovering"); + await expect(page.getByText(/all clear in about 7m/)).toBeVisible(); + await expect(page.getByText("Recovering (3)", { exact: true })).toBeVisible(); + const draining = page + .getByRole("table", { name: "Queues needing attention" }) + .locator("tbody tr"); + await expect(draining).toHaveCount(3); + await expect(draining.first()).toContainText("email-delivery"); + await expect(draining.first()).toContainText("−8,700/min"); + await expect( + page.getByRole("list", { name: "Fleet changes" }).getByRole("listitem"), + ).toHaveCount(5); + await expect(page.getByText("Versions 0.7.1, 0.7.2 → 0.7.2")).toBeVisible(); + await page.screenshot({ path: `${artifacts}/phase-recovery-desktop.png` }); + + await page.setViewportSize({ width: 390, height: 844 }); + await page.unrouteAll({ behavior: "ignoreErrors" }); + await mockOperations(page, "incident"); + await page.goto("/"); + await expect(page.getByTestId("verdict")).toHaveText( + "1 queue not being worked", + ); + await page.screenshot({ + path: `${artifacts}/phase-incident-mobile.png`, + fullPage: true, + }); +}); diff --git a/awa-ui/frontend/e2e/queues.spec.ts b/awa-ui/frontend/e2e/queues.spec.ts index a5c8be72b..38aedf464 100644 --- a/awa-ui/frontend/e2e/queues.spec.ts +++ b/awa-ui/frontend/e2e/queues.spec.ts @@ -26,16 +26,18 @@ test.describe("Queues page", () => { await loadQueuesPage(page); // Desktop table (visible at default viewport) - const queueTable = page.getByRole("table", { name: "Queue pressure" }); + const queueTable = page.getByRole("table", { name: "Queues" }); await expect(queueTable).toBeVisible(); for (const header of [ "Queue", - "Available", - "Oldest available", + "Status", + "Waiting", + "Trend", + "Oldest wait", "Running", - "Retryable / failed", - "Completed last hour", + "Done/min", + "Retrying / failed", "Controls", ]) { await expect( @@ -43,6 +45,16 @@ test.describe("Queues page", () => { ).toBeVisible(); } + // Capacity joins the table only where there is room for it. + const capacity = queueTable.getByRole("columnheader", { + name: "Capacity", + exact: true, + }); + await expect(capacity).toBeHidden(); + await page.setViewportSize({ width: 1920, height: 1080 }); + await expect(capacity).toBeVisible(); + await page.setViewportSize({ width: 1280, height: 720 }); + // Descriptor-backed queue shows display name and raw key await expect(queueTable.getByText("E2E Queue")).toBeVisible(); await expect(queueTable.getByText("e2e_test")).toBeVisible(); @@ -51,20 +63,20 @@ test.describe("Queues page", () => { ).toBeVisible(); }); - test("queue table shows descriptor description", async ({ page }) => { + test("queue table carries the descriptor description", async ({ page }) => { await loadQueuesPage(page); - const queueTable = page.getByRole("table", { name: "Queue pressure" }); + const queueTable = page.getByRole("table", { name: "Queues" }); await expect( - queueTable.getByText("End-to-end queue used for UI coverage"), - ).toBeVisible(); + queueTable.getByRole("link", { name: "E2E Queue" }), + ).toHaveAttribute("title", "End-to-end queue used for UI coverage"); }); test("queue table surfaces available wait and controls", async ({ page }) => { await loadQueuesPage(page); - const queueTable = page.getByRole("table", { name: "Queue pressure" }); + const queueTable = page.getByRole("table", { name: "Queues" }); await expect( - queueTable.getByRole("columnheader", { name: "Oldest available" }), + queueTable.getByRole("columnheader", { name: "Oldest wait" }), ).toBeVisible(); await expect( queueTable.getByRole("columnheader", { name: "Controls" }), @@ -135,7 +147,7 @@ test.describe("Queues page", () => { await page.goto("/queues"); - const queueTable = page.getByRole("table", { name: "Queue pressure" }); + const queueTable = page.getByRole("table", { name: "Queues" }); await expect( queueTable.getByRole("rowheader", { name: /DLQ Off Queue/ }), ).toBeVisible(); @@ -147,7 +159,7 @@ test.describe("Queues page", () => { test("click queue navigates to queue detail", async ({ page }) => { await loadQueuesPage(page); - const queueTable = page.getByRole("table", { name: "Queue pressure" }); + const queueTable = page.getByRole("table", { name: "Queues" }); const queueLink = queueTable.getByRole("link", { name: "E2E Queue" }); await expect(queueLink).toBeVisible(); @@ -313,18 +325,18 @@ test.describe("Queues page", () => { }); await page.goto("/queues"); - const queueTable = page.getByRole("table", { name: "Queue pressure" }); + const queueTable = page.getByRole("table", { name: "Queues" }); await expect( queueTable.getByText("Descriptor stale", { exact: true }), ).toBeVisible(); await expect( - queueTable.getByText("Descriptor drift across live runtimes"), + queueTable.getByText("Descriptor drift", { exact: true }), ).toBeVisible(); }); function queueRow(page: Page) { return page - .getByRole("table", { name: "Queue pressure" }) + .getByRole("table", { name: "Queues" }) .getByRole("row", { name: /E2E Queue/ }); } diff --git a/awa-ui/frontend/e2e/readme-images.spec.ts b/awa-ui/frontend/e2e/readme-images.spec.ts new file mode 100644 index 000000000..35e80a50a --- /dev/null +++ b/awa-ui/frontend/e2e/readme-images.spec.ts @@ -0,0 +1,24 @@ +import { test, expect } from "@playwright/test"; +import { mockOperations, type Phase } from "./operations-fixture"; + +// Regenerates the README screenshots from the synthetic fleet: +// AWA_README_IMAGES_DIR=../../docs/images npm run test:operations -- readme-images +const out = process.env.AWA_README_IMAGES_DIR; + +test.use({ viewport: { width: 1440, height: 1180 }, deviceScaleFactor: 2 }); + +for (const [phase, verdict] of [ + ["incident", "1 queue not being worked"], + ["recovery", "Recovering"], +] as [Phase, string][]) { + test(`README image: overview during ${phase}`, async ({ page }) => { + test.skip(!out, "set AWA_README_IMAGES_DIR to regenerate README images"); + await page.addInitScript(() => localStorage.setItem("theme", "dark")); + await mockOperations(page, phase); + await page.goto("/"); + await expect(page.getByTestId("verdict")).toHaveText(verdict); + await expect(page.locator("tbody tr").first()).toBeVisible(); + await page.evaluate(() => document.fonts.ready); + await page.screenshot({ path: `${out}/awa-ui-overview-${phase}-dark.png` }); + }); +} diff --git a/awa-ui/frontend/playwright.config.ts b/awa-ui/frontend/playwright.config.ts index 916333d72..8eb0ccd1f 100644 --- a/awa-ui/frontend/playwright.config.ts +++ b/awa-ui/frontend/playwright.config.ts @@ -11,7 +11,7 @@ const databaseUrl = export default defineConfig({ globalSetup: "./e2e/seed.ts", testDir: "./e2e", - testIgnore: "**/operations.spec.ts", + testIgnore: ["**/operations.spec.ts", "**/readme-images.spec.ts"], fullyParallel: true, forbidOnly: !!process.env.CI, retries: process.env.CI ? 2 : 0, diff --git a/awa-ui/frontend/playwright.operations.config.ts b/awa-ui/frontend/playwright.operations.config.ts index 748a3adf2..553278579 100644 --- a/awa-ui/frontend/playwright.operations.config.ts +++ b/awa-ui/frontend/playwright.operations.config.ts @@ -1,7 +1,7 @@ import { defineConfig } from "@playwright/test"; export default defineConfig({ testDir: "./e2e", - testMatch: "operations.spec.ts", + testMatch: /(operations|readme-images)\.spec\.ts/, workers: 1, forbidOnly: !!process.env.CI, reporter: "list", diff --git a/awa-ui/frontend/src/__tests__/operations.test.ts b/awa-ui/frontend/src/__tests__/operations.test.ts new file mode 100644 index 000000000..3e787a226 --- /dev/null +++ b/awa-ui/frontend/src/__tests__/operations.test.ts @@ -0,0 +1,237 @@ +import { describe, it, expect } from "vitest"; +import type { + HealthReport, + JobRow, + QueueAssessment, + QueueHistory, + QueueStats, +} from "@/lib/api"; +import { + eventIndexes, + formatDuration, + formatRate, + groupFailures, + joinQueues, + totalBacklog, + totalRate, +} from "@/lib/operations"; + +function queue(name: string, available = 10): QueueStats { + return { + queue: name, + display_name: null, + description: null, + owner: null, + docs_url: null, + tags: [], + extra: {}, + descriptor_last_seen_at: null, + descriptor_stale: false, + descriptor_mismatch: false, + total_queued: available, + scheduled: 0, + available, + retryable: 0, + running: 1, + failed: 0, + waiting_external: 0, + completed_last_hour: 600, + lag_seconds: 5, + paused: false, + }; +} + +function assessment( + name: string, + rank: number, + flow: Partial = {}, + liveWorkers = 1, +): QueueAssessment { + return { + queue: name, + health: "steady", + label: "Steady", + reason: "Backlog holding level", + needs_attention: false, + rank, + limited_by: null, + trend: null, + clear_seconds: null, + flow: { + claimed_per_minute: null, + completed_per_minute: null, + failed_per_minute: null, + retried_per_minute: null, + arrivals_per_minute: null, + mean_claim_wait_seconds: null, + rate_limited: false, + ...flow, + }, + live_workers: liveWorkers, + in_flight: 0, + capacity: null, + }; +} + +function report(queues: QueueAssessment[]): HealthReport { + return { + status: "ok", + title: "Keeping up", + detail: "", + generated_at: "2026-10-25T00:00:00Z", + history_available: true, + history_since: null, + queues, + }; +} + +function history( + queues: Record, + events: QueueHistory["events"] = [], +): QueueHistory { + const length = Object.values(queues)[0]?.length ?? 3; + const empty = () => Array.from({ length }, () => null); + return { + installed: true, + bucket_secs: 60, + minutes: length, + observed_since: null, + buckets: Array.from( + { length }, + (_, index) => `2026-10-25T00:0${index}:00Z`, + ), + queues: Object.fromEntries( + Object.entries(queues).map(([name, available]) => [ + name, + { + available, + running: empty(), + lag_seconds: empty(), + claimed: empty(), + completed: empty(), + failed_attempts: empty(), + rate_limited: empty(), + }, + ]), + ), + fleet: { live_instances: empty(), healthy_instances: empty() }, + events, + }; +} + +describe("joinQueues", () => { + it("follows the report's risk order and appends queues it does not know", () => { + const rows = joinQueues( + [queue("a"), queue("new"), queue("b")], + report([assessment("b", 0), assessment("a", 6)]), + history({ a: [1, 2, 3] }), + ); + expect(rows.map((row) => row.queue.queue)).toEqual(["b", "a", "new"]); + expect(rows[1]?.series?.available).toEqual([1, 2, 3]); + expect(rows[2]?.assessment).toBeUndefined(); + }); +}); + +describe("totalBacklog", () => { + it("sums per bucket and keeps buckets with no samples null", () => { + expect( + totalBacklog(history({ a: [null, 1, 2], b: [null, null, 3] })), + ).toEqual([null, 1, 5]); + }); +}); + +describe("totalRate", () => { + it("sums known rates", () => { + const total = totalRate( + report([ + assessment("a", 0, { completed_per_minute: 10 }), + assessment("b", 0, { completed_per_minute: 5 }), + ]), + (flow) => flow.completed_per_minute, + ); + expect(total).toBe(15); + }); + + it("refuses a partial total while a served queue is unknown", () => { + const total = totalRate( + report([ + assessment("a", 0, { completed_per_minute: 10 }), + assessment("b", 0, {}, 2), + ]), + (flow) => flow.completed_per_minute, + ); + expect(total).toBeNull(); + }); + + it("ignores unknown rates on queues nobody serves", () => { + const total = totalRate( + report([ + assessment("a", 0, { completed_per_minute: 10 }), + assessment("orphan", 0, {}, 0), + ]), + (flow) => flow.completed_per_minute, + ); + expect(total).toBe(10); + }); +}); + +describe("eventIndexes", () => { + it("maps fleet events onto bucket positions", () => { + const indexes = eventIndexes( + history({ a: [1, 2, 3] }, [ + { at: "2026-10-25T00:02:00Z", kind: "version_change", detail: "" }, + { at: "2026-10-24T23:00:00Z", kind: "leader_change", detail: "" }, + ]), + ); + expect(indexes).toEqual([2]); + }); +}); + +describe("groupFailures", () => { + function failed( + id: number, + message: string, + finalized: string, + kind = "SendEmail", + ): JobRow { + return { + id, + kind, + queue: "email", + errors: [{ error: message }], + finalized_at: finalized, + } as unknown as JobRow; + } + + it("groups by kind, queue and error, masking per-job identifiers", () => { + const groups = groupFailures([ + failed(1, "timed out calling order 18273645", "2026-10-25T00:00:01Z"), + failed(2, "timed out calling order 99887766", "2026-10-25T00:00:03Z"), + failed( + 3, + "timed out calling order 12345678", + "2026-10-25T00:00:02Z", + "Other", + ), + ]); + expect(groups).toHaveLength(2); + expect(groups[0]).toMatchObject({ kind: "SendEmail", count: 2 }); + expect(groups[0]!.latest.id).toBe(2); + expect(groups[0]!.message).toBe("timed out calling order 99887766"); + }); +}); + +describe("formatting", () => { + it("formats durations consistently across units", () => { + expect(formatDuration(2.4)).toBe("2s"); + expect(formatDuration(458)).toBe("7m 38s"); + expect(formatDuration(3720)).toBe("1h 2m"); + expect(formatDuration(null)).toBe("—"); + }); + + it("formats signed rates", () => { + expect(formatRate(1234, true)).toBe("+1,234/min"); + expect(formatRate(-0.5, true)).toBe("−0.5/min"); + expect(formatRate(52_000)).toBe("52K/min"); + }); +}); diff --git a/awa-ui/frontend/src/app.css b/awa-ui/frontend/src/app.css index 28486f5f6..0e4a424c0 100644 --- a/awa-ui/frontend/src/app.css +++ b/awa-ui/frontend/src/app.css @@ -308,130 +308,9 @@ If you want to use the typography plugin (prose), uncomment the line below. transform: translateZ(0); } -/* Operator views */ -.operations-overview { display: grid; gap: 24px; min-width: 0; } -.page-intro { display: flex; justify-content: space-between; align-items: center; gap: 16px; } -.page-intro h1 { font-size: clamp(26px, 3vw, 34px); line-height: 1.2; font-weight: 650; letter-spacing: -.045em; margin: 6px 0 8px; } -.page-intro p:not(.eyebrow) { color: var(--muted-fg); font-size: 14px; } -.eyebrow { font-size: 11px; text-transform: uppercase; letter-spacing: .12em; font-weight: 650; color: var(--muted-fg); } -.operator-link { color: var(--primary); font-size: 13px; font-weight: 600; white-space: nowrap; display: inline-flex; align-items: center; min-height: 44px; } -.snapshot-status { display: flex; gap: 8px; align-items: center; color: var(--muted-fg); font-size: 12px; margin-top: -10px; } -.status-dot { flex: 0 0 6px; height: 6px; border-radius: 100%; background: currentColor; } -.snapshot-warning { color: var(--danger-subtle-fg); } -.source-error { padding: 14px 18px; border: 1px solid var(--danger); background: var(--danger-subtle); color: var(--danger-subtle-fg); border-radius: 8px; font-size: 14px; } -.flow-summary { display: grid; grid-template-columns: 1.3fr 1fr; border: 1px solid var(--border); border-radius: 12px; overflow: hidden; } -.completion-summary { padding: 24px; background: var(--primary-subtle); } -.completion-summary .eyebrow { color: var(--primary-subtle-fg); } -.completion-value { display: block; font-size: clamp(32px, 4vw, 44px); font-weight: 600; letter-spacing: -.045em; font-variant-numeric: tabular-nums; margin: 8px 0; line-height: 1.2; } -.completion-value span { font-size: 15px; letter-spacing: 0; font-weight: 400; color: var(--muted-fg); } -.completion-summary > p:last-child { font-size: 12px; color: var(--muted-fg); max-width: 400px; line-height: 1.6; } -.attention-summary { padding: 20px 24px; background: var(--bg); } -.attention-summary > p { margin-bottom: 8px; } -.attention-summary > a { display: grid; grid-template-columns: minmax(42px, auto) 1fr 16px; align-items: center; gap: 12px; min-height: 40px; font-size: 13px; } -.attention-summary > a:hover { color: var(--primary); } -.attention-summary strong { font-variant-numeric: tabular-nums; font-size: 18px; font-weight: 600; } -.attention-summary > a > span:last-child { color: var(--muted-fg); } -.section-label { display: flex; justify-content: space-between; gap: 12px; margin-bottom: 12px; align-items: baseline; } -.section-label h2, .panel-heading h2 { font-weight: 600; font-size: 16px; letter-spacing: -.02em; } -.section-label > span { font-size: 12px; color: var(--muted-fg); } -.work-counters { display: grid; grid-template-columns: repeat(6, minmax(0, 1fr)); border: 1px solid var(--border); border-radius: 10px; overflow: hidden; background: var(--bg); } -.work-counter { padding: 16px; display: flex; flex-direction: column; gap: 7px; border-right: 1px solid var(--border); min-width: 0; } -.work-counter:last-child { border: 0; } -.work-counter:hover { background: var(--muted); } -.work-counter > span { font-size: 12px; font-weight: 550; } -.work-counter strong { font-size: clamp(18px, 1.8vw, 26px); font-weight: 600; letter-spacing: -.04em; font-variant-numeric: tabular-nums; } -.work-counter small { color: var(--muted-fg); font-size: 11px; line-height: 1.5; } -.counter-failed strong, .failure-value { color: var(--danger-subtle-fg); } -.counter-retryable strong, .retry-value { color: var(--warning-subtle-fg); } -.operations-panel { min-width: 0; border: 1px solid var(--border); border-radius: 10px; background: var(--bg); overflow: hidden; } -.panel-heading { padding: 18px 20px; display: flex; align-items: center; justify-content: space-between; gap: 12px; border-bottom: 1px solid var(--border); } -.panel-heading p { font-size: 12px; color: var(--muted-fg); margin-top: 5px; line-height: 1.5; } -.panel-footnote { padding: 12px 20px; border-top: 1px solid var(--border); font-size: 11px; line-height: 1.6; color: var(--muted-fg); background: var(--muted); } -.panel-empty { padding: 32px 20px; color: var(--muted-fg); font-size: 14px; } -.queue-board { overflow-x: auto; } -.pressure-table { width: 100%; border-collapse: collapse; text-align: right; font-size: 13px; } -.pressure-table thead { background: var(--muted); } -.pressure-table thead th { font-size: 11px; color: var(--muted-fg); font-weight: 500; white-space: nowrap; } -.pressure-table thead th span { display: block; } -.pressure-table th, .pressure-table td { padding: 14px 16px; font-variant-numeric: tabular-nums; vertical-align: middle; } -.pressure-table th:first-child { text-align: left; } -.pressure-table tbody tr + tr { border-top: 1px solid var(--border); } -.pressure-table tbody tr:hover { background: var(--muted); } -.pressure-table td { white-space: nowrap; } -.pressure-table td a { display: inline-flex; align-items: center; min-height: 32px; } -.pressure-table a:hover { text-decoration: underline; text-underline-offset: 3px; } -.queue-identity { min-width: 200px; max-width: 330px; font-weight: 500; overflow-wrap: anywhere; } -.queue-identity > a { color: var(--primary); display: inline-flex; align-items: center; min-height: 32px; } -.queue-identity code { display: block; color: var(--muted-fg); font-size: 11px; } -.queue-context { display: flex; align-items: center; flex-wrap: wrap; gap: 6px; color: var(--muted-fg); font-size: 11px; font-weight: 400; line-height: 1.6; } -.signal { border-radius: 4px; font-size: 10px; font-weight: 600; padding: 1px 5px; } -.signal-warning { background: var(--warning-subtle); color: var(--warning-subtle-fg); } -.wait-value { font-weight: 600; } -.queue-action-buttons { display: flex; justify-content: flex-end; gap: 4px; } -.overview-details { display: grid; grid-template-columns: 1fr 1fr; gap: 24px; align-items: start; } -.worker-preview-row { display: flex; align-items: center; justify-content: space-between; padding: 16px 20px; gap: 12px; font-size: 12px; } -.worker-preview-row + .worker-preview-row, .failure-preview-row + .failure-preview-row { border-top: 1px solid var(--border); } -.worker-preview-row:hover, .failure-preview-row:hover { background: var(--muted); } -.worker-preview-row strong { display: block; font-size: 13px; font-weight: 550; margin-bottom: 4px; } -.worker-preview-row > span { text-align: right; color: var(--muted-fg); white-space: nowrap; } -.worker-preview-row > div { min-width: 0; overflow-wrap: anywhere; } -.worker-role { display: block; font-size: 11px; margin-top: 4px; } -.concern-text { color: var(--danger-subtle-fg); } -.healthy-text { color: var(--primary-subtle-fg); } -.failure-preview-row { display: block; padding: 14px 20px; font-size: 12px; overflow-wrap: anywhere; } -.failure-preview-row strong { font-size: 13px; font-weight: 550; } -.failure-preview-row div > span { display: block; color: var(--muted-fg); font-size: 11px; margin-top: 3px; } -.failure-preview-row p { color: var(--danger-subtle-fg); margin-top: 7px; } -.failure-preview-row > span { display: block; font-size: 11px; color: var(--muted-fg); margin-top: 7px; } -.queue-filters { display: flex; flex-wrap: wrap; align-items: end; gap: 16px; } -.queue-filters label { display: grid; gap: 6px; font-size: 12px; font-weight: 500; } -.queue-filters input[type=search], .queue-filters select { border: 1px solid var(--border); background: var(--bg); color: var(--fg); border-radius: 7px; padding: 10px 12px; font-size: 14px; min-height: 44px; } -.queue-filters input[type=search] { min-width: 260px; } -.queue-filters .review-toggle { display: flex; align-items: center; gap: 8px; min-height: 44px; } -.queue-filters input[type=checkbox] { width: 18px; height: 18px; accent-color: var(--primary); } -.operations-overview a:focus-visible, .operations-overview input:focus-visible, .operations-overview select:focus-visible { outline: 2px solid var(--primary); outline-offset: 3px; } -@media (max-width: 1100px) { - .work-counters { grid-template-columns: repeat(3, minmax(0, 1fr)); } - .work-counter:nth-child(-n+3) { border-bottom: 1px solid var(--border); } - .work-counter:nth-child(3n) { border-right: 0; } - .overview-details { grid-template-columns: 1fr; } -} -@media (max-width: 1023px) { - .queue-board { overflow: visible; } - .pressure-table, .pressure-table tbody { display: block; } - .pressure-table thead { position: absolute; width: 1px; height: 1px; overflow: hidden; clip-path: inset(50%); } - .pressure-table tbody tr { display: grid; grid-template-columns: 1fr 1fr; padding: 12px 16px; gap: 8px 12px; } - .pressure-table th, .pressure-table td { padding: 0; display: block; text-align: left; } - .pressure-table .queue-identity { grid-column: 1 / -1; max-width: none; min-width: 0; margin-bottom: 4px; } - .pressure-table .queue-identity > a { min-height: 44px; } - .pressure-table td::before { content: attr(data-label); display: block; font-size: 11px; font-weight: 400; color: var(--muted-fg); margin-bottom: 2px; } - .pressure-table td a { min-height: 44px; } - .pressure-table .queue-controls { grid-column: 1 / -1; } - .pressure-table .queue-controls::before { content: none; } - .queue-action-buttons { justify-content: flex-start; margin-top: 6px; } -} +/* Touch targets on phones: header controls and in-page buttons stay at + least 44px, and long unbroken identifiers wrap instead of widening the page. */ @media (max-width: 767px) { - .flow-summary { grid-template-columns: 1fr; } - .completion-summary { padding: 20px; } - .attention-summary { border-top: 1px solid var(--border); padding: 16px 20px; } - .attention-summary > a { min-height: 44px; } - .operations-overview { gap: 20px; } - .page-intro { align-items: start; } - .page-intro > .operator-link { display: none; } - .section-label { flex-direction: column; gap: 3px; } - .work-counters { grid-template-columns: repeat(2, minmax(0, 1fr)); } - .work-counter { padding: 14px; } - .work-counter:nth-child(3n) { border-right: 1px solid var(--border); } - .work-counter:nth-child(2n) { border-right: 0; } - .work-counter:nth-child(-n+4) { border-bottom: 1px solid var(--border); } - .work-counter strong { font-size: 25px; } - .panel-heading { padding: 16px; align-items: start; } - .panel-footnote { padding: 12px 16px; } - .worker-preview-row, .failure-preview-row { padding: 14px 16px; } - .snapshot-cadence { display: block; } - .queue-filters { gap: 12px; } - .queue-filters > label:not(.review-toggle) { flex: 1 1 100%; } - .queue-filters input[type=search] { min-width: 0; width: 100%; } [data-slot=sidebar-trigger], header button, main button { min-height: 44px; min-width: 44px; } [data-slot=sidebar-item] { min-height: 44px; } main { overflow-wrap: anywhere; } @@ -440,4 +319,3 @@ If you want to use the typography plugin (prose), uncomment the line below. *, *::before, *::after { scroll-behavior: auto !important; animation: none !important; transition: none !important; } } -.dark .operator-link, .dark .queue-identity > a { color: var(--primary-subtle-fg); } diff --git a/awa-ui/frontend/src/components/HistorySpark.tsx b/awa-ui/frontend/src/components/HistorySpark.tsx new file mode 100644 index 000000000..cd64cf9d0 --- /dev/null +++ b/awa-ui/frontend/src/components/HistorySpark.tsx @@ -0,0 +1,108 @@ +/** + * Sparkline for sampled queue history. Unlike `Sparkline`, points keep their + * bucket position (so a gap in sampling shows as a gap, not a shorter line) + * and the y axis starts at zero: a large, steady backlog stays flat instead + * of min-max scaling turning a 1% wobble into a cliff. + */ +export function HistorySpark({ + data, + width = 120, + height = 24, + label, + className, + markers = [], +}: { + data: ReadonlyArray; + width?: number; + height?: number; + label: string; + className?: string; + /** Bucket indexes to mark with a vertical tick, e.g. fleet changes. */ + markers?: ReadonlyArray; +}) { + const values = data.filter((value): value is number => value !== null); + if (values.length === 0) { + return ( + + ); + } + const max = Math.max(...values) || 1; + const step = data.length > 1 ? width / (data.length - 1) : 0; + const pad = 1.5; + const y = (value: number) => pad + (1 - value / max) * (height - pad * 2); + + const segments: string[][] = []; + let current: string[] = []; + data.forEach((value, index) => { + if (value === null) { + if (current.length) segments.push(current); + current = []; + return; + } + current.push(`${(index * step).toFixed(1)},${y(value).toFixed(1)}`); + }); + if (current.length) segments.push(current); + + return ( + + + {markers.map((index) => ( + + ))} + {segments.map((points, index) => + points.length === 1 ? ( + + ) : ( + + ), + )} + + ); +} diff --git a/awa-ui/frontend/src/components/QueueFlowTable.tsx b/awa-ui/frontend/src/components/QueueFlowTable.tsx new file mode 100644 index 000000000..024070617 --- /dev/null +++ b/awa-ui/frontend/src/components/QueueFlowTable.tsx @@ -0,0 +1,504 @@ +import type { ReactNode } from "react"; +import { Link } from "@tanstack/react-router"; +import type { + QueueHealthKind, + QueueRuntimeSummary, + QueueStats, +} from "@/lib/api"; +import { Badge } from "@/components/ui/badge"; +import { HistorySpark } from "@/components/HistorySpark"; +import { + capacityLabel, + formatCount, + formatDuration, + formatRate, + healthOf, + LIMITED_BY_HINT, + LIMITED_BY_LABEL, + type AssessedQueue, +} from "@/lib/operations"; + +export const HEALTH_BADGE: Record< + QueueHealthKind, + "danger" | "warning" | "info" | "secondary" | "outline" +> = { + stalled: "danger", + behind: "warning", + slow: "warning", + paused: "secondary", + draining: "info", + unknown: "outline", + steady: "outline", + idle: "outline", +}; + +/** Text tint for the backlog sparkline, so the trend reads at a glance. */ +export const SPARK_TONE: Record = { + stalled: "text-danger", + behind: "text-warning", + slow: "text-warning", + paused: "text-muted-fg", + draining: "text-info", + unknown: "text-muted-fg", + steady: "text-muted-fg", + idle: "text-muted-fg", +}; + +function queueName(queue: QueueStats): string { + return queue.display_name?.trim() || queue.queue; +} + +function JobsLink({ + queue, + state, + value, + className = "", +}: { + queue: string; + state: string; + value: number; + className?: string; +}) { + return ( + + {value.toLocaleString()} + + ); +} + +function Capacity({ summary }: { summary: QueueRuntimeSummary | undefined }) { + if (summary && summary.live_instances === 0) { + return No live workers; + } + const title = summary + ? `${summary.healthy_instances}/${summary.live_instances} live instances healthy${summary.config_mismatch ? " · config differs between instances" : ""}` + : "No runtime has reported this queue"; + return ( + + {capacityLabel(summary)} + + ); +} + +/** + * Rare conditions worth seeing without opening the queue. They sit by the + * name rather than in a column so they survive when narrow tables drop one. + */ +function QueueFlags({ + queue, + summary, +}: { + queue: QueueStats; + summary: QueueRuntimeSummary | undefined; +}) { + return ( + <> + {summary?.config_mismatch && ( + + Config mismatch + + )} + {summary?.config?.dlq_enabled === false && ( + + DLQ off + + )} + {queue.descriptor_stale && ( + + Descriptor stale + + )} + {queue.descriptor_mismatch && ( + + Descriptor drift + + )} + + ); +} + +function TrendText({ row }: { row: AssessedQueue }) { + const trend = row.assessment?.trend; + if (!trend) return —; + // Below the noise threshold the fitted slope is batching jitter; printing + // "-0.6/min" would invite reading meaning into it. + const health = healthOf(row); + if (health === "steady" || health === "idle" || health === "paused") { + return flat; + } + const clear = row.assessment?.clear_seconds ?? null; + return ( + + {formatRate(trend.per_minute, true)} + {clear !== null && ( + · ~{formatDuration(clear)} + )} + + ); +} + +function LimitedBy({ row }: { row: AssessedQueue }) { + const limited = row.assessment?.limited_by; + if (!limited) return —; + const tone = + limited === "no_workers" + ? "text-danger-subtle-fg" + : limited === "rate_limit" || limited === "worker_capacity" + ? "text-warning-subtle-fg" + : ""; + return ( + + {LIMITED_BY_LABEL[limited]} + + ); +} + +/** Measured completions per minute over the last 5 minutes, or the last + * hour's average when the runtimes have not reported counters. */ +function DoneRate({ row }: { row: AssessedQueue }) { + const measured = row.assessment?.flow.completed_per_minute; + if (measured !== null && measured !== undefined) { + return ( + + {formatCount(Math.round(measured))} + + ); + } + return ( + + {formatCount(Math.round(row.queue.completed_last_hour / 60))} + + ); +} + +/** + * One line per queue, ordered by the caller. Switches to stacked cards when + * its own container is narrow (not the viewport: the docked sidebar takes a + * fixed share of a tablet screen). + */ +export function QueueFlowTable({ + rows, + controls, + runtime, + label, +}: { + rows: ReadonlyArray; + controls?: (queue: QueueStats) => ReactNode; + /** When given, adds a capacity column from live runtime config. */ + runtime?: ReadonlyMap; + label: string; +}) { + return ( +
+
+ + + + + + + + + + + + {runtime && ( + + )} + + + {controls && ( + + )} + + + + {rows.map((row) => { + const { queue, assessment } = row; + const health = healthOf(row); + return ( + + + + + + + + + + {runtime && ( + + )} + + + {controls && ( + + )} + + ); + })} + +
+ Queue + Status + Waiting + Last hour + Trend + + Oldest wait + + Running + + Limited by + + Capacity + + Done/min + + Retrying / failed + + Controls +
+ + {queueName(queue)} + + {queue.display_name?.trim() && ( + + {queue.queue} + + )} + + + + {assessment?.label ?? "…"} + + + + + + + + + {queue.available > 0 ? ( + formatDuration(queue.lag_seconds) + ) : ( + — + )} + + + + + + + + + + + + + + {controls(queue)} +
+
+ +
    + {rows.map((row) => { + const { queue, assessment } = row; + const health = healthOf(row); + return ( +
  • +
    +
    + + {queueName(queue)} + +

    {assessment?.reason}

    + {(queue.descriptor_stale || queue.descriptor_mismatch) && ( +
    + +
    + )} +
    + + {assessment?.label ?? "…"} + +
    +
    + + + + +
    +
    +
    +
    Waiting
    +
    + +
    +
    +
    +
    Oldest wait
    +
    + {queue.available > 0 + ? formatDuration(queue.lag_seconds) + : "—"} +
    +
    +
    +
    Running
    +
    + +
    +
    +
    +
    Done/min
    +
    + +
    +
    +
    +
    Retrying
    +
    + +
    +
    +
    +
    Failed
    +
    + +
    +
    +
    +
    Limited by
    +
    + +
    +
    + {runtime && ( +
    +
    Capacity
    +
    + +
    +
    + )} +
    + {controls &&
    {controls(queue)}
    } +
    • + ); + })} +
+
+ ); +} diff --git a/awa-ui/frontend/src/components/QueuePressure.tsx b/awa-ui/frontend/src/components/QueuePressure.tsx deleted file mode 100644 index f5a63fbde..000000000 --- a/awa-ui/frontend/src/components/QueuePressure.tsx +++ /dev/null @@ -1,138 +0,0 @@ -import type { ReactNode } from "react"; -import { Link } from "@tanstack/react-router"; -import type { QueueStats, QueueRuntimeSummary } from "@/lib/api"; -import { dlqPolicyLabel } from "@/components/RuntimeDisplay"; -import { availableWait } from "@/lib/operations"; - -export function QueuePressure({ - queues, - runtime, - actions, -}: { - queues: QueueStats[]; - runtime?: Map; - actions?: (queue: QueueStats) => ReactNode; -}) { - return ( -
- - - - - - - - - - {actions && } - - - - {queues.map((queue) => { - const workers = runtime?.get(queue.queue); - return ( - - - - - - - - {actions && ( - - )} - - ); - })} - -
QueueAvailableOldest availableRunningRetryable / failed - Completed last hour - Controls
- - {queue.display_name?.trim() || queue.queue} - - {queue.display_name?.trim() && {queue.queue}} -
- {queue.paused && ( - Paused - )} - - {queue.scheduled.toLocaleString()} scheduled ·{" "} - {queue.waiting_external.toLocaleString()} waiting external - -
- {runtime && queue.description && ( -

{queue.description}

- )} - {queue.descriptor_stale && ( -
- Descriptor stale -
- )} - {queue.descriptor_mismatch && ( -
- Descriptor drift across live runtimes -
- )} - {workers && ( -
- {workers.healthy_instances}/{workers.live_instances} live - instances healthy - {workers.config_mismatch && " · config mismatch"} -
- )} - {workers?.config && ( -
- {dlqPolicyLabel(workers.config)} -
- )} - 		- - {queue.available.toLocaleString()} - - - 0 ? "wait-value" : "text-muted-fg" - } - > - {availableWait(queue.lag_seconds)} - - - - {queue.running.toLocaleString()} - - - 0 ? "retry-value" : "text-muted-fg" - } - to="/jobs" - search={{ state: "retryable", q: `queue:${queue.queue}` }} - > - {queue.retryable.toLocaleString()} - - - 0 ? "failure-value" : "text-muted-fg" - } - to="/jobs" - search={{ state: "failed", q: `queue:${queue.queue}` }} - > - {queue.failed.toLocaleString()} - - - {queue.completed_last_hour.toLocaleString()} - - {actions(queue)} -
-
- ); -} diff --git a/awa-ui/frontend/src/components/SnapshotStatus.tsx b/awa-ui/frontend/src/components/SnapshotStatus.tsx index 38fec8cc2..ea8e8eebb 100644 --- a/awa-ui/frontend/src/components/SnapshotStatus.tsx +++ b/awa-ui/frontend/src/components/SnapshotStatus.tsx @@ -22,11 +22,14 @@ export function SnapshotStatus({ sources }: { sources: SnapshotSource[] }) { age !== null && age * 1000 > Math.max(30000, poll.interval * 2); return (
-