From a0eea7b71c6d3320fb89f3413ea3c13ec4e88c0f Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 22:48:28 +0000 Subject: [PATCH] Serve the namespace system metrics like published PHP 2.11 GET|POST /api/system/metrics reports the namespace's workflow task failures, server-wide projection drift, and the namespace's request admission, durable-state and external payload usage, each with PHP's bounded label policy. Native enforces none of those quotas, so limits are null and rejection counters zero, as on an unconfigured PHP server. The new system-metrics fixture fails a workflow task in a named namespace, which native refused. That failure, and its retry, are scoped to the task's namespace, so named namespaces now accept it. --- .github/workflows/server-parity.yml | 4 +- docs/rust-server-port.md | 11 +- rust/src/runtime/execution.rs | 1 + rust/src/runtime/http.rs | 11 +- rust/src/runtime/mod.rs | 1 + rust/src/runtime/store.rs | 8 +- rust/src/runtime/system_metrics.rs | 191 ++++++++++++++++++ rust/tests/execution.rs | 3 + rust/tests/execution/namespaces.rs | 33 +++ rust/tests/execution/system_metrics.rs | 138 +++++++++++++ .../conformance/server-parity/contract.mjs | 3 + scripts/conformance/server-parity/probe.php | 6 + .../server-parity/system-metrics-contract.mjs | 86 ++++++++ .../server-parity/system-metrics-probe.php | 55 +++++ .../Fixtures/ServerParity/system-metrics.json | 32 +++ .../ServerParitySystemMetricsContractTest.mjs | 107 ++++++++++ 16 files changed, 679 insertions(+), 11 deletions(-) create mode 100644 rust/src/runtime/system_metrics.rs create mode 100644 rust/tests/execution/system_metrics.rs create mode 100644 scripts/conformance/server-parity/system-metrics-contract.mjs create mode 100644 scripts/conformance/server-parity/system-metrics-probe.php create mode 100644 tests/Fixtures/ServerParity/system-metrics.json create mode 100644 tests/Unit/ServerParitySystemMetricsContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 2c372df4f..7e134b458 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -36,6 +36,7 @@ on: - tests/Unit/ServerParityOperatorMetricsContractTest.mjs - tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs - tests/Unit/ServerParityServiceCatalogContractTest.mjs + - tests/Unit/ServerParitySystemMetricsContractTest.mjs - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs @@ -86,6 +87,7 @@ on: - tests/Unit/ServerParityOperatorMetricsContractTest.mjs - tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs - tests/Unit/ServerParityServiceCatalogContractTest.mjs + - tests/Unit/ServerParitySystemMetricsContractTest.mjs - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs @@ -1504,7 +1506,7 @@ jobs: -e COMPOSER_HOME=/tmp/composer -v "$PWD:/source" -w /source "$image" \ install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress docker run --rm --user=1000:1000 --entrypoint node -v "$PWD:/source:ro" -w /source \ - "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs > parity-evidence/comparator-tests.txt + "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs > parity-evidence/comparator-tests.txt docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \ --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \ "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 91089e775..bfe201c7a 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -101,10 +101,15 @@ Updated October 22, 2026. These published PHP routes on this tree have no native handler yet, by family. - **Used by the CLI or Waterline.** - - Namespace delete. - Standalone activities: list, describe and start. - - `system` metrics (JSON and Prometheus), the repair pass, and external - payload cleanup and its pass. + - The Prometheus metrics summary, the repair pass, and external payload + cleanup and its pass. (Namespace delete is served with the + `namespace-delete` fixture. The JSON `GET|POST /api/system/metrics` is + served with the `system-metrics` fixture: namespace failure, durable-state + and payload counts are exact; native enforces no admission, durable-state + or payload quota, so limits are null and rejections zero, as on an + unconfigured PHP server. Projection drift is server-wide, and native's + only drift source is a run without its summary.) - Task queue priority and fairness. - **Worker protocol features.** - Worker sessions. diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index a658f84bd..762114d58 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -339,6 +339,7 @@ impl Runtime { delegate!(system_health(config: &super::operator_metrics::MetricsConfig) -> Result<(bool, Value)>); delegate!(operator_dashboard(config: &super::operator_metrics::MetricsConfig, bounded: bool) -> Result); delegate!(repair_status() -> Result); + delegate!(system_metrics() -> Result); delegate!(retention_status(limit: i64) -> Result); delegate!(retention_pass(limit: i64, run_ids: Vec) -> Result<(Value, bool)>); delegate!(activity_timeout_status(limit: i64) -> Result); diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 8511adf76..87d7dbda7 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -203,6 +203,7 @@ pub fn router(runtime: Runtime) -> Router { ) .route("/api/system/operator-metrics", get(operator_metrics)) .route("/api/system/health", get(system_health)) + .route("/api/system/metrics", get(system_metrics).post(system_metrics)) .route("/api/system/operator-dashboard", get(operator_dashboard)) .route("/api/system/operator-dashboard/bounded", get(bounded_operator_dashboard)) .route( @@ -645,6 +646,10 @@ fn named_namespace_operation(method: &str, path: &str) -> bool { if path == "/api/system/retention/pass" || path == "/api/system/activity-timeouts/pass" { return method == "POST"; } + // The metrics describe the namespace's own state, read by GET or POST. + if path == "/api/system/metrics" { + return matches!(method, "GET" | "POST"); + } if let Some(path) = path.strip_prefix("/api/workflows/") { let parts: Vec<_> = path.split('/').collect(); if matches!( @@ -704,7 +709,7 @@ fn named_namespace_operation(method: &str, path: &str) -> bool { | ["workflow-tasks" | "activity-tasks", "poll"] | ["query-tasks", "poll"] | ["query-tasks", _, "complete" | "fail" | "status"] - | ["workflow-tasks", _, "complete" | "history" | "heartbeat"] + | ["workflow-tasks", _, "complete" | "fail" | "history" | "heartbeat"] | [ "activity-tasks", _, @@ -1473,6 +1478,10 @@ async fn activity_timeout_status( Ok(Json(runtime.activity_timeout_status(limit).await?)) } +async fn system_metrics(Extension(runtime): Extension) -> Result> { + Ok(Json(runtime.system_metrics().await?)) +} + /// PHP answers 503 while a check reports an error. async fn system_health(Extension(runtime): Extension) -> Result { let (unavailable, body) = runtime.system_health(&metrics_config(&runtime)).await?; diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 88948c1c4..43733c7fd 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -108,6 +108,7 @@ mod start_definition; mod start_request; mod storage_test; mod store; +mod system_metrics; mod system_health; mod task_failures; mod task_repair; diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index b798a252a..643c246d1 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -1848,12 +1848,8 @@ where // replay block, without a retry, until an operator repairs the run. let range_blocked = range_blocked || (!waiting && super::task_failures::blocks_replay(&body["failure"])); - if !waiting && self.namespace.as_ref() != "default" { - return Err(refuse( - StatusCode::UNPROCESSABLE_ENTITY, - "development_workflow_task_retry_namespace_unqualified", - )); - } + // The task is read within the request's namespace, and its retry + // copies the run's namespace. let mut tx = self.begin().await?; let task = self.task_row(&mut tx, task_id).await?; if DB::string(&task, "task_type")? != "workflow" { diff --git a/rust/src/runtime/system_metrics.rs b/rust/src/runtime/system_metrics.rs new file mode 100644 index 000000000..5e5a97b09 --- /dev/null +++ b/rust/src/runtime/system_metrics.rs @@ -0,0 +1,191 @@ +//! `GET|POST /api/system/metrics`, as published Server 2.11 serves it +//! (`SystemController::metrics`): the namespace's workflow task failures, +//! the server's projection drift, and the namespace's request admission, +//! durable-state and external payload usage, each with PHP's bounded label +//! policy. +//! +//! Native enforces no namespace request admission, durable-state quota or +//! external payload quota, and the configuration audit refuses settings that +//! would configure one. Their limits are therefore null and their rejection +//! counters zero, as PHP reports for an unconfigured server; usage is +//! measured. +use super::Result; +use super::store::now; +use serde_json::{Map, Value, json}; + +/// PHP's default `server.metrics.workflow_task_failure_type_limit`. +const WORKFLOW_TYPE_LIMIT: i64 = 20; + +const PROJECTION_TABLES: [&str; 5] = + ["run_summaries", "run_waits", "run_timeline_entries", "run_timer_entries", "run_lineage_entries"]; + +const ADMISSION_REASONS: [&str; 3] = [ + "namespace_request_rate_exhausted", + "namespace_request_concurrency_exhausted", + "namespace_request_admission_unavailable", +]; + +const PAYLOAD_REASONS: [&str; 3] = [ + "external_payload_namespace_bytes_exhausted", + "external_payload_namespace_objects_exhausted", + "external_payload_namespace_quota_unavailable", +]; + +/// `NamespaceDurableStateQuota::usage`, resource by resource, in PHP's order. +const DURABLE_STATE: [(&str, &str); 16] = [ + ("workflow_instances", "SELECT COUNT(*) FROM workflow_instances WHERE namespace=$1"), + ("workflow_runs", "SELECT COUNT(*) FROM workflow_runs WHERE namespace=$1"), + ("open_workflow_runs", "SELECT COUNT(*) FROM workflow_runs WHERE namespace=$1 AND status IN ('pending','running','waiting')"), + ("schedules", "SELECT COUNT(*) FROM workflow_schedules WHERE namespace=$1"), + ("schedule_history_events", "SELECT COUNT(*) FROM workflow_schedule_history_events WHERE namespace=$1"), + ("worker_registrations", "SELECT COUNT(*) FROM workflow_worker_registrations WHERE namespace=$1"), + ("workflow_history_events", "SELECT COUNT(*) FROM workflow_history_events WHERE workflow_run_id IN (SELECT id FROM workflow_runs WHERE namespace=$1)"), + ("workflow_tasks", "SELECT COUNT(*) FROM workflow_tasks WHERE namespace=$1"), + ("pending_workflow_tasks", "SELECT COUNT(*) FROM workflow_tasks WHERE namespace=$1 AND status='ready'"), + ("workflow_timers", "SELECT COUNT(*) FROM workflow_run_timers WHERE workflow_run_id IN (SELECT id FROM workflow_runs WHERE namespace=$1)"), + ("pending_workflow_timers", "SELECT COUNT(*) FROM workflow_run_timers WHERE workflow_run_id IN (SELECT id FROM workflow_runs WHERE namespace=$1) AND status='pending'"), + ("workflow_run_waits", "SELECT COUNT(*) FROM workflow_run_waits WHERE workflow_instance_id IN (SELECT id FROM workflow_instances WHERE namespace=$1)"), + ("open_workflow_run_waits", "SELECT COUNT(*) FROM workflow_run_waits WHERE workflow_instance_id IN (SELECT id FROM workflow_instances WHERE namespace=$1) AND status='open'"), + ("workflow_commands", "SELECT COUNT(*) FROM workflow_commands WHERE workflow_instance_id IN (SELECT id FROM workflow_instances WHERE namespace=$1)"), + ("workflow_streams", "SELECT (SELECT COUNT(*) FROM workflow_durable_streams WHERE namespace=$1)+(SELECT COUNT(*) FROM workflow_inbound_streams WHERE namespace=$2)"), + ("workflow_stream_items", "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE namespace=$1)+(SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE namespace=$2)"), +]; + +/// The namespace's failed workflow tasks, joined to their runs. +macro_rules! failed_tasks { + () => { + "FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.task_type='workflow' AND t.status='failed' AND t.attempt_count>0" + }; +} + +fn zeros(reasons: &[&str]) -> Value { + Value::Object(reasons.iter().map(|reason| ((*reason).to_owned(), json!(0))).collect()) +} + +fn nulls(keys: &[&str]) -> Value { + Value::Object(keys.iter().map(|key| ((*key).to_owned(), Value::Null)).collect()) +} + +store_methods! { + async fn namespace_count(&self, sql: &'static str, binds: usize) -> Result { + let mut query = Self::scalar(sql); + for _ in 0..binds { + query = query.bind(self.namespace.as_ref()); + } + Ok(query.fetch_one(&self.pool).await?) + } + + pub(crate) async fn system_metrics(&self) -> Result { + let failures = self.workflow_task_failures().await?; + let drift = self.projection_drift().await?; + let admission = json!({ + "max_requests_per_minute": null, "max_concurrent_requests": null, + "requests_this_minute": 0, "rejections_this_minute": 0, + "rejections_by_reason": zeros(&ADMISSION_REASONS), "configuration_status": "valid", + "label_cardinality_policy": {"namespace": "request_scope_not_label", "reason": "finite_three_reason_inventory"}, + }); + let durable = self.durable_state_usage().await?; + let payloads = json!({ + "max_bytes": null, "max_objects": null, + "used_bytes": self.namespace_count(DB::EXTERNAL_PAYLOAD_BYTES, 1).await?.max(0), + "used_objects": self.namespace_count("SELECT COUNT(*) FROM runtime_external_payloads WHERE namespace=$1", 1).await?, + "remaining_bytes": null, "remaining_objects": null, + "rejections_this_minute": 0, "rejections_by_reason": zeros(&PAYLOAD_REASONS), + "configuration_status": "valid", "measurement_status": "available", + "label_cardinality_policy": {"namespace": "request_scope_not_label", "reason": "finite_three_reason_inventory"}, + }); + let families = [ + ("dw_workflow_task_consecutive_failures", failures), + ("dw_projection_drift_total", drift), + ("dw_namespace_request_admission_rejections", admission), + ("dw_namespace_durable_state_usage", durable), + ("dw_runtime_external_payload_namespace_usage", payloads), + ]; + let label_sets: Map = families + .iter() + .map(|(name, family)| ((*name).to_owned(), family["label_cardinality_policy"].clone())) + .collect(); + let metrics: Map = families.into_iter().map(|(name, family)| (name.to_owned(), family)).collect(); + Ok(json!({ + "generated_at": now().to_rfc3339_opts(chrono::SecondsFormat::Micros, true), + "namespace": self.namespace.as_ref(), + "metrics": metrics, + "cardinality": {"metric_label_sets": label_sets}, + })) + } + + /// `WorkflowTaskFailureMetrics::snapshot`: failed workflow tasks, the + /// first failure counting as one. + async fn workflow_task_failures(&self) -> Result { + // PostgreSQL's MAX keeps the integer column's width; decode it by row. + const MAX: &str = concat!("SELECT COALESCE(MAX(t.attempt_count),0) AS n ", failed_tasks!()); + const COUNT: &str = concat!("SELECT COUNT(t.id) ", failed_tasks!()); + const TYPES: &str = concat!("SELECT COUNT(DISTINCT r.workflow_type) ", failed_tasks!()); + const ROWS: &str = concat!( + "SELECT r.workflow_type AS workflow_type,MAX(t.attempt_count) AS max_consecutive_failures,COUNT(*) AS failed_task_count ", + failed_tasks!(), + " GROUP BY r.workflow_type ORDER BY max_consecutive_failures DESC,r.workflow_type LIMIT 20" + ); + let max = DB::number(&Self::query(MAX).bind(self.namespace.as_ref()).fetch_one(&self.pool).await?, "n")?; + let failed = self.namespace_count(COUNT, 1).await?; + let types = self.namespace_count(TYPES, 1).await?; + let rows = Self::query(ROWS).bind(self.namespace.as_ref()).fetch_all(&self.pool).await?; + let mut series = Vec::new(); + let mut reported = 0; + for row in &rows { + let tasks = DB::number(row, "failed_task_count")?; + reported += tasks; + series.push(json!({"workflow_type": DB::string(row, "workflow_type")?, + "max_consecutive_failures": DB::number(row, "max_consecutive_failures")?, "failed_task_count": tasks})); + } + let shown = series.len() as i64; + Ok(json!({ + "max_consecutive_failures": max, "failed_task_count": failed, "workflow_type_count": types, + "workflow_type_limit": WORKFLOW_TYPE_LIMIT, "workflow_types_truncated": types > shown, + "suppressed_workflow_type_count": (types - shown).max(0), "suppressed_failed_task_count": (failed - reported).max(0), + "label_cardinality_policy": {"namespace": "request_scope_not_label", "workflow_type": { + "cardinality_class": "bounded_series", "limit": WORKFLOW_TYPE_LIMIT, + "selection": "top_by_max_consecutive_failures_then_name"}}, + "by_workflow_type": series, + })) + } + + /// `ProjectionDriftMetrics::snapshot`, server-wide. Native maintains + /// run waits with history and derives the timeline, timer and lineage + /// views from it, so only a run without its summary needs a rebuild. + async fn projection_drift(&self) -> Result { + let missing = Self::scalar("SELECT COUNT(*) FROM workflow_runs r WHERE NOT EXISTS (SELECT 1 FROM workflow_run_summaries s WHERE s.id=r.id)") + .fetch_one(&self.pool) + .await?; + let by_table: Vec = PROJECTION_TABLES + .iter() + .map(|table| json!({"table": table, "needs_rebuild": if *table == "run_summaries" { missing } else { 0 }})) + .collect(); + Ok(json!({ + "total": missing, "table_count": PROJECTION_TABLES.len(), "tables_with_drift": i64::from(missing > 0), "scope": "server", + "label_cardinality_policy": {"namespace": "server_scope_no_label", "table": { + "cardinality_class": "finite_projection_table_inventory", "values": PROJECTION_TABLES, + "selection": "fixed_projection_table_inventory"}}, + "by_table": by_table, + })) + } + + /// `NamespaceDurableStateQuota::metrics` with no configured limits. + async fn durable_state_usage(&self) -> Result { + let mut usage = Map::new(); + for (resource, sql) in DURABLE_STATE { + let binds = if sql.contains("$2") { 2 } else { 1 }; + usage.insert(resource.to_owned(), json!(self.namespace_count(sql, binds).await?)); + } + let resources: Vec<&str> = DURABLE_STATE.iter().map(|(resource, _)| *resource).collect(); + let mut reasons: Vec = resources.iter().map(|resource| format!("namespace_{resource}_exhausted")).collect(); + reasons.push("namespace_durable_state_quota_unavailable".into()); + let reasons: Vec<&str> = reasons.iter().map(String::as_str).collect(); + Ok(json!({ + "limits": nulls(&resources), "usage": usage, "remaining": nulls(&resources), + "rejections_this_minute": 0, "rejections_by_reason": zeros(&reasons), + "configuration_status": "valid", "measurement_status": "available", + "label_cardinality_policy": {"namespace": "request_scope_not_label", "reason": "finite_reason_inventory"}, + })) + } +} diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 2f60a1c43..d23b871c5 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -133,6 +133,9 @@ mod version_markers; #[path = "execution/activity_lease_renewal.rs"] mod activity_lease_renewal; +#[path = "execution/system_metrics.rs"] +mod system_metrics; + #[path = "execution/activity_deadlines.rs"] mod activity_deadlines; diff --git a/rust/tests/execution/namespaces.rs b/rust/tests/execution/namespaces.rs index 933665b6a..2572bb5a5 100644 --- a/rust/tests/execution/namespaces.rs +++ b/rust/tests/execution/namespaces.rs @@ -615,6 +615,39 @@ async fn same_worker_id_isolated_claims_completion_and_deregistration_by_namespa database.remove().await; } +#[tokio::test] +async fn named_namespace_workflow_task_failure_retries_only_in_its_own_namespace() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + for name in ["alpha", "beta"] { + create(&app, name).await; + let (status, body) = scoped(&app, name, "POST", "/api/worker/register", json!({ + "worker_id":"shared-worker","task_queue":"test","runtime":"php", + "supported_workflow_types":["echo"],"supported_activity_types":[]})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); + } + let run = start_scoped(&app, "alpha", "retry-alpha").await; + let poll = json!({"worker_id":"shared-worker","task_queue":"test","timeout_seconds":0}); + let task = scoped(&app, "alpha", "POST", "/api/worker/workflow-tasks/poll", poll.clone()).await.1["task"].clone(); + let path = format!("/api/worker/workflow-tasks/{}/fail", task["task_id"].as_str().unwrap()); + let body = json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"], + "failure":{"message":"named failure","type":"RuntimeException"}}); + let refused = scoped(&app, "beta", "POST", &path, body.clone()).await; + assert_eq!(refused.0, StatusCode::NOT_FOUND, "{}", refused.1); + assert_eq!(refused.1["reason"], "task_not_found"); + + let (status, failed) = scoped(&app, "alpha", "POST", &path, body).await; + assert_eq!(status, StatusCode::OK, "{failed}"); + assert_eq!([&failed["outcome"], &failed["recorded"]], [&json!("failed"), &json!(true)]); + let foreign = scoped(&app, "beta", "POST", "/api/worker/workflow-tasks/poll", poll.clone()).await; + assert_eq!(foreign.0, StatusCode::OK, "{}", foreign.1); + assert!(foreign.1["task"].is_null(), "{}", foreign.1); + let retry = scoped(&app, "alpha", "POST", "/api/worker/workflow-tasks/poll", poll).await.1["task"].clone(); + assert_eq!(retry["task_id"], failed["next_task_id"], "{retry}"); + assert_eq!([&retry["run_id"], &retry["workflow_task_attempt"]], [&run["run_id"], &json!(2)]); +} + #[tokio::test] async fn namespace_updates_follow_published_retention_rules() { let database = TestDatabase::new().await; diff --git a/rust/tests/execution/system_metrics.rs b/rust/tests/execution/system_metrics.rs new file mode 100644 index 000000000..2e1fbcfa3 --- /dev/null +++ b/rust/tests/execution/system_metrics.rs @@ -0,0 +1,138 @@ +use super::*; + +const METRICS: &str = "/api/system/metrics"; + +async fn system_metrics(app: &Router, method: &str) -> Value { + let (status, body) = request(app, method, METRICS, Value::Null).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["namespace"], "default"); + let families: Vec<&String> = body["metrics"].as_object().unwrap().keys().collect(); + assert_eq!(families.len(), 5, "{body}"); + for (name, family) in body["metrics"].as_object().unwrap() { + assert_eq!( + body["cardinality"]["metric_label_sets"][name], family["label_cardinality_policy"], + "{name}" + ); + } + body["metrics"].clone() +} + +async fn start_type(app: &Router, workflow: &str, workflow_type: &str) { + let (status, body) = request(app, "POST", "/api/workflows", json!({"workflow_id": workflow, + "workflow_type": workflow_type, "task_queue": "test", + "input": envelope(Payload::Array(vec![]))})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); +} + +#[tokio::test] +async fn system_metrics_count_a_failed_workflow_task_and_the_namespace_state_like_published_php() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let empty = system_metrics(&app, "GET").await; + assert_eq!( + empty["dw_workflow_task_consecutive_failures"], + json!({"max_consecutive_failures": 0, "failed_task_count": 0, "workflow_type_count": 0, + "workflow_type_limit": 20, "workflow_types_truncated": false, "suppressed_workflow_type_count": 0, + "suppressed_failed_task_count": 0, "label_cardinality_policy": {"namespace": "request_scope_not_label", + "workflow_type": {"cardinality_class": "bounded_series", "limit": 20, + "selection": "top_by_max_consecutive_failures_then_name"}}, + "by_workflow_type": []}) + ); + assert_eq!(empty["dw_projection_drift_total"]["total"], 0); + let usage = &empty["dw_namespace_durable_state_usage"]; + assert_eq!(usage["usage"].as_object().unwrap().len(), 16); + assert!(usage["usage"].as_object().unwrap().values().all(|count| count == 0), "{usage}"); + assert_eq!(usage["rejections_by_reason"].as_object().unwrap().len(), 17); + assert_eq!(usage["rejections_by_reason"]["namespace_durable_state_quota_unavailable"], 0); + assert!(usage["limits"].as_object().unwrap().values().all(Value::is_null)); + assert_eq!(empty["dw_runtime_external_payload_namespace_usage"]["used_objects"], 0); + assert_eq!(empty["dw_namespace_request_admission_rejections"]["max_requests_per_minute"], Value::Null); + + register(&app, "metrics-worker", json!(["echo", "other"]), json!([])).await; + start_type(&app, "metrics-a", "echo").await; + start_type(&app, "metrics-b", "other").await; + let task = poll(&app, "metrics-worker", "workflow").await; + assert_eq!(task["workflow_type"], "echo", "{task}"); + let (status, body) = request(&app, "POST", + &format!("/api/worker/workflow-tasks/{}/fail", task["task_id"].as_str().unwrap()), + json!({"lease_owner": task["lease_owner"], "workflow_task_attempt": task["workflow_task_attempt"], + "failure": {"message": "metrics failure", "type": "RuntimeException"}})).await; + assert_eq!(status, StatusCode::OK, "{body}"); + + let read = system_metrics(&app, "GET").await; + assert_eq!(read, system_metrics(&app, "POST").await); + let failures = &read["dw_workflow_task_consecutive_failures"]; + assert_eq!( + [&failures["max_consecutive_failures"], &failures["failed_task_count"], &failures["workflow_type_count"]], + [&json!(1), &json!(1), &json!(1)] + ); + assert_eq!( + failures["by_workflow_type"], + json!([{"workflow_type": "echo", "max_consecutive_failures": 1, "failed_task_count": 1}]) + ); + let usage = &read["dw_namespace_durable_state_usage"]["usage"]; + for (resource, count) in [ + ("workflow_instances", 2), + ("workflow_runs", 2), + ("open_workflow_runs", 2), + ("worker_registrations", 1), + ("workflow_tasks", 3), + ("pending_workflow_tasks", 2), + ("workflow_commands", 2), + ("schedules", 0), + ("workflow_timers", 0), + ("workflow_streams", 0), + ] { + assert_eq!(usage[resource], count, "{resource}: {usage}"); + } +} + +#[tokio::test] +async fn system_metrics_bound_the_failure_series_and_report_server_wide_summary_drift() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + for index in 0..21 { + start_type(&app, &format!("bounded-{index:02}"), &format!("type-{index:02}")).await; + } + database.execute("UPDATE workflow_tasks SET status='failed', attempt_count=1 WHERE task_type='workflow'").await; + database.execute("UPDATE workflow_tasks SET attempt_count=3 WHERE workflow_run_id IN (SELECT id FROM workflow_runs WHERE workflow_type='type-20')").await; + database.execute("DELETE FROM workflow_run_summaries WHERE id IN (SELECT id FROM workflow_runs WHERE workflow_type='type-05')").await; + + let metrics = system_metrics(&app, "GET").await; + let failures = &metrics["dw_workflow_task_consecutive_failures"]; + assert_eq!( + [ + &failures["max_consecutive_failures"], + &failures["failed_task_count"], + &failures["workflow_type_count"], + &failures["workflow_types_truncated"], + &failures["suppressed_workflow_type_count"], + &failures["suppressed_failed_task_count"], + ], + [&json!(3), &json!(21), &json!(21), &json!(true), &json!(1), &json!(1)] + ); + let shown: Vec<&str> = failures["by_workflow_type"] + .as_array() + .unwrap() + .iter() + .map(|row| row["workflow_type"].as_str().unwrap()) + .collect(); + let mut expected = vec!["type-20".to_owned()]; + expected.extend((0..19).map(|index| format!("type-{index:02}"))); + assert_eq!(shown, expected); + assert_eq!(failures["by_workflow_type"][0]["max_consecutive_failures"], 3); + + let drift = &metrics["dw_projection_drift_total"]; + assert_eq!( + [&drift["total"], &drift["tables_with_drift"], &drift["table_count"], &drift["scope"]], + [&json!(1), &json!(1), &json!(5), &json!("server")] + ); + assert_eq!( + drift["by_table"], + json!([{"table": "run_summaries", "needs_rebuild": 1}, {"table": "run_waits", "needs_rebuild": 0}, + {"table": "run_timeline_entries", "needs_rebuild": 0}, {"table": "run_timer_entries", "needs_rebuild": 0}, + {"table": "run_lineage_entries", "needs_rebuild": 0}]) + ); +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 794ad66a9..936504479 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -14,6 +14,7 @@ import {checkSearchAttributes} from './search-attribute-contract.mjs'; import {checkStorageTest} from './storage-test-contract.mjs'; import {checkActivityLeaseRenewal} from './activity-lease-renewal-contract.mjs'; import {checkServiceCatalog} from './service-catalog-contract.mjs'; +import {checkSystemMetrics} from './system-metrics-contract.mjs'; import {checkNamespaceDelete} from './namespace-delete-contract.mjs'; import {checkSchedulePolicies} from './schedule-policies-contract.mjs'; import {checkScheduleCatalog} from './schedule-catalog-contract.mjs'; @@ -146,6 +147,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const storageTest = fixture.storage_test ? checkStorageTest(fixture, observation, workflowId) : null; const activityLeaseRenewal = fixture.activity_lease_renewal ? checkActivityLeaseRenewal(fixture, observation, workflowId) : null; const serviceCatalog = fixture.service_catalog ? checkServiceCatalog(fixture, observation, workflowId) : null; + const systemMetrics = fixture.system_metrics ? checkSystemMetrics(fixture, observation, workflowId) : null; const namespaceDelete = fixture.namespace_delete ? checkNamespaceDelete(fixture, observation, workflowId) : null; const schedulePolicies = fixture.schedule_policies ? checkSchedulePolicies(fixture, observation, workflowId) : null; const scheduleCatalog = fixture.schedule_catalog ? checkScheduleCatalog(fixture, observation, workflowId) : null; @@ -695,6 +697,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(storageTest ? {storage_test: storageTest} : {}), ...(activityLeaseRenewal ? {activity_lease_renewal: activityLeaseRenewal} : {}), ...(serviceCatalog ? {service_catalog: serviceCatalog} : {}), + ...(systemMetrics ? {system_metrics: systemMetrics} : {}), ...(namespaceDelete ? {namespace_delete: namespaceDelete} : {}), ...(schedulePolicies ? {schedule_policies: schedulePolicies} : {}), ...(scheduleCatalog ? {schedule_catalog: scheduleCatalog} : {}), diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index 5ccb00c47..c05c88677 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -432,6 +432,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $storageTest = isset($fixture['storage_test']) ? finishHttpStorageTest($client, $fixture, $workflowId, $url) : null; $activityLeaseRenewal = isset($fixture['activity_lease_renewal']) ? finishHttpActivityLeaseRenewal($client, $fixture, $workflowId, $url) : null; $serviceCatalog = isset($fixture['service_catalog']) ? finishHttpServiceCatalog($client, $fixture, $workflowId, $url) : null; + $systemMetrics = isset($fixture['system_metrics']) ? finishHttpSystemMetrics($client, $fixture, $workflowId, $url) : null; $namespaceDelete = isset($fixture['namespace_delete']) ? finishHttpNamespaceDelete($client, $fixture, $workflowId, $queue, $url) : null; $schedulePolicies = isset($fixture['schedule_policies']) ? finishHttpSchedulePolicies($client, $fixture, $workflowId, $url) : null; $scheduleCatalog = isset($fixture['schedule_catalog']) @@ -534,6 +535,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'storage_test' => $storageTest, 'activity_lease_renewal' => $activityLeaseRenewal, 'service_catalog' => $serviceCatalog, + 'system_metrics' => $systemMetrics, 'namespace_delete' => $namespaceDelete, 'schedule_policies' => $schedulePolicies, 'schedule_catalog' => $scheduleCatalog, @@ -840,6 +842,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/storage-test-probe.php'; require __DIR__.'/activity-lease-renewal-probe.php'; require __DIR__.'/service-catalog-probe.php'; + require __DIR__.'/system-metrics-probe.php'; require __DIR__.'/namespace-delete-probe.php'; require __DIR__.'/schedule-policies-probe.php'; require __DIR__.'/signals-before-worker-probe.php'; @@ -1005,6 +1008,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['service_catalog'])) { $observation['service_catalog'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_service_catalog']; } + if ($mode === 'embedded' && isset($fixture['system_metrics'])) { + $observation['system_metrics'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_metrics']; + } if ($mode === 'embedded' && isset($fixture['namespace_delete'])) { $observation['namespace_delete'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_namespaces']; } diff --git a/scripts/conformance/server-parity/system-metrics-contract.mjs b/scripts/conformance/server-parity/system-metrics-contract.mjs new file mode 100644 index 000000000..4f4bf2efb --- /dev/null +++ b/scripts/conformance/server-parity/system-metrics-contract.mjs @@ -0,0 +1,86 @@ +import assert from 'node:assert/strict'; + +const MICROS_OR_MILLIS = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}(\d{3})?Z$/; + +function nulls(keys) { + return Object.fromEntries(keys.map(key => [key, null])); +} + +function zeros(keys) { + return Object.fromEntries(keys.map(key => [key, 0])); +} + +// Reviewed namespace metrics (SystemController::metrics). Namespace-scoped +// values are exact. Projection drift is server-wide and counts rows other +// fixtures leave stale or orphaned, so only its shape and totals are checked. +export function checkSystemMetrics(fixture, observation, workflowId) { + const definition = fixture.system_metrics; + const state = observation.system_metrics; + const projection = {scope: 'system_metrics', resources: Object.keys(definition.usage).length}; + assert.ok(state && typeof state === 'object', 'actual system metrics observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_http_metrics'}, 'embedded metrics explicitly inapplicable'); + return projection; + } + const namespace = `${workflowId}-metrics`.toLowerCase(); + assert.deepStrictEqual([state.applicable, state.namespace, state.registered, state.started, state.claimed_type, state.failed, state.scheduled], + [true, namespace, 201, [201, 201], definition.workflow_types[0], 200, 201], 'metrics scenario'); + const resources = Object.keys(definition.usage); + const durableReasons = [...resources.map(resource => `namespace_${resource}_exhausted`), 'namespace_durable_state_quota_unavailable']; + const expected = (failures, usage) => ({ + dw_workflow_task_consecutive_failures: failures, + dw_namespace_request_admission_rejections: { + max_requests_per_minute: null, max_concurrent_requests: null, requests_this_minute: 0, rejections_this_minute: 0, + rejections_by_reason: zeros(['namespace_request_rate_exhausted', 'namespace_request_concurrency_exhausted', + 'namespace_request_admission_unavailable']), + configuration_status: 'valid', + label_cardinality_policy: {namespace: 'request_scope_not_label', reason: 'finite_three_reason_inventory'}, + }, + dw_namespace_durable_state_usage: { + limits: nulls(resources), usage, remaining: nulls(resources), rejections_this_minute: 0, + rejections_by_reason: zeros(durableReasons), configuration_status: 'valid', measurement_status: 'available', + label_cardinality_policy: {namespace: 'request_scope_not_label', reason: 'finite_reason_inventory'}, + }, + dw_runtime_external_payload_namespace_usage: { + max_bytes: null, max_objects: null, used_bytes: 0, used_objects: 0, remaining_bytes: null, remaining_objects: null, + rejections_this_minute: 0, rejections_by_reason: zeros(['external_payload_namespace_bytes_exhausted', + 'external_payload_namespace_objects_exhausted', 'external_payload_namespace_quota_unavailable']), + configuration_status: 'valid', measurement_status: 'available', + label_cardinality_policy: {namespace: 'request_scope_not_label', reason: 'finite_three_reason_inventory'}, + }, + }); + const empty = {...definition.failures, max_consecutive_failures: 0, failed_task_count: 0, workflow_type_count: 0, by_workflow_type: []}; + const check = (recorded, failures, usage, label) => { + assert.equal(recorded.status, 200, `${label} status`); + const {generated_at: generatedAt, namespace: observed, metrics, cardinality} = recorded.response; + assert.deepStrictEqual(Object.keys(recorded.response).sort(), ['cardinality', 'generated_at', 'metrics', 'namespace'], `${label} fields`); + assert.match(generatedAt, MICROS_OR_MILLIS, `${label} generated_at in UTC`); + assert.equal(observed, namespace, `${label} namespace`); + const {dw_projection_drift_total: drift, ...scoped} = metrics; + assert.deepStrictEqual(scoped, expected(failures, usage), `${label} namespace metrics`); + assert.ok(drift && Array.isArray(drift.by_table), `${label} projection drift family`); + assert.deepStrictEqual(Object.keys(drift).sort(), ['by_table', 'label_cardinality_policy', 'scope', 'table_count', 'tables_with_drift', 'total'], + `${label} drift fields`); + const tables = definition.projection_tables; + assert.deepStrictEqual(drift.by_table.map(row => row.table), tables, `${label} drift tables`); + for (const row of drift.by_table) { + assert.deepStrictEqual(Object.keys(row).sort(), ['needs_rebuild', 'table'], `${label} ${row.table} drift fields`); + assert.ok(Number.isInteger(row.needs_rebuild) && row.needs_rebuild >= 0, `${label} ${row.table} drift`); + } + const counts = drift.by_table.map(row => row.needs_rebuild); + assert.deepStrictEqual( + {total: drift.total, table_count: drift.table_count, tables_with_drift: drift.tables_with_drift, scope: drift.scope, + label_cardinality_policy: drift.label_cardinality_policy}, + {total: counts.reduce((sum, count) => sum + count, 0), table_count: tables.length, + tables_with_drift: counts.filter(count => count > 0).length, scope: 'server', + label_cardinality_policy: {namespace: 'server_scope_no_label', table: {cardinality_class: 'finite_projection_table_inventory', + values: tables, selection: 'fixed_projection_table_inventory'}}}, + `${label} drift totals`); + assert.deepStrictEqual(cardinality, {metric_label_sets: Object.fromEntries(Object.entries(metrics) + .map(([name, family]) => [name, family.label_cardinality_policy]))}, `${label} cardinality`); + }; + check(state.empty, empty, zeros(resources), 'empty namespace'); + check(state.read, definition.failures, definition.usage, 'after the scenario'); + check(state.posted, definition.failures, definition.usage, 'read by POST'); + return projection; +} diff --git a/scripts/conformance/server-parity/system-metrics-probe.php b/scripts/conformance/server-parity/system-metrics-probe.php new file mode 100644 index 000000000..fc57d5733 --- /dev/null +++ b/scripts/conformance/server-parity/system-metrics-probe.php @@ -0,0 +1,55 @@ +createNamespace($namespace, 'system metrics λ', 30); + $token = getenv('DW_PARITY_TOKEN') ?: null; + $http = new SystemMetricsHttp(['base_uri' => rtrim($url, '/').'/api/', 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 30]); + $raw = static function (string $method, string $path, ?array $body = null, bool $worker = false) use ($http, $token, $namespace): array { + $response = $http->request($method, $path, ['headers' => array_filter([ + 'Authorization' => $token === null ? null : 'Bearer '.$token, + 'Accept' => 'application/json', 'X-Namespace' => $namespace, + $worker ? 'X-Durable-Workflow-Protocol-Version' : 'X-Durable-Workflow-Control-Plane-Version' => $worker ? '1.20' : '2', + ]), ...($body === null ? [] : ['json' => $body])]); + + return ['status' => $response->getStatusCode(), + 'response' => json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR)]; + }; + $input = ['codec' => 'avro', 'blob' => $definition['empty_blob']]; + $state = ['applicable' => true, 'namespace' => $namespace]; + $state['empty'] = $raw('GET', 'system/metrics'); + $state['registered'] = $raw('POST', 'worker/register', ['worker_id' => $worker, 'task_queue' => $queue, 'runtime' => 'php', + 'sdk_version' => 'raw-http', 'supported_workflow_types' => $definition['workflow_types'], 'supported_activity_types' => [], + 'capabilities' => [], 'capability_manifest' => array_fill_keys(['local_activities', 'worker_sessions', 'sticky_execution'], + ['supported' => false, 'minimum_protocol_version' => '1.18', 'reason' => 'raw_http'])], true)['status']; + $state['started'] = []; + foreach ($definition['workflow_types'] as $index => $type) { + $state['started'][] = $raw('POST', 'workflows', ['workflow_id' => $namespace.'-'.$index, + 'workflow_type' => $type, 'task_queue' => $queue, 'input' => $input])['status']; + } + $task = $raw('POST', 'worker/workflow-tasks/poll', ['worker_id' => $worker, 'task_queue' => $queue, 'timeout_seconds' => 0], true)['response']['task'] ?? []; + $state['claimed_type'] = $task['workflow_type'] ?? null; + $state['failed'] = $raw('POST', 'worker/workflow-tasks/'.($task['task_id'] ?? 'none').'/fail', [ + 'lease_owner' => $task['lease_owner'] ?? null, 'workflow_task_attempt' => $task['workflow_task_attempt'] ?? null, + 'failure' => ['message' => 'metrics fixture failure', 'type' => 'RuntimeException'], + ], true)['status']; + $state['scheduled'] = $raw('POST', 'schedules', ['schedule_id' => $namespace.'-schedule', 'spec' => ['intervals' => [['every' => 'P1D']]], + 'action' => ['workflow_type' => $definition['workflow_types'][0], 'task_queue' => $queue, 'input' => $input]])['status']; + $state['read'] = $raw('GET', 'system/metrics'); + $state['posted'] = $raw('POST', 'system/metrics'); + + return $state; +} diff --git a/tests/Fixtures/ServerParity/system-metrics.json b/tests/Fixtures/ServerParity/system-metrics.json new file mode 100644 index 000000000..72c15458d --- /dev/null +++ b/tests/Fixtures/ServerParity/system-metrics.json @@ -0,0 +1,32 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "system-metrics", + "review_state": "reviewed", + "workflow_type": "parity.v1.echo", + "input": {"message": "system metrics λ", "count": 9007199254740993, "optional": null}, + "typed_value": {"type": "map", "value": { + "message": {"type": "string", "value": "system metrics λ"}, + "count": {"type": "int64", "value": "9007199254740993"}, + "optional": {"type": "null", "value": null} + }}, + "activity": false, + "expected_events": ["StartAccepted", "WorkflowStarted", "WorkflowCompleted"], + "system_metrics": { + "workflow_types": ["parity.v1.metrics_a", "parity.v1.metrics_b"], + "empty_blob": "wwHioz3/VYAiNwwA", + "failures": { + "max_consecutive_failures": 1, "failed_task_count": 1, "workflow_type_count": 1, "workflow_type_limit": 20, + "workflow_types_truncated": false, "suppressed_workflow_type_count": 0, "suppressed_failed_task_count": 0, + "label_cardinality_policy": {"namespace": "request_scope_not_label", "workflow_type": {"cardinality_class": "bounded_series", "limit": 20, "selection": "top_by_max_consecutive_failures_then_name"}}, + "by_workflow_type": [{"workflow_type": "parity.v1.metrics_a", "max_consecutive_failures": 1, "failed_task_count": 1}] + }, + "usage": { + "workflow_instances": 2, "workflow_runs": 2, "open_workflow_runs": 2, "schedules": 1, "schedule_history_events": 1, + "worker_registrations": 1, "workflow_history_events": 4, "workflow_tasks": 3, "pending_workflow_tasks": 2, + "workflow_timers": 0, "pending_workflow_timers": 0, "workflow_run_waits": 0, "open_workflow_run_waits": 0, + "workflow_commands": 2, "workflow_streams": 0, "workflow_stream_items": 0 + }, + "projection_tables": ["run_summaries", "run_waits", "run_timeline_entries", "run_timer_entries", "run_lineage_entries"] + }, + "contract": "GET and POST /api/system/metrics (published Server 2.11) report the namespace's workflow task failures, the server's projection drift, and the namespace's request admission, durable-state and external payload usage, each with its bounded label policy. In a fresh namespace that holds two started runs of two workflow types, one failed workflow task of the first type and one schedule, the failure series names that type with one consecutive failure, and the sixteen durable-state counts are exact. No admission, durable-state or payload limit is configured, so limits and remaining values are null and rejection counters zero, and no external payload is stored. Projection drift is server-wide and counts stale or orphaned rows that other work on the server leaves, so it is checked for its fixed table inventory, non-negative counts and consistent totals, not exact values. Embedded applications have no HTTP metrics." +} diff --git a/tests/Unit/ServerParitySystemMetricsContractTest.mjs b/tests/Unit/ServerParitySystemMetricsContractTest.mjs new file mode 100644 index 000000000..4f1886de8 --- /dev/null +++ b/tests/Unit/ServerParitySystemMetricsContractTest.mjs @@ -0,0 +1,107 @@ +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import test from 'node:test'; +import {checkSystemMetrics} from '../../scripts/conformance/server-parity/system-metrics-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/system-metrics.json', import.meta.url))); +const definition = fixture.system_metrics; +const id = 'parity-v1-system-metrics'; +const namespace = `${id}-metrics`; +const resources = Object.keys(definition.usage); +const zeros = keys => Object.fromEntries(keys.map(key => [key, 0])); +const nulls = keys => Object.fromEntries(keys.map(key => [key, null])); + +// A response as published Server 2.11.1 answered the scenario, with the +// timeline drift PHP's asynchronous projection showed. +function response(scenario, timelineDrift = 0) { + const failures = scenario ? definition.failures : {...definition.failures, max_consecutive_failures: 0, + failed_task_count: 0, workflow_type_count: 0, by_workflow_type: []}; + const counts = definition.projection_tables.map(table => table === 'run_timeline_entries' ? timelineDrift : 0); + const metrics = { + dw_workflow_task_consecutive_failures: failures, + dw_projection_drift_total: { + total: timelineDrift, table_count: 5, tables_with_drift: timelineDrift > 0 ? 1 : 0, scope: 'server', + label_cardinality_policy: {namespace: 'server_scope_no_label', table: {cardinality_class: 'finite_projection_table_inventory', + values: definition.projection_tables, selection: 'fixed_projection_table_inventory'}}, + by_table: definition.projection_tables.map((table, index) => ({table, needs_rebuild: counts[index]})), + }, + dw_namespace_request_admission_rejections: { + max_requests_per_minute: null, max_concurrent_requests: null, requests_this_minute: 0, rejections_this_minute: 0, + rejections_by_reason: zeros(['namespace_request_rate_exhausted', 'namespace_request_concurrency_exhausted', + 'namespace_request_admission_unavailable']), + configuration_status: 'valid', + label_cardinality_policy: {namespace: 'request_scope_not_label', reason: 'finite_three_reason_inventory'}, + }, + dw_namespace_durable_state_usage: { + limits: nulls(resources), usage: scenario ? definition.usage : zeros(resources), remaining: nulls(resources), + rejections_this_minute: 0, + rejections_by_reason: zeros([...resources.map(resource => `namespace_${resource}_exhausted`), 'namespace_durable_state_quota_unavailable']), + configuration_status: 'valid', measurement_status: 'available', + label_cardinality_policy: {namespace: 'request_scope_not_label', reason: 'finite_reason_inventory'}, + }, + dw_runtime_external_payload_namespace_usage: { + max_bytes: null, max_objects: null, used_bytes: 0, used_objects: 0, remaining_bytes: null, remaining_objects: null, + rejections_this_minute: 0, rejections_by_reason: zeros(['external_payload_namespace_bytes_exhausted', + 'external_payload_namespace_objects_exhausted', 'external_payload_namespace_quota_unavailable']), + configuration_status: 'valid', measurement_status: 'available', + label_cardinality_policy: {namespace: 'request_scope_not_label', reason: 'finite_three_reason_inventory'}, + }, + }; + const cardinality = {metric_label_sets: Object.fromEntries(Object.entries(metrics) + .map(([name, family]) => [name, family.label_cardinality_policy]))}; + return structuredClone({status: 200, response: {generated_at: '2026-10-22T22:15:23.466270Z', namespace, metrics, cardinality}}); +} + +const observation = (timelineDrift = 1) => ({mode: 'http', system_metrics: { + applicable: true, namespace, empty: response(false), registered: 201, started: [201, 201], + claimed_type: definition.workflow_types[0], failed: 200, scheduled: 201, + read: response(true, timelineDrift), posted: response(true, timelineDrift), +}}); + +test('the system metrics accept the published answers', () => { + assert.deepStrictEqual(checkSystemMetrics(fixture, observation(), id), {scope: 'system_metrics', resources: 16}); +}); + +test('the system metrics accept native answers with no drift and millisecond times', () => { + const value = observation(0); + value.system_metrics.read.response.generated_at = '2026-10-22T22:15:23.466Z'; + checkSystemMetrics(fixture, value, id); +}); + +test('the fixture pins the scenario counts', () => { + assert.deepStrictEqual(resources.length, 16); + assert.deepStrictEqual([definition.usage.workflow_tasks, definition.usage.pending_workflow_tasks, definition.usage.workflow_commands], [3, 2, 2]); + assert.deepStrictEqual(definition.failures.by_workflow_type, [{workflow_type: 'parity.v1.metrics_a', max_consecutive_failures: 1, failed_task_count: 1}]); +}); + +test('the embedded system metrics are explicitly inapplicable', () => { + checkSystemMetrics(fixture, {mode: 'embedded', system_metrics: {applicable: false, reason: 'embedded_has_no_http_metrics'}}, id); + assert.throws(() => checkSystemMetrics(fixture, {mode: 'embedded', system_metrics: {applicable: true}}, id)); +}); + +for (const [label, mutate] of [ + ['an unimplemented route', state => { state.read = {status: 501, response: {reason: 'rust_capability_not_implemented'}}; }], + ['a refused POST', state => { state.posted.status = 405; }], + ['another namespace', state => { state.read.response.namespace = 'default'; }], + ['a wrong durable-state count', state => { state.read.response.metrics.dw_namespace_durable_state_usage.usage.workflow_tasks = 2; }], + ['a configured limit', state => { state.read.response.metrics.dw_namespace_durable_state_usage.limits.schedules = 10; }], + ['a missing quota reason', state => { delete state.read.response.metrics.dw_namespace_durable_state_usage.rejections_by_reason.namespace_durable_state_quota_unavailable; }], + ['the failure on the other type', state => { state.read.response.metrics.dw_workflow_task_consecutive_failures.by_workflow_type[0] = {workflow_type: 'parity.v1.metrics_b', max_consecutive_failures: 1, failed_task_count: 1}; }], + ['a failure counted before any attempt', state => { state.empty.response.metrics.dw_workflow_task_consecutive_failures.failed_task_count = 1; }], + ['a stored external payload', state => { state.read.response.metrics.dw_runtime_external_payload_namespace_usage.used_objects = 1; }], + ['a drift total unlike its tables', state => { state.read.response.metrics.dw_projection_drift_total.total = 0; }], + ['a negative drift', state => { state.read.response.metrics.dw_projection_drift_total.by_table[0].needs_rebuild = -1; }], + ['a missing drift table', state => { state.read.response.metrics.dw_projection_drift_total.by_table.pop(); }], + ['a namespace-scoped drift', state => { state.read.response.metrics.dw_projection_drift_total.scope = 'namespace'; }], + ['an extra metric family', state => { state.read.response.metrics.dw_extra = {label_cardinality_policy: {}}; }], + ['a cardinality summary unlike its families', state => { state.read.response.cardinality.metric_label_sets.dw_projection_drift_total = {}; }], + ['an extra top-level field', state => { state.read.response.control_plane = {}; }], + ['a local time', state => { state.read.response.generated_at = '2026-10-22 22:15:23'; }], + ['the second type claimed first', state => { state.claimed_type = definition.workflow_types[1]; }], +]) { + test(`the system metrics reject ${label}`, () => { + const value = observation(); + mutate(value.system_metrics); + assert.throws(() => checkSystemMetrics(fixture, value, id)); + }); +}