diff --git a/.claude/skills/work-with-subscriptions/SKILL.md b/.claude/skills/work-with-subscriptions/SKILL.md index a7e69ba47..6ef92d3f3 100644 --- a/.claude/skills/work-with-subscriptions/SKILL.md +++ b/.claude/skills/work-with-subscriptions/SKILL.md @@ -28,9 +28,43 @@ Use this when working in `helios-subscriptions`, the FHIR topic-based Subscripti | `HFS_SUBSCRIPTION_MESSAGING_ENABLED` | Enable the messaging channel | | `HFS_SUBSCRIPTION_ALLOW_PRIVATE_ENDPOINTS` | Allow rest-hook delivery to private/loopback URLs (opt-in) | | `HFS_BASE_URL` | Base URL used in notification links | +| `HFS_SUBSCRIPTION_PERSIST_STATUS` | Write server-driven status transitions back into the stored `Subscription` (default `true`) | +| `HFS_SUBSCRIPTION_STATUS_WRITE_TIMEOUT_MS` | Per-write timeout for that write-back (default `5000`) | +| `HFS_SUBSCRIPTION_REHYDRATE` | Replay stored topics/subscriptions into the engine at startup (default `true`) | +| `HFS_SUBSCRIPTION_REHYDRATE_HANDSHAKE` | Re-run activation for stored `requested`/`error` subscriptions (default `true`) | +| `HFS_SUBSCRIPTION_REHYDRATE_HANDSHAKE_CONCURRENCY` | Cap on concurrent startup handshakes (default `8`) | +| `HFS_SUBSCRIPTION_REHYDRATE_BATCH_SIZE` | Resources fetched per storage round-trip during rehydration (default `500`) | Exact defaults are read in `crates/rest/src/lib.rs` and `crates/subscriptions/src/config.rs`. +## Status persistence (issue #357) + +Server-driven status transitions (`requested` → `active` after a successful +handshake, `→ error`, `→ off` when the delivery circuit breaker trips) are +written back into the stored `Subscription` resource through the ordinary +`ResourceStorage::update`, so the behaviour is identical on SQLite, PostgreSQL, +MongoDB, S3, and composite deployments with no schema change. + +Things to know when working here: + +- The write happens only when `SubscriptionManager::update_status` accepts the + transition, so it is bounded to at most three writes per subscription + lifetime. `SubscriptionEngine::transition_status` is the single choke point — + do not call `manager.update_status` directly from the engine. +- It **bumps `meta.versionId`** and appends to `_history`, so a client holding + an ETag from its own POST can see `412` on a later `If-Match` write. That is + correct optimistic-locking behaviour, not a bug. +- It fails **open**: a storage error leaves the in-memory transition in place + and logs at `error`. Grep for `Failed to persist subscription status + transition` to spot a server that has silently reverted to losing its + decisions on restart. +- `off` and `entered-in-error` are terminal and are rehydrated **dormant** + (registered so `$status` still answers, never matched, never handshaken). + `error` is re-activated on restart, because nothing else ever performs + `error` → `active`. +- `$status` / `$events` fall back to the stored resource when the engine does + not hold the subscription, instead of `404`. + ## Key API `SubscriptionEngine`, `SubscriptionConfig` / `MessagingSettings`, `ResourceEvent` / `ResourceEventType`, `WebSocketManager`, `WsBindingTokenManager`, `MessagingChannel`, `SubscriptionError`. diff --git a/crates/rest/src/handlers/subscriptions.rs b/crates/rest/src/handlers/subscriptions.rs index 725fa8ffe..f11ec872e 100644 --- a/crates/rest/src/handlers/subscriptions.rs +++ b/crates/rest/src/handlers/subscriptions.rs @@ -39,17 +39,108 @@ where feature: "Subscriptions".to_string(), })?; - let sub = engine - .manager() - .get_subscription(tenant.tenant_id(), &id) - .ok_or(RestError::NotFound { + let snapshot = resolve_snapshot(&state, engine, &tenant, &id).await?; + + let status_resource = build_subscription_status(&snapshot, &id, state.base_url()); + + Ok((StatusCode::OK, Json(status_resource)).into_response()) +} + +/// The runtime facts `$status` / `$events` report, from whichever source can +/// supply them. +struct StatusSnapshot { + status: helios_subscriptions::manager::SubscriptionStatusCode, + topic_url: String, + events_since_start: u64, + fhir_version: helios_fhir::FhirVersion, +} + +/// Resolves a subscription's runtime state, falling back to storage when the +/// engine does not hold it. +/// +/// The engine is authoritative when it has the subscription: that entry is what +/// delivery decisions are actually made from, so reporting anything else would +/// be a worse lie than a lag. +/// +/// The fallback exists because the engine legitimately does not hold every +/// stored subscription. Rehydration can fail for one tenant, an operator can +/// disable it (`HFS_SUBSCRIPTION_REHYDRATE=false`), or a topic can fail to +/// parse. Answering `404` in those cases while `GET /Subscription/{id}` answers +/// `200` is a self-contradiction the client cannot act on — and the R5 Backport +/// IG asks a server to keep responding to `$status` regardless. Falling back to +/// the stored resource turns "I have lost track of this" into the honest +/// "here is what the record says", with the event counter reported as `0` +/// because this process genuinely has not delivered anything for it. +/// +/// The storage read uses the request's own resolved [`TenantExtractor`] context, +/// never a fabricated full-access one: this is a request path, so the tenant and +/// its permissions must be the ones the request actually resolved to. +async fn resolve_snapshot( + state: &AppState, + engine: &std::sync::Arc, + tenant: &TenantExtractor, + id: &str, +) -> RestResult +where + S: ResourceStorage + Send + Sync, +{ + if let Some(sub) = engine.manager().get_subscription(tenant.tenant_id(), id) { + return Ok(StatusSnapshot { + status: sub.status, + topic_url: sub.topic_url, + events_since_start: sub.events_since_start, + fhir_version: sub.fhir_version, + }); + } + + let stored = state + .storage() + .read(tenant.context(), "Subscription", id) + .await? + .ok_or_else(|| RestError::NotFound { resource_type: "Subscription".to_string(), - id: id.clone(), + id: id.to_string(), })?; - let status_resource = build_subscription_status(&sub, &id, state.base_url()); + Ok(snapshot_from_stored( + stored.content(), + stored.fhir_version(), + )) +} - Ok((StatusCode::OK, Json(status_resource)).into_response()) +/// Derives a [`StatusSnapshot`] from a stored `Subscription`'s content. +/// +/// Split out from [`resolve_snapshot`] so the parsing — the only part with real +/// branching — is unit-testable without standing up an app and an engine. +/// +/// `events_since_start` is reported as `0` rather than guessed: this process has +/// genuinely delivered nothing for a subscription it is not holding, and the +/// counter is not persisted (see the PR for #357). +fn snapshot_from_stored( + content: &serde_json::Value, + fhir_version: helios_fhir::FhirVersion, +) -> StatusSnapshot { + let status = content + .get("status") + .and_then(|v| v.as_str()) + .and_then(helios_subscriptions::manager::SubscriptionStatusCode::from_fhir_str) + .unwrap_or(helios_subscriptions::manager::SubscriptionStatusCode::Requested); + + // `topic` (R4B+) or `criteria` (R4 backport); mirrors the manager's own + // version-aware extraction closely enough for a status report. + let topic_url = content + .get("topic") + .and_then(|v| v.as_str()) + .or_else(|| content.get("criteria").and_then(|v| v.as_str())) + .unwrap_or_default() + .to_string(); + + StatusSnapshot { + status, + topic_url, + events_since_start: 0, + fhir_version, + } } /// Handler for the `$events` operation on Subscription resources. @@ -74,16 +165,12 @@ where feature: "Subscriptions".to_string(), })?; - let sub = engine - .manager() - .get_subscription(tenant.tenant_id(), &id) - .ok_or(RestError::NotFound { - resource_type: "Subscription".to_string(), - id: id.clone(), - })?; + // Same storage fallback as `$status`: both handlers had the identical miss, + // and fixing only one would look complete in review. + let snapshot = resolve_snapshot(&state, engine, &tenant, &id).await?; // Return a Bundle with a SubscriptionStatus indicating query-status - let bundle_type = sub.fhir_version.notification_bundle_type(); + let bundle_type = snapshot.fhir_version.notification_bundle_type(); let bundle = json!({ "resourceType": "Bundle", @@ -91,13 +178,13 @@ where "entry": [{ "resource": { "resourceType": "SubscriptionStatus", - "status": sub.status.as_fhir_str(), + "status": snapshot.status.as_fhir_str(), "type": "query-status", - "eventsSinceSubscriptionStart": sub.events_since_start, + "eventsSinceSubscriptionStart": snapshot.events_since_start, "subscription": { "reference": format!("Subscription/{}", id) }, - "topic": sub.topic_url + "topic": snapshot.topic_url } }] }); @@ -182,11 +269,7 @@ where } /// Builds a SubscriptionStatus resource for the $status response. -fn build_subscription_status( - sub: &helios_subscriptions::manager::ActiveSubscription, - id: &str, - base_url: &str, -) -> serde_json::Value { +fn build_subscription_status(sub: &StatusSnapshot, id: &str, base_url: &str) -> serde_json::Value { if uses_backport_ig(sub.fhir_version) { // R4 backport: return Parameters resource json!({ @@ -236,3 +319,75 @@ fn build_subscription_status( fn uses_backport_ig(version: helios_fhir::FhirVersion) -> bool { version.as_str() == "R4" } + +#[cfg(test)] +mod tests { + use super::*; + use helios_fhir::FhirVersion; + use helios_subscriptions::manager::SubscriptionStatusCode; + use serde_json::json; + + /// The `$status` / `$events` storage fallback (#357): when the engine does + /// not hold a subscription, its stored status is reported rather than `404`. + #[test] + fn snapshot_reads_status_and_native_topic() { + let snapshot = snapshot_from_stored( + &json!({ + "resourceType": "Subscription", + "status": "off", + "topic": "http://example.org/topic/x", + "channelType": { "code": "rest-hook" } + }), + FhirVersion::default(), + ); + assert_eq!(snapshot.status, SubscriptionStatusCode::Off); + assert_eq!(snapshot.topic_url, "http://example.org/topic/x"); + assert_eq!(snapshot.events_since_start, 0); + } + + /// R4 backport carries the topic in `criteria`, not `topic`. + #[test] + fn snapshot_reads_r4_backport_criteria_topic() { + let snapshot = snapshot_from_stored( + &json!({ + "resourceType": "Subscription", + "status": "error", + "criteria": "http://example.org/topic/y", + "channel": { "type": "rest-hook" } + }), + FhirVersion::default(), + ); + assert_eq!(snapshot.status, SubscriptionStatusCode::Error); + assert_eq!(snapshot.topic_url, "http://example.org/topic/y"); + } + + /// `entered-in-error` must survive the fallback as itself, not decay to + /// `requested` — the same trap the manager had before #357. + #[test] + fn snapshot_preserves_entered_in_error() { + let snapshot = snapshot_from_stored( + &json!({ "resourceType": "Subscription", "status": "entered-in-error" }), + FhirVersion::default(), + ); + assert_eq!(snapshot.status, SubscriptionStatusCode::EnteredInError); + } + + /// An absent or unrecognised status falls back to `requested`, and a missing + /// topic yields an empty string rather than panicking. + #[test] + fn snapshot_tolerates_missing_and_unknown_fields() { + let snapshot = snapshot_from_stored( + &json!({ "resourceType": "Subscription", "status": "not-a-code" }), + FhirVersion::default(), + ); + assert_eq!(snapshot.status, SubscriptionStatusCode::Requested); + assert_eq!(snapshot.topic_url, ""); + + let bare = snapshot_from_stored( + &json!({ "resourceType": "Subscription" }), + FhirVersion::default(), + ); + assert_eq!(bare.status, SubscriptionStatusCode::Requested); + assert_eq!(bare.topic_url, ""); + } +} diff --git a/crates/rest/src/lib.rs b/crates/rest/src/lib.rs index 55db475dd..7509a8ba6 100644 --- a/crates/rest/src/lib.rs +++ b/crates/rest/src/lib.rs @@ -801,6 +801,14 @@ where "HFS_SUBSCRIPTION_HANDSHAKE_RETRY_MAX_MS", default_sub_config.handshake_retry_max_delay, ), + persist_status: subscription_bool_from_env( + "HFS_SUBSCRIPTION_PERSIST_STATUS", + default_sub_config.persist_status, + ), + status_write_timeout: subscription_duration_ms_from_env( + "HFS_SUBSCRIPTION_STATUS_WRITE_TIMEOUT_MS", + default_sub_config.status_write_timeout, + ), ..default_sub_config }; // Outbound auth provider was built above (static bearer when @@ -810,7 +818,30 @@ where config.base_url.clone(), outbound_auth_provider, ); - info!("Subscriptions engine ENABLED"); + // Server-driven status transitions are written back into the stored + // `Subscription` (issue #357), so the engine's own decisions survive + // a restart and `GET /Subscription/{id}` stops contradicting + // `$status`. + // + // This is the same `storage_arc` the whole app is served from — on a + // composite deployment that is the *composite*, not the primary, so + // the rewritten resource also reaches the Elasticsearch index and + // `GET /Subscription?status=active` answers from current documents. + let engine = + engine + .with_status_store(Arc::clone(&storage_arc) + as Arc); + // Reported from the engine rather than the config flag, so the line + // reflects what is actually wired: it reads OFF if either the flag is + // false or no store was attached. + if engine.persists_status() { + info!("Subscriptions engine ENABLED (status write-back ON)"); + } else { + info!( + "Subscriptions engine ENABLED (status write-back OFF; \ + transitions will not survive a restart)" + ); + } let engine = Arc::new(engine); spawn_subscription_rehydration(Arc::clone(&engine), Arc::clone(&storage_arc), &config); state.with_subscription_engine(engine) diff --git a/crates/subscriptions/src/config.rs b/crates/subscriptions/src/config.rs index 498ebf794..88965c0e0 100644 --- a/crates/subscriptions/src/config.rs +++ b/crates/subscriptions/src/config.rs @@ -49,6 +49,34 @@ pub struct SubscriptionConfig { /// FHIR Messaging channel settings. `None` disables the messaging dispatcher. pub messaging: Option, + + /// Whether server-driven status transitions are written back into the + /// stored `Subscription` resource (issue #357). + /// + /// `true` (the default) makes the engine's own decisions durable and + /// visible: a subscription it activated reads back `active`, and one its + /// delivery circuit breaker turned `off` stays `off` across a restart + /// instead of resuming delivery to a dead endpoint. + /// + /// Set `HFS_SUBSCRIPTION_PERSIST_STATUS=false` to restore the previous + /// in-memory-only behaviour. That is the kill switch for the case where + /// this write-back is itself the incident: it costs at most three writes + /// per subscription lifetime, but it does write to the primary backend from + /// the delivery path. + pub persist_status: bool, + + /// How long a single status write-back may take before it is abandoned. + /// + /// A hung backend must not pin the spawned dispatch task that is waiting on + /// it. On timeout the transition stays in memory and is logged at `error`. + pub status_write_timeout: Duration, + + /// Maximum status write-backs in flight at once. + /// + /// A subscriber outage can trip every subscription in a tenant at nearly the + /// same instant; without a bound those transitions would each check out a + /// backend connection and could starve ordinary reads. + pub status_write_concurrency: usize, } /// FHIR Messaging channel configuration. @@ -82,6 +110,9 @@ impl Default for SubscriptionConfig { ws_token_lifetime_secs: 30, smtp: None, messaging: None, + persist_status: true, + status_write_timeout: Duration::from_secs(5), + status_write_concurrency: 4, } } } diff --git a/crates/subscriptions/src/engine/mod.rs b/crates/subscriptions/src/engine/mod.rs index a035f6623..68533f06a 100644 --- a/crates/subscriptions/src/engine/mod.rs +++ b/crates/subscriptions/src/engine/mod.rs @@ -9,6 +9,10 @@ pub mod retry; use std::sync::Arc; use dashmap::DashMap; +use helios_persistence::core::ResourceStorage; +use helios_persistence::error::StorageError; +use helios_persistence::tenant::{TenantContext, TenantId, TenantPermissions}; +use tokio::sync::Semaphore; use tracing::{debug, error, info, warn}; use crate::channels::email::EmailChannel; @@ -45,6 +49,15 @@ pub struct SubscriptionEngine { messaging_channel: Option>, config: SubscriptionConfig, base_url: String, + /// Storage handle used to write a server-driven status transition back into + /// the stored `Subscription` resource. `None` leaves every transition + /// in-memory only, which is the pre-#357 behaviour. + status_store: Option>, + /// Bounds how many status write-backs may be in flight at once, so a + /// transition storm (every subscription tripping at the same instant during + /// a subscriber outage) cannot drain the primary backend's connection pool + /// and take reads down with it. + status_write_slots: Arc, } fn calculate_handshake_retry_delay( @@ -103,6 +116,8 @@ impl SubscriptionEngine { ) }); + let status_write_slots = Arc::new(Semaphore::new(config.status_write_concurrency.max(1))); + Self { topic_registry, topic_resource_index: DashMap::new(), @@ -116,9 +131,38 @@ impl SubscriptionEngine { messaging_channel, config, base_url, + status_store: None, + status_write_slots, } } + /// Attaches the storage handle that server-driven status transitions are + /// written back to (issue #357). + /// + /// Without this the engine's decisions — `requested` → `active` after a + /// successful handshake, `→ error`, `→ off` once the delivery circuit + /// breaker trips — live only in the in-memory manager. They are then lost on + /// restart, and `GET /Subscription/{id}` contradicts + /// `GET /Subscription/{id}/$status` for the lifetime of the resource. + /// + /// This deliberately takes the plain [`ResourceStorage`] every backend + /// already implements rather than a new per-backend store, so the fix + /// applies uniformly to SQLite, PostgreSQL, MongoDB, S3, and the composite + /// deployments with no schema change and no migration. + /// + /// On a composite deployment pass the **composite**, not the primary, so the + /// rewritten resource also reaches the search index — otherwise + /// `GET /Subscription?status=active` keeps answering from stale documents. + pub fn with_status_store(mut self, storage: Arc) -> Self { + self.status_store = Some(storage); + self + } + + /// Whether server-driven status transitions are written back to storage. + pub fn persists_status(&self) -> bool { + self.status_store.is_some() && self.config.persist_status + } + /// Returns a reference to the topic registry. pub fn topic_registry(&self) -> &Arc { &self.topic_registry @@ -544,9 +588,8 @@ impl SubscriptionEngine { error = %e, "Failed to build handshake" ); - let _ = - self.manager - .update_status(tenant_id, sub_id, SubscriptionStatusCode::Error); + self.transition_status(tenant_id, sub_id, SubscriptionStatusCode::Error) + .await; return; } }; @@ -596,11 +639,8 @@ impl SubscriptionEngine { endpoint = subscription.channel.endpoint.as_deref().unwrap_or(""), "Email channel requested but no SMTP settings configured" ); - let _ = self.manager.update_status( - tenant_id, - sub_id, - SubscriptionStatusCode::Error, - ); + self.transition_status(tenant_id, sub_id, SubscriptionStatusCode::Error) + .await; return; } }, @@ -613,11 +653,8 @@ impl SubscriptionEngine { endpoint = subscription.channel.endpoint.as_deref().unwrap_or(""), "Messaging channel requested but messaging settings not configured" ); - let _ = self.manager.update_status( - tenant_id, - sub_id, - SubscriptionStatusCode::Error, - ); + self.transition_status(tenant_id, sub_id, SubscriptionStatusCode::Error) + .await; return; } }, @@ -643,11 +680,8 @@ impl SubscriptionEngine { attempt, "Handshake successful, activating subscription" ); - let _ = self.manager.update_status( - tenant_id, - sub_id, - SubscriptionStatusCode::Active, - ); + self.transition_status(tenant_id, sub_id, SubscriptionStatusCode::Active) + .await; return; } Ok(DispatchResult::PermanentError(msg)) => { @@ -660,11 +694,8 @@ impl SubscriptionEngine { error = %msg, "Handshake failed with permanent error" ); - let _ = self.manager.update_status( - tenant_id, - sub_id, - SubscriptionStatusCode::Error, - ); + self.transition_status(tenant_id, sub_id, SubscriptionStatusCode::Error) + .await; return; } Ok(DispatchResult::RetryableError(msg)) => { @@ -678,11 +709,8 @@ impl SubscriptionEngine { error = %msg, "Handshake retries exhausted" ); - let _ = self.manager.update_status( - tenant_id, - sub_id, - SubscriptionStatusCode::Error, - ); + self.transition_status(tenant_id, sub_id, SubscriptionStatusCode::Error) + .await; return; } @@ -711,11 +739,8 @@ impl SubscriptionEngine { error = %e, "Handshake error" ); - let _ = self.manager.update_status( - tenant_id, - sub_id, - SubscriptionStatusCode::Error, - ); + self.transition_status(tenant_id, sub_id, SubscriptionStatusCode::Error) + .await; return; } } @@ -802,7 +827,7 @@ impl SubscriptionEngine { error = %msg, "Permanent delivery error" ); - self.handle_delivery_failure(tenant_id, sub_id); + self.handle_delivery_failure(tenant_id, sub_id).await; return; } Ok(DispatchResult::RetryableError(msg)) => { @@ -819,7 +844,7 @@ impl SubscriptionEngine { error = %msg, "Max retries exhausted" ); - self.handle_delivery_failure(tenant_id, sub_id); + self.handle_delivery_failure(tenant_id, sub_id).await; return; } @@ -846,39 +871,225 @@ impl SubscriptionEngine { error = %e, "Dispatch error" ); - self.handle_delivery_failure(tenant_id, sub_id); + self.handle_delivery_failure(tenant_id, sub_id).await; return; } } } } + /// The single choke point for every server-driven status transition. + /// + /// Applies the transition to the in-memory manager and, only when the + /// manager accepted it as a genuine state change, writes the new status back + /// into the stored `Subscription` resource. + /// + /// Gating the write on the manager's `Ok` is what keeps the write volume + /// bounded. [`SubscriptionManager::update_status`] rejects same-state + /// transitions and transitions out of a terminal state, so when a subscriber + /// outage parks hundreds of concurrent dispatch tasks that all observe + /// `consecutive_failures >= off_threshold`, exactly one of them gets `Ok` + /// and therefore exactly one issues a write. Over a subscription's whole + /// lifetime that is at most three writes (`→ active`, `→ error`, `→ off`). + /// + /// Returns whether the in-memory transition was applied. Persistence failure + /// does **not** change that answer: see [`Self::write_status_back`]. + async fn transition_status( + &self, + tenant_id: &str, + subscription_id: &str, + new_status: SubscriptionStatusCode, + ) -> bool { + if self + .manager + .update_status(tenant_id, subscription_id, new_status) + .is_err() + { + return false; + } + self.write_status_back(tenant_id, subscription_id, new_status) + .await; + true + } + + /// Writes a server-driven status transition into the stored `Subscription`. + /// + /// Fails **open**, always. The in-memory transition has already been applied + /// by [`Self::transition_status`] and is what this process acts on, so a + /// storage error must never undo it: refusing to mark a subscription `off` + /// because the database is briefly unavailable would put the server straight + /// back to hammering the endpoint the circuit breaker just gave up on. The + /// cost of failing open is that this one transition does not survive a + /// restart — exactly the pre-#357 behaviour, and no worse. + /// + /// A concurrent client write wins. `ResourceStorage::update` takes the + /// version that was read and every backend rejects a stale one, so a racing + /// `PUT /Subscription/{id}` surfaces here as a version conflict, which is + /// **dropped rather than retried**: the client's copy is the newer intent, + /// and its write re-registers the subscription anyway. + async fn write_status_back( + &self, + tenant_id: &str, + subscription_id: &str, + new_status: SubscriptionStatusCode, + ) { + if !self.config.persist_status { + return; + } + let Some(storage) = self.status_store.as_ref() else { + return; + }; + + // A transition storm must not drain the backend's connection pool. + // `acquire_owned` is not used: the permit lives only for this call. + let Ok(_permit) = self.status_write_slots.acquire().await else { + return; // Semaphore closed — engine shutting down. + }; + + // A server-authored context: this runs on a background task with no + // request behind it, so there is no caller context to inherit. Same + // construction startup rehydration uses (`crate::rehydrate`). The tenant + // id comes from the subscription's own registration, never from a + // client-supplied header, so this cannot widen tenant reach. + let tenant = TenantContext::new(TenantId::new(tenant_id), TenantPermissions::full_access()); + let timeout = self.config.status_write_timeout; + + let result = tokio::time::timeout( + timeout, + persist_status(storage.as_ref(), &tenant, subscription_id, new_status), + ) + .await; + + match result { + Ok(Ok(PersistOutcome::Written)) => debug!( + tenant_id, + subscription_id, + status = %new_status, + "Persisted subscription status transition" + ), + Ok(Ok(PersistOutcome::AlreadyCurrent)) => debug!( + tenant_id, + subscription_id, + status = %new_status, + "Stored subscription status already current; no write issued" + ), + Ok(Ok(PersistOutcome::Vanished)) => debug!( + tenant_id, + subscription_id, "Subscription resource is gone; nothing to persist status onto" + ), + Ok(Ok(PersistOutcome::RacedClientWrite)) => debug!( + tenant_id, + subscription_id, + status = %new_status, + "Concurrent client write won; dropping server status write-back" + ), + // The one line that tells an operator this server has silently + // reverted to losing its subscription decisions on restart. + Ok(Err(e)) => error!( + tenant_id, + subscription_id, + status = %new_status, + error = %e, + "Failed to persist subscription status transition; \ + this transition will not survive a restart" + ), + Err(_) => error!( + tenant_id, + subscription_id, + status = %new_status, + timeout_ms = timeout.as_millis() as u64, + "Timed out persisting subscription status transition; \ + this transition will not survive a restart" + ), + } + } + /// Handle a delivery failure: increment failure count and potentially /// transition status to error or off. - fn handle_delivery_failure(&self, tenant_id: &str, subscription_id: &str) { - if let Some(failure_count) = self.manager.record_failure(tenant_id, subscription_id) { - if failure_count >= self.config.off_threshold { - warn!( - subscription_id, - failures = failure_count, - "Turning off subscription after repeated failures" - ); - let _ = self.manager.update_status( - tenant_id, - subscription_id, - SubscriptionStatusCode::Off, - ); - } else if failure_count >= self.config.error_threshold { - let _ = self.manager.update_status( - tenant_id, - subscription_id, - SubscriptionStatusCode::Error, - ); - } + async fn handle_delivery_failure(&self, tenant_id: &str, subscription_id: &str) { + let Some(failure_count) = self.manager.record_failure(tenant_id, subscription_id) else { + return; + }; + if failure_count >= self.config.off_threshold { + warn!( + tenant_id, + subscription_id, + failures = failure_count, + threshold = self.config.off_threshold, + "Turning off subscription after repeated failures" + ); + self.transition_status(tenant_id, subscription_id, SubscriptionStatusCode::Off) + .await; + } else if failure_count >= self.config.error_threshold { + self.transition_status(tenant_id, subscription_id, SubscriptionStatusCode::Error) + .await; } } } +/// What [`persist_status`] did, so the caller can log it at the right level. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum PersistOutcome { + /// The stored resource was rewritten with the new status. + Written, + /// The stored resource already carried this status — no write issued. + /// + /// This is the common case on the first boot after upgrading: rehydration + /// re-activates every subscription that reads back `requested`, and without + /// this check each success would rewrite a resource that write-back had + /// already corrected on a previous run. + AlreadyCurrent, + /// The resource no longer exists (deleted between the transition and the + /// write). Not an error. + Vanished, + /// A concurrent client write bumped the version first. Dropped, not retried. + RacedClientWrite, +} + +/// Reads the stored `Subscription`, replaces its `status`, and writes it back. +/// +/// Only the `status` field is touched: the resource is read, mutated in place, +/// and handed back to `update`, so a field this engine knows nothing about +/// cannot be dropped on the way through. In particular the version-shape markers +/// (`topic`/`channelType` vs `criteria`/`channel`) survive untouched, which +/// matters because startup rehydration infers the FHIR version from exactly +/// those keys (`crate::rehydrate::infer_fhir_version`). +async fn persist_status( + storage: &dyn ResourceStorage, + tenant: &TenantContext, + subscription_id: &str, + new_status: SubscriptionStatusCode, +) -> Result { + let current = match storage.read(tenant, "Subscription", subscription_id).await { + Ok(Some(stored)) => stored, + Ok(None) => return Ok(PersistOutcome::Vanished), + // A deleted subscription is `Gone`, not an error worth shouting about. + Err(StorageError::Resource(_)) => return Ok(PersistOutcome::Vanished), + Err(e) => return Err(e), + }; + + if current.content().get("status").and_then(|v| v.as_str()) == Some(new_status.as_fhir_str()) { + return Ok(PersistOutcome::AlreadyCurrent); + } + + let mut updated = current.content().clone(); + let Some(object) = updated.as_object_mut() else { + // Not a JSON object — nothing sane to write back onto. + return Ok(PersistOutcome::Vanished); + }; + object.insert( + "status".to_string(), + serde_json::Value::String(new_status.as_fhir_str().to_string()), + ); + + match storage.update(tenant, ¤t, updated).await { + Ok(_) => Ok(PersistOutcome::Written), + Err(StorageError::Concurrency(_)) => Ok(PersistOutcome::RacedClientWrite), + Err(StorageError::Resource(_)) => Ok(PersistOutcome::Vanished), + Err(e) => Err(e), + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/subscriptions/src/manager/mod.rs b/crates/subscriptions/src/manager/mod.rs index e7e49893b..c36a9890b 100644 --- a/crates/subscriptions/src/manager/mod.rs +++ b/crates/subscriptions/src/manager/mod.rs @@ -229,6 +229,33 @@ impl SubscriptionManager { .and_then(SubscriptionStatusCode::from_fhir_str) .unwrap_or(SubscriptionStatusCode::Requested); + // Carry the runtime counters across a re-registration. + // + // `register` runs on *every* write to the resource (create, update, + // patch — see `SubscriptionEngine::handle_subscription_event`), and it + // used to rebuild the entry with both counters at zero. That made an + // unrelated edit — adding a filter, rotating an endpoint header — a + // silent reset of the delivery circuit breaker, re-arming the server + // against an endpoint it had already given up on, with no restart + // involved. It also rewound `eventsSinceSubscriptionStart`, which + // subscribers use to detect missed notifications. + // + // The counters are therefore preserved unless the client is *re-arming* + // the subscription (see [`is_rearm`]), which is the one deliberate, + // client-driven reset and the documented way back from `error`/`off`. + let key = (tenant_id.to_string(), subscription_id.to_string()); + let previous = self.subscriptions.get(&key).map(|entry| { + let sub = entry.value(); + (sub.status, sub.events_since_start, sub.consecutive_failures) + }); + + let (events_since_start, consecutive_failures) = match previous { + Some((previous_status, events, failures)) if !is_rearm(previous_status, status) => { + (events, failures) + } + _ => (0, 0), + }; + let subscription = ActiveSubscription { id: subscription_id.to_string(), topic_url, @@ -236,8 +263,8 @@ impl SubscriptionManager { channel, filters: parsed_filters, fhir_version, - events_since_start: 0, - consecutive_failures: 0, + events_since_start, + consecutive_failures, tenant_id: tenant_id.to_string(), }; @@ -245,10 +272,13 @@ impl SubscriptionManager { tenant = tenant_id, subscription_id, topic = %subscription.topic_url, + status = %subscription.status, + events_since_start, + consecutive_failures, + re_registered = previous.is_some(), "Registered subscription" ); - let key = (tenant_id.to_string(), subscription_id.to_string()); self.subscriptions.insert(key, subscription.clone()); Ok(subscription) @@ -367,6 +397,37 @@ impl SubscriptionManager { } } +/// Whether re-registering with `incoming` status re-arms a subscription that +/// was previously in `previous` status, and so should reset its runtime +/// counters. +/// +/// Two cases count as re-arming, and nothing else does: +/// +/// 1. The client wrote `requested` — the spec's "please (re-)activate this" +/// signal, whatever state the subscription was in. +/// 2. The client moved it out of a non-serving state (`error`, `off`, +/// `entered-in-error`) under its own steam, e.g. straight to `active`. +/// +/// An ordinary edit that leaves the status alone — or any server-authored +/// status already written back into the resource — is *not* a re-arm, so it +/// keeps the delivery failure streak and the event counter intact. +fn is_rearm(previous: SubscriptionStatusCode, incoming: SubscriptionStatusCode) -> bool { + if incoming == SubscriptionStatusCode::Requested { + return true; + } + matches!( + previous, + SubscriptionStatusCode::Error + | SubscriptionStatusCode::Off + | SubscriptionStatusCode::EnteredInError + ) && !matches!( + incoming, + SubscriptionStatusCode::Error + | SubscriptionStatusCode::Off + | SubscriptionStatusCode::EnteredInError + ) +} + // --- Version-aware field extraction --- /// Extract the topic canonical URL from a Subscription resource. @@ -1286,4 +1347,193 @@ pub(crate) mod tests { Err(SubscriptionError::InvalidFilter { .. }) )); } + + // ── Runtime counters across re-registration (issue #357) ────────────── + // + // `register` runs on every write to the Subscription resource. It used to + // rebuild the entry with both counters at zero, which made an ordinary edit + // a silent reset of the delivery circuit breaker. + + /// Builds the default test subscription with `status` replaced. + fn subscription_with_status(status: &str) -> serde_json::Value { + let mut resource = default_subscription_json(); + resource["status"] = json!(status); + resource + } + + /// Drives a subscription to `status` with the given counters, the way the + /// engine does: register, then transition and record activity. + fn manager_with_state( + status: SubscriptionStatusCode, + events: u64, + failures: u32, + ) -> SubscriptionManager { + let manager = + SubscriptionManager::new(create_test_registry(), vec!["rest-hook".to_string()]); + manager + .register( + "t1", + "sub-1", + &subscription_with_status("requested"), + FhirVersion::default(), + ) + .expect("initial register"); + if status != SubscriptionStatusCode::Requested { + manager + .update_status("t1", "sub-1", status) + .expect("drive to status"); + } + for _ in 0..events { + manager.increment_event_count("t1", "sub-1"); + } + for _ in 0..failures { + manager.record_failure("t1", "sub-1"); + } + manager + } + + /// An edit that does not change the status must not reset the counters. + /// This is the client-PUT circuit-breaker reset: before #357, adding a + /// filter or rotating a header re-armed the server against an endpoint it + /// had already given up on, with no restart involved. + #[test] + fn test_reregistration_preserves_counters() { + let manager = manager_with_state(SubscriptionStatusCode::Active, 7, 2); + + let re = manager + .register( + "t1", + "sub-1", + &subscription_with_status("active"), + FhirVersion::default(), + ) + .expect("re-register"); + + assert_eq!(re.events_since_start, 7, "event counter must not rewind"); + assert_eq!( + re.consecutive_failures, 2, + "failure streak must survive an ordinary edit" + ); + assert_eq!(re.status, SubscriptionStatusCode::Active); + } + + /// Writing `requested` is the spec's "please (re-)activate this" signal and + /// is the documented way back from `error`/`off`. It resets the counters. + #[test] + fn test_client_rearm_with_requested_resets_counters() { + let manager = manager_with_state(SubscriptionStatusCode::Error, 9, 5); + + let re = manager + .register( + "t1", + "sub-1", + &subscription_with_status("requested"), + FhirVersion::default(), + ) + .expect("re-register"); + + assert_eq!(re.status, SubscriptionStatusCode::Requested); + assert_eq!(re.events_since_start, 0); + assert_eq!(re.consecutive_failures, 0); + } + + /// Moving a subscription out of `off` under the client's own steam is also + /// a re-arm: keeping the failure streak would re-trip the breaker on the + /// very next failure. + #[test] + fn test_client_rearm_out_of_off_resets_counters() { + let manager = manager_with_state(SubscriptionStatusCode::Off, 4, 10); + + let re = manager + .register( + "t1", + "sub-1", + &subscription_with_status("active"), + FhirVersion::default(), + ) + .expect("re-register"); + + assert_eq!(re.status, SubscriptionStatusCode::Active); + assert_eq!(re.consecutive_failures, 0); + } + + /// A server-authored status written back into the resource (#357) is NOT a + /// re-arm — otherwise the write-back would erase the very counters it is + /// meant to make durable the next time the resource is registered. + #[test] + fn test_written_back_status_is_not_a_rearm() { + let manager = manager_with_state(SubscriptionStatusCode::Error, 3, 4); + + // The resource now carries `error`, because write-back put it there. + let re = manager + .register( + "t1", + "sub-1", + &subscription_with_status("error"), + FhirVersion::default(), + ) + .expect("re-register"); + + assert_eq!(re.status, SubscriptionStatusCode::Error); + assert_eq!(re.events_since_start, 3); + assert_eq!(re.consecutive_failures, 4); + } + + /// A first registration always starts from zero — there is nothing to carry. + #[test] + fn test_first_registration_starts_at_zero() { + let manager = + SubscriptionManager::new(create_test_registry(), vec!["rest-hook".to_string()]); + let sub = manager + .register( + "t1", + "sub-1", + &subscription_with_status("active"), + FhirVersion::default(), + ) + .expect("register"); + assert_eq!(sub.events_since_start, 0); + assert_eq!(sub.consecutive_failures, 0); + } + + /// Counters are per `(tenant, subscription)`: re-registering one tenant's + /// subscription must not read another tenant's state. + #[test] + fn test_counters_are_tenant_scoped() { + let manager = manager_with_state(SubscriptionStatusCode::Active, 6, 3); + + let other = manager + .register( + "t2", + "sub-1", + &subscription_with_status("active"), + FhirVersion::default(), + ) + .expect("register for other tenant"); + + assert_eq!(other.events_since_start, 0); + assert_eq!(other.consecutive_failures, 0); + // The first tenant is untouched. + let first = manager.get_subscription("t1", "sub-1").unwrap(); + assert_eq!(first.events_since_start, 6); + } + + /// `entered-in-error` used to be unparseable, so `register` fell back to + /// `requested` and rehydration would handshake and activate a subscription + /// the client had explicitly retracted. + #[test] + fn test_entered_in_error_is_registered_as_itself() { + let manager = + SubscriptionManager::new(create_test_registry(), vec!["rest-hook".to_string()]); + let sub = manager + .register( + "t1", + "sub-1", + &subscription_with_status("entered-in-error"), + FhirVersion::default(), + ) + .expect("register"); + assert_eq!(sub.status, SubscriptionStatusCode::EnteredInError); + assert!(sub.status.is_terminal()); + } } diff --git a/crates/subscriptions/src/manager/status.rs b/crates/subscriptions/src/manager/status.rs index 03de845fd..439486838 100644 --- a/crates/subscriptions/src/manager/status.rs +++ b/crates/subscriptions/src/manager/status.rs @@ -14,10 +14,25 @@ pub enum SubscriptionStatusCode { Error, /// The subscription has been turned off by the server or client. Off, + /// The subscription was created in error and must never be acted on. + /// + /// Client-owned and terminal: the server never transitions *to* it and + /// never transitions *out* of it. Present in the R4B/R5/R6 value set (and + /// absent from R4's), but parsed for every version — an unknown code + /// previously fell back to [`Requested`](Self::Requested), which made the + /// engine handshake and *activate* a resource the client had explicitly + /// retracted. + EnteredInError, } impl SubscriptionStatusCode { /// Returns whether a transition from `self` to `target` is valid. + /// + /// Note that both [`Off`](Self::Off) and + /// [`EnteredInError`](Self::EnteredInError) are terminal: no transition + /// leaves them, so a subscription only returns to service when the client + /// re-arms it by rewriting the resource (see + /// [`SubscriptionManager::register`](crate::manager::SubscriptionManager::register)). pub fn can_transition_to(self, target: Self) -> bool { matches!( (self, target), @@ -34,6 +49,14 @@ impl SubscriptionStatusCode { ) } + /// Whether the server should stop acting on a subscription in this state. + /// + /// Terminal states are still *registered* (so `$status` keeps answering for + /// them) but never dispatched to and never handshaken. + pub fn is_terminal(self) -> bool { + matches!(self, Self::Off | Self::EnteredInError) + } + /// Parses a status string from a FHIR Subscription resource. pub fn from_fhir_str(s: &str) -> Option { match s { @@ -41,6 +64,7 @@ impl SubscriptionStatusCode { "active" => Some(Self::Active), "error" => Some(Self::Error), "off" => Some(Self::Off), + "entered-in-error" => Some(Self::EnteredInError), _ => None, } } @@ -52,6 +76,7 @@ impl SubscriptionStatusCode { Self::Active => "active", Self::Error => "error", Self::Off => "off", + Self::EnteredInError => "entered-in-error", } } } @@ -103,6 +128,45 @@ mod tests { assert!(!Active.can_transition_to(Active)); } + /// `entered-in-error` is client-owned and inert: the server neither enters + /// nor leaves it. Before it was parsed, `from_fhir_str` returned `None` and + /// `register` fell back to `Requested`, so rehydration would handshake and + /// *activate* a subscription the client had retracted. + #[test] + fn test_entered_in_error_is_inert() { + use SubscriptionStatusCode::*; + + assert_eq!( + SubscriptionStatusCode::from_fhir_str("entered-in-error"), + Some(EnteredInError) + ); + assert_eq!(EnteredInError.as_fhir_str(), "entered-in-error"); + + for target in [Requested, Active, Error, Off, EnteredInError] { + assert!( + !EnteredInError.can_transition_to(target), + "entered-in-error must not transition to {target}" + ); + assert!( + !target.can_transition_to(EnteredInError), + "{target} must not transition to entered-in-error" + ); + } + } + + #[test] + fn test_is_terminal() { + use SubscriptionStatusCode::*; + + assert!(Off.is_terminal()); + assert!(EnteredInError.is_terminal()); + assert!(!Requested.is_terminal()); + assert!(!Active.is_terminal()); + // `error` is NOT terminal: a restart re-runs activation for it, which is + // the only recovery path the server has today. + assert!(!Error.is_terminal()); + } + #[test] fn test_fhir_str_roundtrip() { for status in [ @@ -110,6 +174,7 @@ mod tests { SubscriptionStatusCode::Active, SubscriptionStatusCode::Error, SubscriptionStatusCode::Off, + SubscriptionStatusCode::EnteredInError, ] { let s = status.as_fhir_str(); let parsed = SubscriptionStatusCode::from_fhir_str(s).unwrap(); diff --git a/crates/subscriptions/src/rehydrate.rs b/crates/subscriptions/src/rehydrate.rs index 248cfecf8..781a840bb 100644 --- a/crates/subscriptions/src/rehydrate.rs +++ b/crates/subscriptions/src/rehydrate.rs @@ -42,24 +42,39 @@ //! //! | Stored status | Action | Why | //! |---|---|---| -//! | `off` | skipped entirely | Terminal: `can_transition_to` permits no target, so a registered `off` subscription could never legally become useful. | -//! | `active` | registered, no handshake | The handshake completed in a previous process lifetime; re-sending it is noise. | -//! | `error` | registered, no handshake | Never matches (only `Active` does), but registering keeps `$status` honest instead of reporting an unknown subscription. | //! | `requested` | registered **and activated** | The spec state for "server has not yet activated"; runs the same activation path the write handler runs. | +//! | `active` | registered, no handshake | The handshake completed in a previous process lifetime; re-sending it is noise. | +//! | `error` | registered **and activated** | Nothing in the engine performs `error` → `active` while a subscription is dormant, so re-running activation here is its only recovery path. | +//! | `off` | registered, dormant | Terminal. Registered so `$status` keeps answering; never handshaken, never matched. | +//! | `entered-in-error` | registered, dormant | Client-retracted. Same treatment as `off`. | +//! +//! Two of those rows changed with #357, and both changes exist to stop a +//! *persisted* status from being worse than the volatile one it replaced. +//! +//! `off` used to be skipped outright on the reasoning that a terminal status +//! could never become useful again. That was defensible only while the status +//! was volatile. Now that the delivery circuit breaker's decision survives a +//! restart, skipping would mean an `off` subscription is absent from the engine +//! forever — and `$status` answers `404` for a resource that reads back `200` +//! from `GET /Subscription/{id}`. Registering it dormant costs one map entry and +//! keeps the two endpoints telling the same story. It cannot leak delivery: +//! [`crate::manager::SubscriptionManager::active_subscriptions_for_topic`] +//! selects only `active`. //! -//! Activating `requested` subscriptions is not an optional nicety. Subscription -//! status transitions are held **in memory only** — -//! [`crate::manager::SubscriptionManager::update_status`] never writes back to -//! storage — so a subscription that a client POSTed as `requested` and that -//! handshook successfully still reads `"requested"` from the database forever. -//! Rehydrating only `active` resources would therefore miss essentially every -//! subscription the server itself activated, and would not fix the reported -//! symptom at all. +//! `error` is now re-activated rather than left dormant, for the mirror reason. +//! A dormant `error` subscription is never dispatched to, so it never succeeds, +//! so nothing ever resets it — before #357 the *loss* of the status on restart +//! was what quietly recovered it. Persisting `error` without re-running +//! activation would have converted a transient subscriber outage into permanent +//! silent death. Re-running it preserves the recovery a restart always gave, +//! now explicitly rather than by accident. //! //! The cost is that a restart re-handshakes those subscriptions. That is bounded //! by [`RehydrationConfig::max_concurrent_handshakes`], and can be turned off //! with [`RehydrationConfig::handshake_requested`] by operators who would rather -//! bring subscriptions back with no outbound traffic. +//! bring subscriptions back with no outbound traffic. The population is much +//! smaller than it used to be: with status write-back enabled, a subscription +//! the server activated now reads back `active` and is not handshaken again. use std::collections::BTreeSet; use std::sync::Arc; @@ -104,13 +119,16 @@ pub struct RehydrationConfig { /// batch is held at a time regardless of how many resources exist. pub batch_size: u32, - /// Whether stored `requested` subscriptions are activated (handshaked) - /// during rehydration. + /// Whether stored `requested` and `error` subscriptions are activated + /// (handshaked) during rehydration. /// - /// Defaults to `true` because status transitions are not persisted, so most - /// live subscriptions read back as `requested` — see the module docs. Set to - /// `false` to rehydrate with no outbound traffic at all, accepting that - /// `requested` subscriptions stay inert until they are re-written. + /// Defaults to `true`. With status write-back enabled (#357) this is no + /// longer the bulk-reactivation path it once was — a subscription the server + /// activated reads back `active` and is skipped — but it remains the only + /// recovery an `error` subscription has, and the only way a subscription + /// that died mid-handshake ever leaves `requested`. Set to `false` to + /// rehydrate with no outbound traffic at all, accepting that those two + /// populations stay inert until the resource is re-written. pub handshake_requested: bool, /// Maximum handshakes in flight at once, across all tenants. @@ -149,8 +167,16 @@ pub struct RehydrationReport { pub subscriptions_registered: usize, /// Subscriptions that failed to register (unknown topic, bad channel, …). pub subscriptions_failed: usize, - /// Subscriptions skipped because their stored status is `off`. - pub subscriptions_skipped: usize, + /// Subscriptions registered in a terminal status (`off`, `entered-in-error`). + /// + /// They are counted in [`subscriptions_registered`](Self::subscriptions_registered) + /// too: they *are* registered, so `$status` answers for them, but they are + /// never handshaken and never matched against an event. + /// + /// Before #357 an `off` subscription was skipped outright and never entered + /// the engine at all, which made `$status` answer `404` for it after every + /// restart — a worse answer than the stale one it replaced. + pub subscriptions_dormant: usize, /// Subscriptions handed to the activation (handshake) path. pub handshakes_started: usize, /// Storage errors encountered while paging. A non-zero value means the pass @@ -173,7 +199,7 @@ impl RehydrationReport { self.topics_failed += other.topics_failed; self.subscriptions_registered += other.subscriptions_registered; self.subscriptions_failed += other.subscriptions_failed; - self.subscriptions_skipped += other.subscriptions_skipped; + self.subscriptions_dormant += other.subscriptions_dormant; self.handshakes_started += other.handshakes_started; self.storage_errors += other.storage_errors; } @@ -355,7 +381,7 @@ impl SubscriptionEngine { tenants = report.tenants, topics = report.topics_registered, subscriptions = report.subscriptions_registered, - skipped = report.subscriptions_skipped, + dormant = report.subscriptions_dormant, failed = report.subscriptions_failed, handshakes = report.handshakes_started, "Subscription engine rehydrated" @@ -575,22 +601,6 @@ impl SubscriptionEngine { continue; }; - // `off` is terminal — `can_transition_to` permits no target from - // it — so registering one could never lead anywhere useful. - let stored_status = resource - .get("status") - .and_then(|v| v.as_str()) - .and_then(SubscriptionStatusCode::from_fhir_str); - if stored_status == Some(SubscriptionStatusCode::Off) { - debug!( - tenant = tenant_id, - subscription_id = id, - "Skipping `off` subscription during rehydration" - ); - report.subscriptions_skipped += 1; - continue; - } - let fhir_version = infer_fhir_version(resource, default_version); match self .manager() @@ -608,11 +618,34 @@ impl SubscriptionEngine { ); report.subscriptions_registered += 1; - // Only `requested` needs the handshake. `active` was - // established in a previous process lifetime; `error` - // stays registered but dormant. - if sub.status == SubscriptionStatusCode::Requested - && config.handshake_requested + // `off` and `entered-in-error` are terminal: registered + // so `$status` keeps answering for them, never + // handshaken, and never matched (the evaluator only + // selects `active`). Counted separately because a + // deployment wants to see how many of its subscriptions + // are parked rather than serving. + if sub.status.is_terminal() { + debug!( + tenant = tenant_id, + subscription_id = %sub.id, + status = %sub.status, + "Registered terminal subscription as dormant" + ); + report.subscriptions_dormant += 1; + continue; + } + + // `requested` has never been activated. `error` is a + // subscription the delivery circuit breaker gave up on + // — and, because nothing in the engine performs an + // `error` → `active` transition while it is dormant, a + // restart is the *only* recovery it has. Re-running + // activation for both is what stops a persisted `error` + // from becoming a permanent one. + if matches!( + sub.status, + SubscriptionStatusCode::Requested | SubscriptionStatusCode::Error + ) && config.handshake_requested { pending.push(sub); } diff --git a/crates/subscriptions/tests/rehydrate_integration.rs b/crates/subscriptions/tests/rehydrate_integration.rs index b5c6045c7..1b6b6c248 100644 --- a/crates/subscriptions/tests/rehydrate_integration.rs +++ b/crates/subscriptions/tests/rehydrate_integration.rs @@ -285,10 +285,18 @@ async fn topic_is_registered_before_subscriptions() { ); } -/// `off` is a terminal status — it can transition nowhere — so rehydrating one -/// would register a subscription that could never legally become useful. +/// `off` is terminal — it can transition nowhere and is never dispatched to — +/// but it is still **registered**, dormant (issue #357). +/// +/// It used to be skipped outright. That was defensible only while the status +/// was volatile: once the delivery circuit breaker's decision survives a +/// restart, skipping means the subscription is absent from the engine forever, +/// so `$status` answers `404` for a resource that `GET /Subscription/{id}` +/// answers `200` for. Registering it dormant keeps the two endpoints telling the +/// same story, and cannot leak delivery because the evaluator selects only +/// `active`. #[tokio::test] -async fn off_subscriptions_are_skipped() { +async fn off_subscriptions_are_registered_dormant() { let backend = storage(); seed(&backend, TENANT_ID, topic_resource_type(), topic_resource()).await; seed( @@ -309,14 +317,102 @@ async fn off_subscriptions_are_skipped() { ) .await; - assert_eq!(report.subscriptions_skipped, 1, "report: {report:?}"); - assert_eq!(report.subscriptions_registered, 0, "report: {report:?}"); + assert_eq!(report.subscriptions_dormant, 1, "report: {report:?}"); + assert_eq!(report.subscriptions_registered, 1, "report: {report:?}"); + assert_eq!( + report.handshakes_started, 0, + "a terminal subscription must never be handshaken: {report:?}" + ); + + let sub = engine + .manager() + .get_subscription(TENANT_ID, "sub-off") + .expect("an `off` subscription must be registered so $status can answer"); + assert_eq!(sub.status, SubscriptionStatusCode::Off); + + // Registered, but inert: it must not be selected for delivery. assert!( engine .manager() - .get_subscription(TENANT_ID, "sub-off") - .is_none(), - "an `off` subscription must not be registered" + .active_subscriptions_for_topic(TENANT_ID, TOPIC_URL) + .is_empty(), + "an `off` subscription must never be dispatched to" + ); +} + +/// `entered-in-error` gets the same dormant treatment as `off`. +/// +/// It previously failed to parse, so `register` fell back to `requested` — and +/// rehydration would then handshake and *activate* a subscription the client had +/// explicitly retracted. +#[tokio::test] +async fn entered_in_error_subscriptions_are_registered_dormant() { + let backend = storage(); + seed(&backend, TENANT_ID, topic_resource_type(), topic_resource()).await; + seed( + &backend, + TENANT_ID, + "Subscription", + subscription_resource("sub-eie", "entered-in-error", "http://127.0.0.1:1/hook"), + ) + .await; + + let engine = fresh_engine(); + let report = engine + .rehydrate( + &backend, + TENANT_ID, + current_fhir_version(), + &RehydrationConfig::default(), // handshakes ENABLED — must still not fire + ) + .await; + + assert_eq!(report.subscriptions_dormant, 1, "report: {report:?}"); + assert_eq!( + report.handshakes_started, 0, + "a retracted subscription must never be activated: {report:?}" + ); + let sub = engine + .manager() + .get_subscription(TENANT_ID, "sub-eie") + .expect("registered so $status can answer"); + assert_eq!(sub.status, SubscriptionStatusCode::EnteredInError); +} + +/// An `error` subscription is re-activated on restart, not left dormant. +/// +/// Nothing in the engine performs `error` → `active` while a subscription is +/// dormant (the evaluator only selects `active`, so it is never dispatched to, +/// so it never succeeds, so nothing resets it). Before #357 the *loss* of the +/// status on restart is what quietly recovered it. Now that `error` persists, +/// re-running activation here is what stops a transient subscriber outage from +/// becoming permanent silent death. +#[tokio::test] +async fn error_subscriptions_are_reactivated_on_restart() { + let backend = storage(); + seed(&backend, TENANT_ID, topic_resource_type(), topic_resource()).await; + seed( + &backend, + TENANT_ID, + "Subscription", + subscription_resource("sub-err", "error", "http://127.0.0.1:1/hook"), + ) + .await; + + let engine = fresh_engine(); + let report = engine + .rehydrate( + &backend, + TENANT_ID, + current_fhir_version(), + &RehydrationConfig::default(), + ) + .await; + + assert_eq!(report.subscriptions_registered, 1, "report: {report:?}"); + assert_eq!( + report.handshakes_started, 1, + "`error` must be handed to the activation path: {report:?}" ); } diff --git a/crates/subscriptions/tests/status_persistence.rs b/crates/subscriptions/tests/status_persistence.rs new file mode 100644 index 000000000..4033fe70a --- /dev/null +++ b/crates/subscriptions/tests/status_persistence.rs @@ -0,0 +1,648 @@ +//! Server-driven subscription status transitions are written back into the +//! stored `Subscription` resource (issue #357). +//! +//! Before this, every runtime decision the engine made — `requested` → `active` +//! after a successful handshake, `→ error`, `→ off` once the delivery circuit +//! breaker tripped — lived only in the engine's in-memory `DashMap`. Two +//! consequences these tests pin down: +//! +//! 1. `GET /Subscription/{id}` (storage) contradicted +//! `GET /Subscription/{id}/$status` (engine) for the life of the resource. +//! 2. The circuit breaker was amnesiac: a subscription the server turned `off` +//! after repeated delivery failures reverted to its stored status on restart +//! and resumed hammering the dead endpoint. +//! +//! Each test drives the engine through its **public** entry point +//! (`on_resource_event`), exactly as the REST write handlers do, and then reads +//! the resource back out of a real storage backend. Asserting the stored +//! resource — not the in-memory entry — is what makes these regression tests: +//! the in-memory half already worked. + +use helios_fhir::FhirVersion; +use helios_persistence::backends::sqlite::SqliteBackend; +use helios_persistence::core::ResourceStorage; +use helios_persistence::tenant::{TenantContext, TenantId, TenantPermissions}; +use helios_subscriptions::manager::SubscriptionStatusCode; +use helios_subscriptions::{ + ResourceEvent, ResourceEventType, SubscriptionConfig, SubscriptionEngine, +}; +use serde_json::{Value, json}; +use std::sync::Arc; +use std::time::Duration; +use wiremock::matchers::method; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +const TENANT_ID: &str = "tenant-a"; +const TOPIC_URL: &str = "http://example.org/topic/encounter-start"; + +fn current_fhir_version() -> FhirVersion { + FhirVersion::default() +} + +fn uses_backport_ig() -> bool { + current_fhir_version() == FhirVersion::R4 +} + +fn tenant_context(tenant_id: &str) -> TenantContext { + TenantContext::new(TenantId::new(tenant_id), TenantPermissions::full_access()) +} + +fn storage() -> Arc { + let backend = SqliteBackend::in_memory().expect("in-memory sqlite backend"); + backend.init_schema().expect("init schema"); + Arc::new(backend) +} + +fn topic_resource() -> Value { + if uses_backport_ig() { + json!({ + "resourceType": "Basic", + "id": "topic-1", + "code": { + "coding": [{ + "system": "http://hl7.org/fhir/fhir-types", + "code": "SubscriptionTopic" + }] + }, + "extension": [{ + "url": "http://hl7.org/fhir/5.0/StructureDefinition/extension-SubscriptionTopic.url", + "valueUri": TOPIC_URL + }, { + "url": "http://hl7.org/fhir/4.3/StructureDefinition/extension-SubscriptionTopic.resourceTrigger", + "extension": [{ + "url": "resource", + "valueUri": "http://hl7.org/fhir/StructureDefinition/Encounter" + }, { + "url": "supportedInteraction", + "valueCode": "create" + }] + }] + }) + } else { + json!({ + "resourceType": "SubscriptionTopic", + "id": "topic-1", + "url": TOPIC_URL, + "status": "active", + "resourceTrigger": [{ + "resource": "Encounter", + "supportedInteraction": ["create"] + }] + }) + } +} + +fn topic_resource_type() -> &'static str { + if uses_backport_ig() { + "Basic" + } else { + "SubscriptionTopic" + } +} + +fn subscription_resource(id: &str, status: &str, endpoint: &str) -> Value { + if uses_backport_ig() { + json!({ + "resourceType": "Subscription", + "id": id, + "status": status, + "criteria": TOPIC_URL, + "channel": { + "type": "rest-hook", + "endpoint": endpoint, + "payload": "application/fhir+json" + }, + "extension": [{ + "url": "http://hl7.org/fhir/uv/subscriptions-backport/StructureDefinition/backport-topic-canonical", + "valueCanonical": TOPIC_URL + }, { + "url": "http://hl7.org/fhir/uv/subscriptions-backport/StructureDefinition/backport-payload-content", + "valueCode": "id-only" + }] + }) + } else { + json!({ + "resourceType": "Subscription", + "id": id, + "status": status, + "topic": TOPIC_URL, + "channelType": { "code": "rest-hook" }, + "endpoint": endpoint, + "content": "id-only" + }) + } +} + +/// An engine wired to `storage` for status write-back, with retries and delays +/// collapsed so a single failed dispatch trips the breaker inside a test. +/// +/// `error_threshold` / `off_threshold` are explicit because +/// `handle_delivery_failure` checks `off` **first**: setting both to 1 makes the +/// very first failure go straight to `off`, which is how the `off` case is +/// reached without needing a second delivery (an `error` subscription is no +/// longer matched, so there is no second delivery to be had). +fn engine_with_thresholds( + storage: Arc, + error_threshold: u32, + off_threshold: u32, +) -> SubscriptionEngine { + SubscriptionEngine::new( + SubscriptionConfig { + // No delivery retries: one failed dispatch == one recorded failure. + max_retries: 0, + retry_initial_delay: Duration::from_millis(1), + retry_max_delay: Duration::from_millis(1), + handshake_max_attempts: 1, + handshake_retry_initial_delay: Duration::from_millis(1), + handshake_retry_max_delay: Duration::from_millis(1), + error_threshold, + off_threshold, + ..SubscriptionConfig::default() + }, + "http://localhost:8080".to_string(), + ) + .with_status_store(storage as Arc) +} + +/// The common case: thresholds far enough away that only the handshake matters. +fn engine_with_writeback(storage: Arc) -> SubscriptionEngine { + engine_with_thresholds(storage, 3, 10) +} + +/// A subscriber that accepts the activation handshake and then fails every +/// notification — the shape that trips the delivery circuit breaker. +async fn handshake_ok_then_failing_subscriber() -> MockServer { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .up_to_n_times(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(500)) + .mount(&server) + .await; + server +} + +async fn seed(backend: &SqliteBackend, resource_type: &str, resource: Value) { + backend + .create( + &tenant_context(TENANT_ID), + resource_type, + resource, + current_fhir_version(), + ) + .await + .expect("seed resource"); +} + +/// Registers the topic in the engine's in-memory registry. +/// +/// `SubscriptionManager::register` rejects a subscription whose topic is not +/// already registered (`TopicNotFound`), and the registry is populated only by +/// resource events or by `rehydrate`. Writing the topic to storage is therefore +/// not enough — these tests drive the engine the way the REST write handlers do, +/// so they must fire the topic's own event first. +async fn register_topic(engine: &SubscriptionEngine) { + engine + .on_resource_event(ResourceEvent { + tenant_id: TenantId::new(TENANT_ID), + fhir_version: current_fhir_version(), + resource_type: topic_resource_type().to_string(), + resource_id: "topic-1".to_string(), + version_id: "1".to_string(), + event_type: ResourceEventType::Create, + resource: Some(topic_resource()), + previous_resource: None, + timestamp: chrono::Utc::now(), + }) + .await; + assert!( + engine.topic_registry().get_topic(TOPIC_URL).is_some(), + "test setup: the topic must be registered before any subscription is" + ); +} + +/// Asserts the subscription actually made it into the engine. +/// +/// Without this, a setup failure (a `TopicNotFound` from an unregistered topic, +/// say) leaves the stored status at `requested` — which is exactly what several +/// of these tests assert as their *negative* case, so they would pass +/// vacuously. +fn assert_registered(engine: &SubscriptionEngine, id: &str) -> SubscriptionStatusCode { + engine + .manager() + .get_subscription(TENANT_ID, id) + .unwrap_or_else(|| panic!("test setup: subscription {id} was never registered")) + .status +} + +/// Reads `Subscription.status` straight out of storage — the value a plain +/// `GET /Subscription/{id}` would return. +async fn stored_status(backend: &SqliteBackend, id: &str) -> String { + backend + .read(&tenant_context(TENANT_ID), "Subscription", id) + .await + .expect("read subscription") + .expect("subscription exists") + .content() + .get("status") + .and_then(|v| v.as_str()) + .expect("status present") + .to_string() +} + +async fn stored_version(backend: &SqliteBackend, id: &str) -> String { + backend + .read(&tenant_context(TENANT_ID), "Subscription", id) + .await + .expect("read subscription") + .expect("subscription exists") + .version_id() + .to_string() +} + +fn subscription_written_event(resource: Value, id: &str) -> ResourceEvent { + ResourceEvent { + tenant_id: TenantId::new(TENANT_ID), + fhir_version: current_fhir_version(), + resource_type: "Subscription".to_string(), + resource_id: id.to_string(), + version_id: "1".to_string(), + event_type: ResourceEventType::Create, + resource: Some(resource), + previous_resource: None, + timestamp: chrono::Utc::now(), + } +} + +fn encounter_created_event() -> ResourceEvent { + ResourceEvent { + tenant_id: TenantId::new(TENANT_ID), + fhir_version: current_fhir_version(), + resource_type: "Encounter".to_string(), + resource_id: "enc-1".to_string(), + version_id: "1".to_string(), + event_type: ResourceEventType::Create, + resource: Some(json!({ "resourceType": "Encounter", "id": "enc-1" })), + previous_resource: None, + timestamp: chrono::Utc::now(), + } +} + +/// The headline symptom: a subscription the server activates must read back +/// `active`, not `requested`, from storage. +#[tokio::test] +async fn successful_handshake_persists_active() { + let subscriber = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&subscriber) + .await; + + let backend = storage(); + seed(&backend, topic_resource_type(), topic_resource()).await; + let resource = subscription_resource("sub-1", "requested", &subscriber.uri()); + seed(&backend, "Subscription", resource.clone()).await; + + assert_eq!(stored_status(&backend, "sub-1").await, "requested"); + + let engine = engine_with_writeback(Arc::clone(&backend)); + register_topic(&engine).await; + engine + .on_resource_event(subscription_written_event(resource, "sub-1")) + .await; + + // In-memory and stored now agree — that agreement is the whole issue. + assert_eq!( + engine + .manager() + .get_subscription(TENANT_ID, "sub-1") + .expect("registered") + .status, + SubscriptionStatusCode::Active + ); + assert_eq!( + stored_status(&backend, "sub-1").await, + "active", + "the stored resource must not keep saying `requested` after activation" + ); +} + +/// A delivery failure that trips the `error` threshold must reach storage. +#[tokio::test] +async fn delivery_failure_persists_error() { + let subscriber = handshake_ok_then_failing_subscriber().await; + + let backend = storage(); + seed(&backend, topic_resource_type(), topic_resource()).await; + let resource = subscription_resource("sub-1", "requested", &subscriber.uri()); + seed(&backend, "Subscription", resource.clone()).await; + + // error on the 1st failure; `off` far away so this test isolates `error`. + let engine = engine_with_thresholds(Arc::clone(&backend), 1, 99); + register_topic(&engine).await; + engine + .on_resource_event(subscription_written_event(resource, "sub-1")) + .await; + assert_eq!(stored_status(&backend, "sub-1").await, "active"); + + engine.on_resource_event(encounter_created_event()).await; + + assert_eq!( + engine + .manager() + .get_subscription(TENANT_ID, "sub-1") + .expect("registered") + .status, + SubscriptionStatusCode::Error + ); + assert_eq!( + stored_status(&backend, "sub-1").await, + "error", + "a delivery failure that trips the breaker must reach storage" + ); +} + +/// The `off` decision — the one that exists to stop the server hammering a dead +/// endpoint — must be durable. Losing it is the amnesiac circuit breaker: on +/// restart the subscription reverts to its stored status and delivery resumes. +#[tokio::test] +async fn delivery_circuit_breaker_persists_off() { + let subscriber = handshake_ok_then_failing_subscriber().await; + + let backend = storage(); + seed(&backend, topic_resource_type(), topic_resource()).await; + let resource = subscription_resource("sub-1", "requested", &subscriber.uri()); + seed(&backend, "Subscription", resource.clone()).await; + + // `handle_delivery_failure` tests `off` first, so both at 1 sends the very + // first failure straight to `off`. + let engine = engine_with_thresholds(Arc::clone(&backend), 1, 1); + register_topic(&engine).await; + engine + .on_resource_event(subscription_written_event(resource, "sub-1")) + .await; + assert_eq!(stored_status(&backend, "sub-1").await, "active"); + + engine.on_resource_event(encounter_created_event()).await; + + assert_eq!( + engine + .manager() + .get_subscription(TENANT_ID, "sub-1") + .expect("registered") + .status, + SubscriptionStatusCode::Off + ); + assert_eq!( + stored_status(&backend, "sub-1").await, + "off", + "the circuit breaker's `off` decision must survive a restart" + ); +} + +/// Write-back must touch `status` and nothing else — in particular the +/// version-shape markers rehydration infers the FHIR version from +/// (`topic`/`channelType` vs `criteria`/`channel`) must survive, or the next +/// restart parses the resource as the wrong version. +#[tokio::test] +async fn write_back_preserves_every_other_field() { + let subscriber = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&subscriber) + .await; + + let backend = storage(); + seed(&backend, topic_resource_type(), topic_resource()).await; + let mut resource = subscription_resource("sub-1", "requested", &subscriber.uri()); + resource["reason"] = json!("a field the engine knows nothing about"); + seed(&backend, "Subscription", resource.clone()).await; + + let engine = engine_with_writeback(Arc::clone(&backend)); + register_topic(&engine).await; + engine + .on_resource_event(subscription_written_event(resource.clone(), "sub-1")) + .await; + + let stored = backend + .read(&tenant_context(TENANT_ID), "Subscription", "sub-1") + .await + .expect("read") + .expect("exists"); + let content = stored.content(); + + assert_eq!(content.get("status").unwrap(), "active"); + assert_eq!( + content.get("reason").unwrap(), + "a field the engine knows nothing about", + "write-back must not drop unknown fields" + ); + if uses_backport_ig() { + assert!(content.get("criteria").is_some(), "R4 topic marker lost"); + assert!(content.get("channel").is_some(), "R4 channel marker lost"); + } else { + assert!(content.get("topic").is_some(), "native topic marker lost"); + assert!( + content.get("channelType").is_some(), + "native channel marker lost" + ); + } +} + +/// Write-back is bounded: re-running activation against a resource that already +/// reads `active` must issue no write at all, so a restart with many +/// subscriptions does not produce a version-bump burst. +#[tokio::test] +async fn no_write_when_stored_status_already_matches() { + let subscriber = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&subscriber) + .await; + + let backend = storage(); + seed(&backend, topic_resource_type(), topic_resource()).await; + let resource = subscription_resource("sub-1", "requested", &subscriber.uri()); + seed(&backend, "Subscription", resource.clone()).await; + + let engine = engine_with_writeback(Arc::clone(&backend)); + register_topic(&engine).await; + engine + .on_resource_event(subscription_written_event(resource, "sub-1")) + .await; + + // Anti-vacuity: the first activation must genuinely have written, or the + // version comparison below compares two untouched versions and passes for + // the wrong reason. + assert_eq!( + assert_registered(&engine, "sub-1"), + SubscriptionStatusCode::Active + ); + assert_eq!(stored_status(&backend, "sub-1").await, "active"); + let version_after_activation = stored_version(&backend, "sub-1").await; + + // Re-register from the now-`active` stored resource and re-run activation: + // the status is unchanged, so no second version may be minted. + let active_resource = subscription_resource("sub-1", "requested", &subscriber.uri()); + engine + .on_resource_event(subscription_written_event(active_resource, "sub-1")) + .await; + + assert_eq!( + assert_registered(&engine, "sub-1"), + SubscriptionStatusCode::Active + ); + assert_eq!( + stored_version(&backend, "sub-1").await, + version_after_activation, + "a no-op status transition must not bump versionId" + ); +} + +/// With no status store attached the engine behaves exactly as it did before +/// #357: transitions apply in memory and storage is untouched. This is the +/// `HFS_SUBSCRIPTION_PERSIST_STATUS=false` / embedder path. +#[tokio::test] +async fn without_a_status_store_storage_is_untouched() { + let subscriber = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&subscriber) + .await; + + let backend = storage(); + seed(&backend, topic_resource_type(), topic_resource()).await; + let resource = subscription_resource("sub-1", "requested", &subscriber.uri()); + seed(&backend, "Subscription", resource.clone()).await; + + // Note: no `.with_status_store(...)`. + let engine = SubscriptionEngine::new( + SubscriptionConfig { + max_retries: 0, + handshake_max_attempts: 1, + ..SubscriptionConfig::default() + }, + "http://localhost:8080".to_string(), + ); + register_topic(&engine).await; + engine + .on_resource_event(subscription_written_event(resource, "sub-1")) + .await; + + assert_eq!( + engine + .manager() + .get_subscription(TENANT_ID, "sub-1") + .expect("registered") + .status, + SubscriptionStatusCode::Active, + "the in-memory transition must still happen" + ); + assert_eq!( + stored_status(&backend, "sub-1").await, + "requested", + "without a status store, storage must not be written" + ); +} + +/// `persist_status = false` is the kill switch: the store is attached but the +/// engine must not write through it. +#[tokio::test] +async fn persist_status_false_suppresses_write_back() { + let subscriber = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&subscriber) + .await; + + let backend = storage(); + seed(&backend, topic_resource_type(), topic_resource()).await; + let resource = subscription_resource("sub-1", "requested", &subscriber.uri()); + seed(&backend, "Subscription", resource.clone()).await; + + let engine = SubscriptionEngine::new( + SubscriptionConfig { + max_retries: 0, + handshake_max_attempts: 1, + persist_status: false, + ..SubscriptionConfig::default() + }, + "http://localhost:8080".to_string(), + ) + .with_status_store(Arc::clone(&backend) as Arc); + register_topic(&engine).await; + + assert!(!engine.persists_status()); + + engine + .on_resource_event(subscription_written_event(resource, "sub-1")) + .await; + + // Anti-vacuity: the transition itself must have happened, so the assertion + // below proves the write was *suppressed* rather than never attempted. + assert_eq!( + assert_registered(&engine, "sub-1"), + SubscriptionStatusCode::Active + ); + assert_eq!( + stored_status(&backend, "sub-1").await, + "requested", + "HFS_SUBSCRIPTION_PERSIST_STATUS=false must suppress the write" + ); +} + +/// A deleted subscription must not resurrect itself: if the resource is gone by +/// the time a transition lands, write-back is a no-op rather than an error or a +/// recreate. +#[tokio::test] +async fn write_back_on_a_deleted_subscription_is_a_no_op() { + let subscriber = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&subscriber) + .await; + + let backend = storage(); + seed(&backend, topic_resource_type(), topic_resource()).await; + let resource = subscription_resource("sub-1", "requested", &subscriber.uri()); + seed(&backend, "Subscription", resource.clone()).await; + + let engine = engine_with_writeback(Arc::clone(&backend)); + register_topic(&engine).await; + + // Register without activating, then delete the resource out from under it. + engine + .manager() + .register(TENANT_ID, "sub-1", &resource, current_fhir_version()) + .expect("register"); + backend + .delete(&tenant_context(TENANT_ID), "Subscription", "sub-1") + .await + .expect("delete"); + + // Activation now transitions in memory and finds nothing to write onto. + engine + .on_resource_event(subscription_written_event(resource, "sub-1")) + .await; + + // Anti-vacuity: the in-memory transition must have happened, so write-back + // really was attempted against a resource that had vanished. + assert_eq!( + assert_registered(&engine, "sub-1"), + SubscriptionStatusCode::Active + ); + + let read_back = backend + .read(&tenant_context(TENANT_ID), "Subscription", "sub-1") + .await; + match read_back { + Ok(None) | Err(_) => {} + Ok(Some(stored)) => panic!( + "write-back resurrected a deleted subscription: {:?}", + stored.content() + ), + } +}