fix(subscriptions): persist server-driven status transitions (#357)

.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`.

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<S>(
+    state: &AppState<S>,
+    engine: &std::sync::Arc<helios_subscriptions::SubscriptionEngine>,
+    tenant: &TenantExtractor,
+    id: &str,
+) -> RestResult<StatusSnapshot>
+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,30 +165,26 @@ 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",
         "type": bundle_type,
         "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, "");
+    }
+}

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<dyn helios_persistence::core::ResourceStorage>);
+            // 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)

crates/subscriptions/src/config.rs

@@ -49,6 +49,34 @@ pub struct SubscriptionConfig {
 
     /// FHIR Messaging channel settings. `None` disables the messaging dispatcher.
     pub messaging: Option<MessagingSettings>,
+
+    /// 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,
         }
     }
 }