diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 9929589ec..32101a29c 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -63,8 +63,9 @@ use crate::observability::{ use crate::plugin::{ ATIF_RUNTIME_DELIVERY_FAILURE_MARKER, ConfigDiagnostic, ConfigPolicy, DiagnosticLevel, OTEL_RUNTIME_DELIVERY_FAILURE_MARKER, Plugin, PluginComponentSpec, PluginError, - PluginRegistration, PluginRegistrationContext, Result as PluginResult, UnsupportedBehavior, - apply_global_config_policy, deregister_plugin, register_builtin_plugin, + PluginRegistration, PluginRegistrationCleanupOutcome, PluginRegistrationContext, + Result as PluginResult, UnsupportedBehavior, apply_global_config_policy, deregister_plugin, + register_builtin_plugin, }; use crate::plugin::{RuntimeDiagnostic, record_active_plugin_runtime_diagnostic}; @@ -892,41 +893,62 @@ fn register_atif_dispatcher( ); ctx.register_subscriber("atif", dispatcher)?; let shutdown_storage = Arc::clone(&storage); - ctx.add_registration(PluginRegistration::new( + ctx.add_registration(PluginRegistration::new_with_outcome( "observability", ctx.qualify_name("atif.shutdown"), Box::new(move || { - let work = { - let mut guard = manager.lock().map_err(|err| { - PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}")) - })?; - guard.flush_open_agents() - }; - for (scope_uuid, name) in work.scope_subscribers { - deregister_atif_shutdown_subscriber(&scope_uuid, &name)?; - } - for export in work.exports { - let write = prepare_atif_shutdown_file(&export, Arc::clone(&manager)) - .map_err(observability_registration_error)?; - let agent_uuid = write.agent_uuid; - let targets = { - let guard = manager.lock().map_err(|err| { + let work = match (|| -> PluginResult<_> { + let work = { + let mut guard = manager.lock().map_err(|err| { PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}")) })?; - guard.sink_targets() + guard.flush_open_agents() }; - let results = write_atif(&write, shutdown_storage.as_slice(), &targets); - let mut guard = manager.lock().map_err(|err| { - PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}")) - })?; - let _ = guard.complete_scope_write(agent_uuid, results); + for (scope_uuid, name) in &work.scope_subscribers { + deregister_atif_shutdown_subscriber(scope_uuid, name)?; + } + Ok(work) + })() { + Ok(work) => work, + Err(error) => return PluginRegistrationCleanupOutcome::NotRemoved(error), + }; + + let delivery = (|| -> PluginResult<()> { + for export in work.exports { + let write = prepare_atif_shutdown_file(&export, Arc::clone(&manager)) + .map_err(observability_registration_error)?; + let agent_uuid = write.agent_uuid; + let targets = { + let guard = manager.lock().map_err(|err| { + PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}")) + })?; + guard.sink_targets() + }; + let results = write_atif(&write, shutdown_storage.as_slice(), &targets); + let mut guard = manager.lock().map_err(|err| { + PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}")) + })?; + let _ = guard.complete_scope_write(agent_uuid, results); + } + Ok(()) + })(); + if let Err(error) = delivery { + return PluginRegistrationCleanupOutcome::RemovedWithError(error); + } + let guard = match manager.lock() { + Ok(guard) => guard, + Err(error) => { + return PluginRegistrationCleanupOutcome::RemovedWithError( + PluginError::Internal(format!("ATIF dispatcher lock poisoned: {error}")), + ); + } + }; + match guard.last_error_result() { + Ok(()) => PluginRegistrationCleanupOutcome::Removed, + Err(error) => PluginRegistrationCleanupOutcome::RemovedWithError( + observability_registration_error(error), + ), } - let guard = manager.lock().map_err(|err| { - PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}")) - })?; - guard - .last_error_result() - .map_err(observability_registration_error) }), )); Ok(()) @@ -999,10 +1021,20 @@ fn register_opentelemetry( // Retain the subscribers as long as the registered fan-out callback exists. // Their tracer providers and exporter runtimes must outlive event delivery. let delivery_subscribers = subscribers.clone(); - ctx.add_registration(PluginRegistration::new( + ctx.add_registration(PluginRegistration::new_with_outcome( "observability", ctx.qualify_name("opentelemetry.shutdown"), - Box::new(move || shutdown_opentelemetry_subscribers(&subscribers).map_or(Ok(()), Err)), + Box::new( + move || match shutdown_opentelemetry_subscribers(&subscribers) { + None => PluginRegistrationCleanupOutcome::Removed, + Some(OpenTelemetryShutdownFailure::Delivery(error)) => { + PluginRegistrationCleanupOutcome::RemovedWithError(error) + } + Some(OpenTelemetryShutdownFailure::Other(error)) => { + PluginRegistrationCleanupOutcome::NotRemoved(error) + } + }, + ), )); ctx.register_subscriber( "opentelemetry", @@ -1059,9 +1091,14 @@ fn build_opentelemetry_subscribers( Ok(subscribers) } +enum OpenTelemetryShutdownFailure { + Delivery(PluginError), + Other(PluginError), +} + fn shutdown_opentelemetry_subscribers( subscribers: &[Arc], -) -> Option { +) -> Option { let mut errors = Vec::new(); if let Err(error) = flush_subscribers() { errors.push(crate::observability::otel::OpenTelemetryError::Core(error)); @@ -1086,7 +1123,12 @@ fn shutdown_opentelemetry_subscribers( } else { format!("OpenTelemetry shutdown failures: {summary}") }; - Some(PluginError::RegistrationFailed(message)) + let error = PluginError::RegistrationFailed(message); + Some(if all_delivery_failures { + OpenTelemetryShutdownFailure::Delivery(error) + } else { + OpenTelemetryShutdownFailure::Other(error) + }) } fn shutdown_opentelemetry_providers( @@ -1624,16 +1666,31 @@ fn render_atif_filename( })?; let expression = rendered[selector_start..end].to_string(); let (selector, fallback) = parse_atif_metadata_expression(&expression)?; - let value = selector - .split('.') - .fold(metadata, |value, segment| value?.get(segment)) - .and_then(Json::as_str) - .or(fallback) - .ok_or_else(|| { + let mut resolved = metadata; + for segment in selector.split('.') { + resolved = match resolved { + Some(Json::Object(object)) => object.get(segment), + None | Some(Json::Null) => break, + Some(_) => { + return Err(format!( + "filename_template placeholder '{{metadata.{selector}}}' traversed a non-object value" + )); + } + }; + } + let value = match resolved { + Some(Json::String(value)) => value.as_str(), + None | Some(Json::Null) => fallback.ok_or_else(|| { format!( "filename_template placeholder '{{metadata.{selector}}}' must resolve to a string" ) - })?; + })?, + Some(_) => { + return Err(format!( + "filename_template placeholder '{{metadata.{selector}}}' resolved to a non-string value" + )); + } + }; if !is_safe_atif_metadata_path(value) { return Err(format!( "metadata path '{selector}' must be a path-safe relative fragment" diff --git a/crates/core/src/plugin.rs b/crates/core/src/plugin.rs index 9fbce2c81..335ec7afd 100644 --- a/crates/core/src/plugin.rs +++ b/crates/core/src/plugin.rs @@ -337,7 +337,13 @@ pub struct PluginRegistration { pub kind: String, /// Runtime-qualified registration name. pub name: String, - deregister: Box Result<()> + Send>, + deregister: Box PluginRegistrationCleanupOutcome + Send>, +} + +pub(crate) enum PluginRegistrationCleanupOutcome { + Removed, + RemovedWithError(PluginError), + NotRemoved(PluginError), } impl fmt::Debug for PluginRegistration { @@ -354,7 +360,22 @@ impl PluginRegistration { pub fn new( kind: impl Into, name: impl Into, - deregister: Box Result<()> + Send>, + mut deregister: Box Result<()> + Send>, + ) -> Self { + Self { + kind: kind.into(), + name: name.into(), + deregister: Box::new(move || match deregister() { + Ok(()) => PluginRegistrationCleanupOutcome::Removed, + Err(error) => PluginRegistrationCleanupOutcome::NotRemoved(error), + }), + } + } + + pub(crate) fn new_with_outcome( + kind: impl Into, + name: impl Into, + deregister: Box PluginRegistrationCleanupOutcome + Send>, ) -> Self { Self { kind: kind.into(), @@ -1558,12 +1579,39 @@ async fn initialize_plugins_exact_inner( }; if let Some(mut previous_state) = previous { - let teardown_errors = rollback_registrations_checked(&mut previous_state.registrations); - if !teardown_errors.is_empty() { - record_rollback_failures(rollback_failures.as_ref(), teardown_errors.clone()); + // Keep the previous report installed while teardown callbacks run so + // runtime diagnostics emitted by teardown remain observable. + { + let mut guard = ACTIVE_PLUGIN_CONFIGURATION.lock().map_err(|err| { + PluginError::Internal(format!("active plugin configuration lock poisoned: {err}")) + })?; + *guard = Some(ActivePluginConfiguration { + config: previous_state.config.clone(), + report: previous_state.report.clone(), + registrations: Vec::new(), + }); + } + let teardown = rollback_registrations_checked(&mut previous_state.registrations); + let teardown_report = ACTIVE_PLUGIN_CONFIGURATION + .lock() + .map_err(|err| { + PluginError::Internal(format!("active plugin configuration lock poisoned: {err}")) + })? + .take() + .map(|state| state.report); + if !teardown.errors.is_empty() { + if let Some(report) = + teardown_report.filter(|report| !report.runtime_diagnostics.is_empty()) + && let Ok(mut guard) = LAST_FAILED_RUNTIME_DIAGNOSTICS_REPORT.lock() + { + *guard = Some(report); + } + if !teardown.callbacks_cleared { + record_rollback_failures(rollback_failures.as_ref(), teardown.errors.clone()); + } return Err(PluginError::RegistrationFailed(format!( "previous plugin configuration could not be cleared: {}", - teardown_errors.join("; ") + teardown.errors.join("; ") ))); } match initialize_plugin_components_catching_panics( @@ -2123,7 +2171,7 @@ fn clear_plugin_configuration_inner() -> PluginHostClearOutcome { }; // Keep the report installed while callbacks run so runtime diagnostics // emitted by teardown work can be recorded against it. - let deregistration_errors = registrations + let deregistration = registrations .as_mut() .map(rollback_registrations_checked) .unwrap_or_default(); @@ -2138,16 +2186,10 @@ fn clear_plugin_configuration_inner() -> PluginHostClearOutcome { }; } }; - // Runtime delivery failures are reported by an otherwise successful - // deregistration callback. They must propagate without treating callback - // removal itself as unsafe. - let callbacks_cleared = deregistration_errors - .iter() - .all(|error| is_runtime_delivery_failure(error)); - let deregistration_error = (!deregistration_errors.is_empty()).then(|| { + let deregistration_error = (!deregistration.errors.is_empty()).then(|| { PluginError::RegistrationFailed(format!( "plugin teardown failed: {}", - deregistration_errors.join("; ") + deregistration.errors.join("; ") )) }); let result = match (flush_error, deregistration_error) { @@ -2170,19 +2212,10 @@ fn clear_plugin_configuration_inner() -> PluginHostClearOutcome { } PluginHostClearOutcome { result, - callbacks_cleared, + callbacks_cleared: deregistration.callbacks_cleared, } } -fn is_runtime_delivery_failure(error: &str) -> bool { - [ - ATIF_RUNTIME_DELIVERY_FAILURE_MARKER, - OTEL_RUNTIME_DELIVERY_FAILURE_MARKER, - ] - .iter() - .any(|marker| error.contains(&format!("registration failed: {marker}:"))) -} - pub(crate) fn plugin_configuration_is_active() -> Result { ACTIVE_PLUGIN_CONFIGURATION .lock() @@ -2342,26 +2375,53 @@ pub fn rollback_registrations(registrations: &mut Vec) { let _ = rollback_registrations_checked(registrations); } -fn rollback_registrations_checked(registrations: &mut Vec) -> Vec { - let mut errors = Vec::new(); +struct PluginRollbackOutcome { + errors: Vec, + callbacks_cleared: bool, +} + +impl Default for PluginRollbackOutcome { + fn default() -> Self { + Self { + errors: Vec::new(), + callbacks_cleared: true, + } + } +} + +fn rollback_registrations_checked( + registrations: &mut Vec, +) -> PluginRollbackOutcome { + let mut outcome = PluginRollbackOutcome::default(); for registration in registrations.iter_mut().rev() { - let failure = match catch_unwind(AssertUnwindSafe(|| (registration.deregister)())) { - Ok(Ok(())) => None, - Ok(Err(error)) => Some(error.to_string()), - Err(payload) => Some(format!( - "deregistration panicked: {}", - panic_payload_message(payload) - )), - }; - if let Some(error) = failure { - errors.push(format!( - "{} registration '{}' could not be removed: {error}", - registration.kind, registration.name - )); + match catch_unwind(AssertUnwindSafe(|| (registration.deregister)())) { + Ok(PluginRegistrationCleanupOutcome::Removed) => {} + Ok(PluginRegistrationCleanupOutcome::RemovedWithError(error)) => { + outcome.errors.push(format!( + "{} registration '{}' reported a delivery failure: {error}", + registration.kind, registration.name + )); + } + Ok(PluginRegistrationCleanupOutcome::NotRemoved(error)) => { + outcome.callbacks_cleared = false; + outcome.errors.push(format!( + "{} registration '{}' could not be removed: {error}", + registration.kind, registration.name + )); + } + Err(payload) => { + outcome.callbacks_cleared = false; + outcome.errors.push(format!( + "{} registration '{}' could not be removed: deregistration panicked: {}", + registration.kind, + registration.name, + panic_payload_message(payload) + )); + } } } registrations.clear(); - errors + outcome } fn panic_payload_message(payload: Box) -> String { @@ -2444,8 +2504,10 @@ impl PendingPluginRegistrations { impl Drop for PendingPluginRegistrations { fn drop(&mut self) { - let errors = rollback_registrations_checked(&mut self.registrations); - record_rollback_failures(self.rollback_failures.as_ref(), errors); + let outcome = rollback_registrations_checked(&mut self.registrations); + if !outcome.callbacks_cleared { + record_rollback_failures(self.rollback_failures.as_ref(), outcome.errors); + } } } @@ -2469,8 +2531,10 @@ impl PendingPluginRegistrationContext { impl Drop for PendingPluginRegistrationContext { fn drop(&mut self) { - let errors = rollback_registrations_checked(&mut self.context.registrations); - record_rollback_failures(self.rollback_failures.as_ref(), errors); + let outcome = rollback_registrations_checked(&mut self.context.registrations); + if !outcome.callbacks_cleared { + record_rollback_failures(self.rollback_failures.as_ref(), outcome.errors); + } } } diff --git a/crates/core/tests/integration/atif_storage_tests.rs b/crates/core/tests/integration/atif_storage_tests.rs index 09d3d9f04..84a1083b0 100644 --- a/crates/core/tests/integration/atif_storage_tests.rs +++ b/crates/core/tests/integration/atif_storage_tests.rs @@ -23,7 +23,8 @@ use nemo_relay::api::scope::{PopScopeParams, PushScopeParams, ScopeType, pop_sco use nemo_relay::api::subscriber::flush_subscribers; use nemo_relay::observability::plugin_component::OBSERVABILITY_PLUGIN_KIND; use nemo_relay::plugin::{ - PluginComponentSpec, PluginConfig, clear_plugin_configuration, initialize_plugins, + PluginComponentSpec, PluginConfig, active_plugin_report, clear_plugin_configuration, + initialize_plugins, }; use object_store::{ObjectStore, ObjectStoreExt as _}; use serde_json::{Value as Json, json}; @@ -497,13 +498,18 @@ fn atif_storage_posts_trajectory_to_http_endpoints() { fn atif_storage_http_non_2xx_retries_on_the_next_trajectory() { let _guard = PLUGIN_TEST_LOCK.lock().unwrap(); reset_runtime(); + let recovery_directory = tempfile::tempdir().expect("create recovery directory"); let mut server = start_http_server(2, vec![("/fail", 500)]); // SAFETY: this uniquely named env var is only touched by this test. unsafe { std::env::set_var("NEMO_RELAY_ATIF_HTTP_TEST_TOKEN", "Bearer test-token"); } - let config = build_http_observability_config(&[format!("{}/fail", server.base_url)]); + let mut config = build_http_observability_config(&[format!("{}/fail", server.base_url)]); + config.components[0].config["atif"] + .as_object_mut() + .expect("ATIF config should be an object") + .insert("output_directory".into(), json!(recovery_directory.path())); futures::executor::block_on(initialize_plugins(config)) .expect("observability plugin should initialize with HTTP storage"); @@ -529,6 +535,15 @@ fn atif_storage_http_non_2xx_retries_on_the_next_trajectory() { .expect("pop second agent scope"); flush_subscribers().expect("HTTP upload subscriber should flush after failure"); + let report = active_plugin_report().expect("active plugin report should remain readable"); + let diagnostic = report + .runtime_diagnostics + .iter() + .find(|diagnostic| diagnostic.code == "atif.remote_delivery_failed") + .expect("failed uploads should be visible before teardown"); + assert_eq!(diagnostic.field.as_deref(), Some("storage[0]")); + assert_eq!(diagnostic.count, 2); + server.stop(); { let requests = server.received.lock().unwrap(); @@ -557,6 +572,19 @@ fn atif_storage_http_non_2xx_retries_on_the_next_trajectory() { teardown.to_string().contains("atif.remote_delivery_failed"), "teardown should identify the failed remote destination: {teardown}" ); + assert!( + !teardown.to_string().contains("could not be removed"), + "delivery failure should not imply a leaked registration: {teardown}" + ); + let retained_report = + active_plugin_report().expect("failed teardown should retain the plugin report"); + let retained_diagnostic = retained_report + .runtime_diagnostics + .iter() + .find(|diagnostic| diagnostic.code == "atif.remote_delivery_failed") + .expect("failed upload diagnostic should remain readable after teardown"); + assert_eq!(retained_diagnostic.field.as_deref(), Some("storage[0]")); + assert_eq!(retained_diagnostic.count, 2); // SAFETY: cleanup of test-only env var. unsafe { std::env::remove_var("NEMO_RELAY_ATIF_HTTP_TEST_TOKEN"); diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index 48f3f6775..abddf29be 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -2447,6 +2447,35 @@ fn atif_metadata_template_values_must_be_safe_path_fragments() { .prepare_destination("session-1", Some(&non_string)) .is_err() ); + let dispatcher_with_fallback = AtifDispatcher::new(AtifSectionConfig { + filename_template: "{metadata.artifact_path:-unassigned}/trajectory-{session_id}.json" + .to_string(), + ..AtifSectionConfig::default() + }); + let error = dispatcher_with_fallback + .prepare_destination("session-1", Some(&non_string)) + .unwrap_err(); + assert!(error.contains("resolved to a non-string value"), "{error}"); + let nested_non_string = json!({"artifact": 123}); + let nested_dispatcher = AtifDispatcher::new(AtifSectionConfig { + filename_template: "{metadata.artifact.path:-unassigned}/trajectory-{session_id}.json" + .to_string(), + ..AtifSectionConfig::default() + }); + let error = nested_dispatcher + .prepare_destination("session-1", Some(&nested_non_string)) + .unwrap_err(); + assert!(error.contains("traversed a non-object value"), "{error}"); + let nested_null = json!({"artifact": null}); + let destination = nested_dispatcher + .prepare_destination("session-1", Some(&nested_null)) + .unwrap(); + assert_eq!(destination.0, "unassigned/trajectory-session-1.json"); + let nested_string = json!({"artifact": {"path": "tenant-a/team_1"}}); + let destination = nested_dispatcher + .prepare_destination("session-1", Some(&nested_string)) + .unwrap(); + assert_eq!(destination.0, "tenant-a/team_1/trajectory-session-1.json"); for template in [ "/tmp/trajectory-{session_id}.json", @@ -3037,9 +3066,13 @@ fn opentelemetry_shutdown_helper_retains_every_endpoint_failure() { }) .collect::>(); - let error = shutdown_opentelemetry_subscribers(&subscribers) - .expect("mixed endpoint shutdown failures should be reported") - .to_string(); + let OpenTelemetryShutdownFailure::Other(error) = + shutdown_opentelemetry_subscribers(&subscribers) + .expect("mixed endpoint shutdown failures should be reported") + else { + panic!("mixed endpoint shutdown failures must retain the registration failure outcome"); + }; + let error = error.to_string(); assert_eq!(dropped_calls.load(Ordering::SeqCst), 1); assert_eq!(timeout_calls.load(Ordering::SeqCst), 1); diff --git a/crates/core/tests/unit/plugin_tests.rs b/crates/core/tests/unit/plugin_tests.rs index 816f2ab8d..815239734 100644 --- a/crates/core/tests/unit/plugin_tests.rs +++ b/crates/core/tests/unit/plugin_tests.rs @@ -1477,6 +1477,37 @@ fn test_pending_registration_records_rollback_failures() { assert!(failures[0].contains("rollback remained registered")); } +#[test] +fn test_pending_rollbacks_ignore_delivery_only_errors() { + let failures = Arc::new(Mutex::new(Vec::new())); + let delivery_error = || { + PluginRegistrationCleanupOutcome::RemovedWithError(PluginError::RegistrationFailed( + "delivery failed".into(), + )) + }; + { + let mut pending = PendingPluginRegistrations::new(Some(Arc::clone(&failures))); + pending.extend(vec![PluginRegistration::new_with_outcome( + "fixture", + "delivery-only-registration", + Box::new(delivery_error), + )]); + } + { + let mut pending = + PendingPluginRegistrationContext::new("fixture.".into(), Some(Arc::clone(&failures))); + pending + .context + .add_registration(PluginRegistration::new_with_outcome( + "fixture", + "delivery-only-context-registration", + Box::new(delivery_error), + )); + } + + assert!(failures.lock().unwrap().is_empty()); +} + #[test] fn test_checked_teardown_reports_unremoved_registrations() { let _guard = lock_runtime_owner(); @@ -1506,13 +1537,41 @@ fn test_checked_teardown_reports_unremoved_registrations() { } #[test] -fn test_teardown_runtime_diagnostics_remain_in_the_plugin_report() { +fn test_teardown_marker_text_does_not_imply_successful_removal() { let _guard = lock_runtime_owner(); reset_global(); store_active_plugin_configuration( PluginConfig::default(), ConfigReport::default(), vec![PluginRegistration::new( + "fixture", + "stale-marker-callback", + Box::new(|| { + Err(PluginError::RegistrationFailed(format!( + "unrelated failure mentioning {}", + crate::plugin::ATIF_RUNTIME_DELIVERY_FAILURE_MARKER + ))) + }), + )], + ) + .unwrap(); + + let outcome = clear_plugin_configuration_inner(); + assert!(!outcome.callbacks_cleared); + let error = outcome.result.unwrap_err().to_string(); + assert!(error.contains("stale-marker-callback"), "{error}"); + assert!(error.contains("could not be removed"), "{error}"); + reset_global(); +} + +#[test] +fn test_teardown_runtime_diagnostics_remain_in_the_plugin_report() { + let _guard = lock_runtime_owner(); + reset_global(); + store_active_plugin_configuration( + PluginConfig::default(), + ConfigReport::default(), + vec![PluginRegistration::new_with_outcome( "fixture", "atif-shutdown", Box::new(|| { @@ -1524,10 +1583,11 @@ fn test_teardown_runtime_diagnostics_remain_in_the_plugin_report() { session_id: Some("session-123".into()), count: 1, }); - Err(PluginError::RegistrationFailed(format!( + let error = PluginError::RegistrationFailed(format!( "{}: atif.remote_delivery_failed (1)", crate::plugin::ATIF_RUNTIME_DELIVERY_FAILURE_MARKER - ))) + )); + PluginRegistrationCleanupOutcome::RemovedWithError(error) }), )], ) @@ -1535,7 +1595,9 @@ fn test_teardown_runtime_diagnostics_remain_in_the_plugin_report() { let outcome = clear_plugin_configuration_inner(); assert!(outcome.callbacks_cleared); - assert!(outcome.result.is_err()); + let error = outcome.result.unwrap_err().to_string(); + assert!(error.contains("atif.remote_delivery_failed"), "{error}"); + assert!(!error.contains("could not be removed"), "{error}"); let report = active_plugin_report().expect("failed teardown should retain its report"); assert_eq!(report.runtime_diagnostics.len(), 1); let diagnostic = &report.runtime_diagnostics[0]; @@ -1556,14 +1618,16 @@ fn test_opentelemetry_delivery_failure_allows_later_plugin_configuration() { store_active_plugin_configuration( PluginConfig::default(), ConfigReport::default(), - vec![PluginRegistration::new( + vec![PluginRegistration::new_with_outcome( "fixture", "opentelemetry-shutdown", Box::new(|| { - Err(PluginError::RegistrationFailed(format!( - "{}: otel.spans_dropped (2)", - crate::plugin::OTEL_RUNTIME_DELIVERY_FAILURE_MARKER - ))) + PluginRegistrationCleanupOutcome::RemovedWithError(PluginError::RegistrationFailed( + format!( + "{}: otel.spans_dropped (2)", + crate::plugin::OTEL_RUNTIME_DELIVERY_FAILURE_MARKER + ), + )) }), )], ) @@ -1582,14 +1646,16 @@ fn test_mixed_opentelemetry_shutdown_failure_blocks_later_configuration() { store_active_plugin_configuration( PluginConfig::default(), ConfigReport::default(), - vec![PluginRegistration::new( + vec![PluginRegistration::new_with_outcome( "fixture", "opentelemetry-shutdown", Box::new(|| { - Err(PluginError::RegistrationFailed(format!( - "OpenTelemetry shutdown failures: provider error: {}: otel.spans_dropped (2); endpoint shutdown timed out", - crate::plugin::OTEL_RUNTIME_DELIVERY_FAILURE_MARKER - ))) + PluginRegistrationCleanupOutcome::NotRemoved(PluginError::RegistrationFailed( + format!( + "OpenTelemetry shutdown failures: provider error: {}: otel.spans_dropped (2); endpoint shutdown timed out", + crate::plugin::OTEL_RUNTIME_DELIVERY_FAILURE_MARKER + ), + )) }), )], ) @@ -1602,6 +1668,72 @@ fn test_mixed_opentelemetry_shutdown_failure_blocks_later_configuration() { reset_global(); } +#[test] +fn test_replacement_teardown_runtime_diagnostics_remain_in_the_plugin_report() { + let _guard = lock_runtime_owner(); + reset_global(); + store_active_plugin_configuration( + PluginConfig::default(), + ConfigReport::default(), + vec![PluginRegistration::new_with_outcome( + "fixture", + "atif-shutdown", + Box::new(|| { + record_active_plugin_runtime_diagnostic(RuntimeDiagnostic { + code: "atif.remote_delivery_failed".into(), + component: "observability".into(), + field: Some("storage[0]".into()), + message: "HTTP 500".into(), + session_id: Some("session-123".into()), + count: 1, + }); + PluginRegistrationCleanupOutcome::RemovedWithError(PluginError::RegistrationFailed( + "ATIF delivery failed".into(), + )) + }), + )], + ) + .unwrap(); + + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let error = runtime + .block_on(initialize_plugins_exact(PluginConfig::default())) + .unwrap_err(); + assert!( + error.to_string().contains("ATIF delivery failed"), + "{error}" + ); + assert!( + error + .to_string() + .contains("fixture registration 'atif-shutdown' reported a delivery failure"), + "{error}" + ); + assert!( + !plugin_configuration_is_active().unwrap(), + "a replacement aborted by delivery failure must not leave a configuration active" + ); + let report = active_plugin_report().expect("failed replacement should retain its report"); + assert_eq!(report.runtime_diagnostics.len(), 1); + assert_eq!( + report.runtime_diagnostics[0].code, + "atif.remote_delivery_failed" + ); + assert_eq!( + report.runtime_diagnostics[0].field.as_deref(), + Some("storage[0]") + ); + assert_eq!(report.runtime_diagnostics[0].count, 1); + + runtime + .block_on(initialize_plugins_exact(PluginConfig::default())) + .expect("delivery-only teardown errors must not block a later initialization"); + reset_global(); +} + #[test] fn test_legacy_clear_retains_mutation_owner_after_incomplete_teardown() { let _guard = lock_runtime_owner(); @@ -2207,11 +2339,18 @@ fn test_plugin_registration_context_maps_deregistration_errors() { set_conflicting_runtime_owner_for_tests(); for (registration, expected) in registrations.iter_mut().zip(expected_messages) { match (registration.deregister)() { - Err(PluginError::RegistrationFailed(message)) => { + PluginRegistrationCleanupOutcome::NotRemoved(PluginError::RegistrationFailed( + message, + )) => { assert!(message.contains(expected), "{message}"); } - Err(other) => panic!("unexpected deregistration failure: {other}"), - Ok(()) => panic!("expected deregistration to fail"), + PluginRegistrationCleanupOutcome::NotRemoved(other) => { + panic!("unexpected deregistration failure: {other}") + } + PluginRegistrationCleanupOutcome::Removed + | PluginRegistrationCleanupOutcome::RemovedWithError(_) => { + panic!("expected deregistration to fail") + } } } diff --git a/crates/python/src/py_plugin.rs b/crates/python/src/py_plugin.rs index 281ae0eca..ec11ce32e 100644 --- a/crates/python/src/py_plugin.rs +++ b/crates/python/src/py_plugin.rs @@ -933,16 +933,20 @@ impl PluginConfigurationClearState { let result = std::panic::catch_unwind(clear_plugin_configuration) .map_err(|_| PluginTeardownError::runtime("plugin teardown task panicked")) .and_then(|result| result.map_err(PluginTeardownError::from_plugin_error)); - clear_state.completion.finish(result); + clear_state.finish(result); }); if let Err(error) = spawn { - self.completion - .finish(Err(PluginTeardownError::runtime(format!( - "failed to start plugin teardown task: {error}" - )))); + self.finish(Err(PluginTeardownError::runtime(format!( + "failed to start plugin teardown task: {error}" + )))); } } + fn finish(self: &Arc, result: PluginTeardownResult) { + reset_plugin_configuration_clear_state_if(self); + self.completion.finish(result); + } + async fn wait_for_clear(&self) -> PluginTeardownResult { self.completion.wait("plugin teardown").await } @@ -965,6 +969,15 @@ fn reset_plugin_configuration_clear_state() { Arc::new(PluginConfigurationClearState::new()); } +fn reset_plugin_configuration_clear_state_if(completed: &Arc) { + let mut current = PLUGIN_CONFIGURATION_CLEAR_STATE + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if Arc::ptr_eq(¤t, completed) { + *current = Arc::new(PluginConfigurationClearState::new()); + } +} + #[pymethods] impl PyPluginHostActivation { /// Return the activation report captured during initialization. diff --git a/crates/python/tests/coverage/py_plugin_coverage_tests.rs b/crates/python/tests/coverage/py_plugin_coverage_tests.rs index 88127cdea..b8818295b 100644 --- a/crates/python/tests/coverage/py_plugin_coverage_tests.rs +++ b/crates/python/tests/coverage/py_plugin_coverage_tests.rs @@ -165,6 +165,7 @@ fn async_clear_binding_completes_on_python_event_loop() { let _python = crate::test_support::init_python_test(); let _plugin_test_state = lock_plugin_test_state_for_tests(); Python::attach(|py| { + let first_clear_state = plugin_configuration_clear_state(); let module = PyModule::new(py, "_plugin_async_clear").unwrap(); register(&module).unwrap(); let helpers = load_module( @@ -174,6 +175,19 @@ async def clear(module): await module.clear_plugin_configuration_async() "#, ); + with_event_loop(py, |event_loop| { + let clear = helpers + .getattr("clear") + .unwrap() + .call1((module.clone(),)) + .unwrap(); + event_loop + .call_method1("run_until_complete", (clear,)) + .unwrap(); + }); + let second_clear_state = plugin_configuration_clear_state(); + assert!(!Arc::ptr_eq(&first_clear_state, &second_clear_state)); + with_event_loop(py, |event_loop| { let clear = helpers.getattr("clear").unwrap().call1((module,)).unwrap(); event_loop @@ -184,6 +198,22 @@ async def clear(module): }); } +#[test] +fn stale_async_clear_completion_keeps_the_newer_state() { + let _plugin_test_state = lock_plugin_test_state_for_tests(); + let older = plugin_configuration_clear_state(); + let newer = Arc::new(PluginConfigurationClearState::new()); + *PLUGIN_CONFIGURATION_CLEAR_STATE + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Arc::clone(&newer); + + reset_plugin_configuration_clear_state_if(&older); + + let newer_state_remained_current = Arc::ptr_eq(&plugin_configuration_clear_state(), &newer); + reset_plugin_configuration_clear_state(); + assert!(newer_state_remained_current); +} + #[test] fn plugin_context_registers_all_runtime_hooks_and_drains_registrations() { let _python = crate::test_support::init_python_test(); diff --git a/docs/configure-plugins/observability/atif.mdx b/docs/configure-plugins/observability/atif.mdx index dec1ad05c..077d11c2d 100644 --- a/docs/configure-plugins/observability/atif.mdx +++ b/docs/configure-plugins/observability/atif.mdx @@ -89,18 +89,20 @@ With scope metadata `{"atif_prefix":"tenant-a/session-123"}`, Relay writes still contain `{session_id}`. Use `:-` to provide a literal fallback when metadata can be absent, for example -`{metadata.atif_prefix:-unassigned}`. Relay also uses the fallback when the -metadata value is not a string. +`{metadata.atif_prefix:-unassigned}`. Relay uses the fallback when the selected +metadata field is absent or `null`. A present non-string value is rejected +instead of being routed through the fallback. Each metadata placeholder must resolve to a string containing a non-empty, relative path fragment. Slash-separated segments can contain ASCII letters, digits, `-`, `_`, `.`, and `~`. Relay rejects empty segments, `.` and `..` segments, absolute paths, backslashes, spaces, and other characters. If a -placeholder has no fallback and is missing or non-string, or if its resolved -value is unsafe, Relay skips that trajectory. It records a runtime diagnostic -in `plugin.report()` and causes plugin teardown to fail. Literal template text -must also be a relative, traversal-free path. The rendered filename applies to -local, S3, and HTTP storage in the same way as a static filename. +placeholder is missing without a fallback, is present but non-string, or +resolves to an unsafe value, Relay skips that trajectory. It records a runtime +diagnostic in `plugin.report()` and causes plugin teardown to fail. Literal +template text must also be a relative, traversal-free path. The rendered +filename applies to local, S3, and HTTP storage in the same way as a static +filename. The CLI gateway parses `x-nemo-relay-session-metadata` as JSON and merges it into the top-level scope metadata: @@ -245,10 +247,15 @@ delivery diagnostic and retries that destination for each later trajectory. Other destinations continue to receive writes. When every remote destination fails for one trajectory, Relay writes a local recovery copy under `output_directory`; a successful remote destination does not create that copy. -The diagnostic remains visible in `plugin.report()` and is retained there after -a failed teardown. Teardown also reports the degraded delivery, even when the -local recovery write succeeds. Fatal dispatcher failures, such as trajectory -serialization failures, are also reported during teardown. +The diagnostic becomes visible in `plugin.report()` after subscriber delivery +is flushed and is retained there after a failed teardown. Teardown also reports +the degraded delivery, even when the local recovery write succeeds. This error +reports delivery degradation, not a registration leak; callbacks have already +been removed. If the failure is still pending when initialization replaces the +configuration, that replacement returns the delivery error and leaves no +configuration active; a subsequent clear or initialization is safe. Fatal +dispatcher failures, such as trajectory serialization failures, are also +reported during teardown. ## Expected Output diff --git a/python/tests/test_observability_plugin.py b/python/tests/test_observability_plugin.py index 68b7dd3b6..b457f703a 100644 --- a/python/tests/test_observability_plugin.py +++ b/python/tests/test_observability_plugin.py @@ -14,7 +14,7 @@ import pytest -from nemo_relay import ScopeType, plugin, scope +from nemo_relay import ScopeType, plugin, scope, subscribers from nemo_relay.observability import ( OBSERVABILITY_PLUGIN_KIND, AtifConfig, @@ -439,6 +439,51 @@ async def test_atif_flushes_open_agent_on_clear(self, tmp_path): finally: scope.pop(handle) + async def test_atif_non_string_metadata_is_reported_and_failed_clear_is_drainable(self, tmp_path): + await plugin.initialize( + plugin.PluginConfig( + components=[ + ComponentSpec( + ObservabilityConfig( + atif=AtifConfig( + enabled=True, + output_directory=str(tmp_path), + filename_template="{metadata.atif_prefix:-unassigned}/trajectory-{session_id}.json", + ) + ) + ) + ] + ) + ) + + try: + with scope.scope("python-invalid-metadata-agent", ScopeType.Agent, metadata={"atif_prefix": 123}): + pass + await subscribers.flush_async() + + report = plugin.report() + assert report is not None + assert any( + diagnostic["code"] == "atif.destination_render_failed" and "non-string" in diagnostic["message"] + for diagnostic in report["runtime_diagnostics"] + ) + assert not (tmp_path / "unassigned").exists() + + with pytest.raises(RuntimeError, match=r"atif\.destination_render_failed") as teardown: + await plugin.clear_async() + assert "could not be removed" not in str(teardown.value) + retained = plugin.report() + assert retained is not None + assert any( + diagnostic["code"] == "atif.destination_render_failed" for diagnostic in retained["runtime_diagnostics"] + ) + finally: + try: + await plugin.clear_async() + except RuntimeError: + await plugin.clear_async() + assert plugin.report() is None + async def test_atif_splits_multiple_top_level_agent_scopes(self, tmp_path): await plugin.initialize( plugin.PluginConfig(