merge: CLOUD-216 per-subscription fail-closed delivery semantics (spec 0028)

Adds opt-in FailureMode {FailOpen, FailClosed} per subscriber. Fail-closed
subscribers halt (hold position, fire on_halt, DLQ the bad event) on
deserialize/observer failures in live, catch-up, and drain paths instead
of limping past; wedged subscribers are excluded from the shared floor
and driven by a private re-seeding fetch so peers keep advancing; gaps
are refused (TimeoutBackstop) with FenceCleared rollback recovery;
release_halt gives operators a forward-only release; the ReplayAlways
HWM advances only across the contiguous prefix (closes CLOUD-227).

Reviewed: 3-round spec loop, 3-round plan loop, per-phase reviews
(P1-P6), and a 2-round parent-orchestrated pre-merge loop; one P1
(panicking HaltCallback killed the listener) found and fixed in the
final round. Consolidated epoch_pg suite 259 passed / 0 failed.

Author: Istar-Eldritch

CHANGELOG.md

@@ -9,6 +9,49 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Per-subscriber fail-closed delivery semantics** (`epoch_core`, `epoch_pg`, CLOUD-216) —
+  an opt-in `FailureMode { FailOpen (default), FailClosed }` letting a subscriber halt
+  rather than silently skip an event it cannot apply in order:
+  - **`epoch_core`** — `FailureMode` enum (`#[non_exhaustive]`, re-exported from the
+    prelude); defaulted `failure_mode()` on `EventObserver`, `Projection`, and `Saga`,
+    forwarded by `ProjectionHandler`, `SagaHandler`, `SagaAdapter`, and the
+    `impl Saga for Arc<S>` blanket. Existing implementors are unchanged (default `FailOpen`).
+  - **Live batch path** — a `FailClosed` subscriber that hits a deserialize failure or an
+    observer-retry exhaustion holds its contiguous checkpoint below the bad sequence
+    (writing a `unrecoverable: deserialize: …` DLQ row for the deserialize case), fires
+    the new `on_halt` callback once on entry, and self-heals on the next batch once the
+    cause is fixed. A panicking observer is now contained (caught in
+    `process_event_with_retry`) instead of killing the listener task — a deliberate
+    fail-open behaviour change: the panic routes through the existing retry/DLQ
+    machinery.
+  - **Catch-up and drain paths** — the same hold semantics apply during initial catch-up
+    and inline-drain: a `FailClosed` subscriber that hits a bad row while catching up
+    halts at that row (the subscriber is still registered and the live listener takes
+    over from there), and self-heals once the cause is fixed.
+  - **Gap refusal** — a `FailClosed` subscriber refuses the `gap_timeout` backstop: if
+    the backstop would advance past an unproven gap, the subscriber halts with
+    `HaltReason::GapUnproven` and stays held until the gap is proven permanent (fence
+    clears) or an operator releases it. The fence-clear branch is untouched: a gap
+    proven never to have existed advances under both modes.
+  - **Wedge isolation** — a halted `FailClosed` subscriber is excluded from the shared
+    event-window floor and served by its own private re-seeding fetch each cycle, so a
+    permanently wedged subscriber cannot pin the delivery window of healthy peers.
+  - **`PgEventBus::release_halt(subscriber_id, past_sequence)`** — new operator API
+    for advancing a wedged subscriber's persisted checkpoint past a held sequence it
+    never finished. Forward-only (rejects `past_sequence` at or below the current
+    checkpoint with `PgEventBusError::BackwardRelease`). Fires the `on_halt` callback
+    with `HaltReason::Released` on success. Events already applied above the released
+    position are folded in without re-delivery.
+  - **`ReplayAlways` contiguous HWM** (CLOUD-227) — a `FailClosed` `ReplayAlways`
+    subscriber now tracks a contiguous high-water mark that advances only across the
+    unbroken prefix of successfully applied sequences, matching the checkpoint semantics
+    of `Checkpointed` subscribers. A wedged `ReplayAlways` subscriber is likewise
+    excluded from the shared floor; its remedy is a fresh `subscribe()` call for a full
+    replay (no `release_halt` — `ReplayAlways` has no persisted checkpoint row).
+  - New public API in `epoch_pg::event_bus`: `HaltCallback`, `HaltInfo`, `HaltReason`
+    (`#[non_exhaustive]`, variants: `DeserializeFailure`, `ObserverFailure`,
+    `GapUnproven`, `Released`). No schema migration.
+
 - **Subscriber readiness + startup safety** (`epoch_core`, `epoch_pg`, CLOUD-221) —
   first-class lag/readiness API on `PgEventBus`, a `ReplayAlways` subscription mode for
   in-memory projections that replay from zero every boot, and startup catch-up so a
@@ -212,9 +255,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `setup_trigger()` once per bus to drop the legacy fixed-name trigger; leaving it in
   place is not unsafe, only redundant — it produces one duplicate NOTIFY per insert,
   which is discarded by the per-event checkpoint check.
-- **BREAKING** (`epoch_pg`): `PgEventBusError` gains two new variants,
-  `SubscriberNotFound` and `InlineDispatchNotSupported`. The enum is not
-  `#[non_exhaustive]`, so a downstream exhaustive `match` on it will stop compiling.
+- **BREAKING** (`epoch_pg`): `PgEventBusError` gains three new variants,
+  `SubscriberNotFound`, `InlineDispatchNotSupported`, and `BackwardRelease`
+  (returned by `release_halt` when `past_sequence` is at or below the current
+  persisted checkpoint). The enum is not `#[non_exhaustive]`, so a downstream
+  exhaustive `match` on it will stop compiling.
 - Released the `projections` lock across the listener's batch drain instead of
   holding it for the whole backlog (`epoch_pg`, CLOUD-225): every readiness method
   also locks `projections`, so `subscriber_lag`, `wait_until_caught_up`, and
@@ -248,6 +293,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `on_gap_timeout: Option<Arc<dyn GapTimeoutCallback>>` field (defaults to `None`).
   Code that constructs `ReliableDeliveryConfig` using struct-literal syntax (rather than
   `..Default::default()`) must add `on_gap_timeout: None` to the literal.
+- **Source-compat note** (`epoch_pg`, CLOUD-216, `feat(pg)!`): `ReliableDeliveryConfig`
+  gains the new `on_halt: Option<Arc<dyn HaltCallback>>` field (defaults to `None`),
+  fired when a fail-closed subscriber halts delivery. Code that constructs
+  `ReliableDeliveryConfig` using struct-literal syntax (rather than
+  `..Default::default()`) must add `on_halt: None` to the literal.
 - **Source-compat note**: `ReliableDeliveryConfig` (`epoch_pg`) gains the new
   `snapshot_fencing: bool` field (defaults to `true`). Code that constructs
   `ReliableDeliveryConfig` using struct-literal syntax (rather than

epoch_core/src/event_store.rs

@@ -187,6 +187,52 @@ pub trait EventBus {
         T: EventObserver<Self::EventType> + Send + Sync + 'static;
 }
 
+/// How a subscriber reacts when an event cannot be applied.
+///
+/// Controls the bus's behaviour at every failure point (deserialisation error,
+/// observer error, unproven gap) for a given subscriber.
+///
+/// # Default
+///
+/// The default is [`FailureMode::FailOpen`]: on any failure the bus logs, skips
+/// the event, and advances the checkpoint past it. This is the correct choice
+/// for most subscribers.
+///
+/// # `FailClosed` and frozen read models
+///
+/// When a subscriber opts into [`FailureMode::FailClosed`], the bus halts
+/// delivery at the first unrecoverable failure: the subscriber's checkpoint is
+/// held below the bad event and no further events are applied to *that
+/// subscriber* until the condition is resolved (the bad row is fixed, or an
+/// operator calls `release_halt`). The halt is **subscriber-local**: healthy
+/// subscribers keep advancing because a wedged subscriber is excluded from the
+/// shared event-window floor and served by its own private re-seeding fetch,
+/// so it cannot pin the delivery window of any healthy peer.
+///
+/// **Cross-group consequence (R12):** while the halted subscriber's read model
+/// is frozen, later priority groups keep running against that stale view. For a
+/// deny-heavy oracle (e.g. an auth projection), stale data errs on the side of
+/// denying rather than over-admitting — the freeze fails safe. Recovery is
+/// self-healing: once the cause is fixed, the held event and everything after
+/// it are applied exactly once in order on the next batch cycle. For a gap
+/// that never resolves, an operator release advances the subscriber's persisted
+/// position past the held sequence. A permanent unreleased halt means a
+/// permanently frozen read model for that subscriber; it does not starve or
+/// slow any other subscriber on the bus.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+#[non_exhaustive]
+pub enum FailureMode {
+    /// Default. On any delivery failure, log, skip, and advance the checkpoint
+    /// past the failed event. The read model may silently diverge from the
+    /// event log, but delivery continues unimpeded.
+    #[default]
+    FailOpen,
+    /// Halt delivery at the first unrecoverable failure. The checkpoint is held
+    /// below the bad event; no further events are applied until the condition
+    /// clears. See the [`FailureMode`] docs for the cross-group wedge risk.
+    FailClosed,
+}
+
 /// How a subscriber relates to persisted checkpoints.
 ///
 /// Controls whether the event bus reads and writes a checkpoint row for this
@@ -258,6 +304,15 @@ where
     fn subscription_mode(&self) -> SubscriptionMode {
         SubscriptionMode::Checkpointed
     }
+
+    /// How the bus reacts when this subscriber cannot apply an event.
+    ///
+    /// Defaults to [`FailureMode::FailOpen`]: log, skip, and advance past the
+    /// failed event. Override and return [`FailureMode::FailClosed`] for
+    /// subscribers that must never silently diverge from the event log.
+    fn failure_mode(&self) -> FailureMode {
+        FailureMode::FailOpen
+    }
 }
 
 /// Used to construct an event stream from a slice.
@@ -288,7 +343,7 @@ where
 /// A reference-based event stream constructed from a slice.
 ///
 /// This is an optimized version of [`SliceEventStream`] that yields references
-/// to events instead of cloning them. Used internally by [`Aggregate::handle`]
+/// to events instead of cloning them. Used internally by `Aggregate::handle`
 /// to avoid cloning freshly created events during re-hydration.
 ///
 /// # Performance

epoch_core/src/lib.rs

@@ -36,6 +36,7 @@ pub mod upcasting;
 #[cfg(feature = "testing")]
 pub mod testing;
 
+pub use event_store::FailureMode;
 pub use event_store::SubscriptionMode;
 pub use subscriber_id::SubscriberId;
 

epoch_core/src/projection.rs

@@ -11,7 +11,7 @@
 //! be subscribed to the event bus for their own events. This would cause
 //! duplicate writes and race conditions.
 //!
-//! The type system enforces this: `Aggregate` extends [`EventApplicator`](crate::event_applicator::EventApplicator),
+//! The type system enforces this: `Aggregate` extends [`EventApplicator`],
 //! NOT `Projection`, so aggregates cannot be wrapped in `ProjectionHandler`.
 
 use std::sync::Arc;
@@ -174,12 +174,21 @@ where
 
     /// How this projection relates to persisted checkpoints.
     ///
-    /// Defaults to [`SubscriptionMode::Checkpointed`]. Override and return
-    /// [`SubscriptionMode::ReplayAlways`] for in-memory projections that must
+    /// Defaults to [`crate::event_store::SubscriptionMode::Checkpointed`]. Override and return
+    /// [`crate::event_store::SubscriptionMode::ReplayAlways`] for in-memory projections that must
     /// replay from sequence 0 on every process start.
     fn subscription_mode(&self) -> crate::event_store::SubscriptionMode {
         crate::event_store::SubscriptionMode::Checkpointed
     }
+
+    /// How the bus reacts when this projection cannot apply an event.
+    ///
+    /// Defaults to [`crate::event_store::FailureMode::FailOpen`]: log, skip, and advance past the
+    /// failed event. Override and return [`crate::event_store::FailureMode::FailClosed`] for
+    /// projections that must never silently diverge from the event log.
+    fn failure_mode(&self) -> crate::event_store::FailureMode {
+        crate::event_store::FailureMode::FailOpen
+    }
 }
 
 /// Wraps a [`Projection`] to implement [`EventObserver`](crate::event_store::EventObserver)
@@ -260,6 +269,10 @@ where
     fn subscription_mode(&self) -> crate::event_store::SubscriptionMode {
         self.0.subscription_mode()
     }
+
+    fn failure_mode(&self) -> crate::event_store::FailureMode {
+        self.0.failure_mode()
+    }
 }
 
 #[cfg(test)]

epoch_core/src/saga.rs

@@ -129,13 +129,22 @@ where
 
     /// How this saga relates to persisted checkpoints.
     ///
-    /// Defaults to [`SubscriptionMode::Checkpointed`]. Override and return
-    /// [`SubscriptionMode::ReplayAlways`] for in-memory sagas that must
+    /// Defaults to [`crate::event_store::SubscriptionMode::Checkpointed`]. Override and return
+    /// [`crate::event_store::SubscriptionMode::ReplayAlways`] for in-memory sagas that must
     /// replay from sequence 0 on every process start.
     fn subscription_mode(&self) -> crate::event_store::SubscriptionMode {
         crate::event_store::SubscriptionMode::Checkpointed
     }
 
+    /// How the bus reacts when this saga cannot apply an event.
+    ///
+    /// Defaults to [`crate::event_store::FailureMode::FailOpen`]: log, skip, and advance past the
+    /// failed event. Override and return [`crate::event_store::FailureMode::FailClosed`] for
+    /// sagas that must never silently diverge from the event log.
+    fn failure_mode(&self) -> crate::event_store::FailureMode {
+        crate::event_store::FailureMode::FailOpen
+    }
+
     /// Processes an incoming event, applies it to the saga, and persists the resulting state.
     ///
     /// This method is called by the blanket [`EventObserver`] implementation. It handles:
@@ -230,6 +239,10 @@ where
     fn subscription_mode(&self) -> crate::event_store::SubscriptionMode {
         (**self).subscription_mode()
     }
+
+    fn failure_mode(&self) -> crate::event_store::FailureMode {
+        (**self).failure_mode()
+    }
 }
 
 /// A wrapper type that provides an [`EventObserver`] implementation for [`Saga`] types.
@@ -297,6 +310,10 @@ where
     fn subscription_mode(&self) -> crate::event_store::SubscriptionMode {
         self.0.subscription_mode()
     }
+
+    fn failure_mode(&self) -> crate::event_store::FailureMode {
+        self.0.failure_mode()
+    }
 }
 
 /// Adapter that lets a [`Saga`] subscribe to a *foreign* event bus carrying a
@@ -458,6 +475,10 @@ where
     fn subscription_mode(&self) -> crate::event_store::SubscriptionMode {
         self.saga.subscription_mode()
     }
+
+    fn failure_mode(&self) -> crate::event_store::FailureMode {
+        self.saga.failure_mode()
+    }
 }
 
 #[cfg(test)]

epoch_core/src/upcasting.rs

@@ -1,7 +1,7 @@
 //! Event schema evolution via forward-only **upcasting**.
 //!
 //! Once an event variant is persisted, its serialized shape is frozen in the store
-//! forever, yet the in-memory [`EventData`](crate::event::EventData) type it must
+//! forever, yet the in-memory [`EventData`] type it must
 //! deserialize into keeps changing as the domain evolves. Upcasting bridges that gap:
 //! a stored payload is transformed forward, **one version at a time**, to the current
 //! schema *before* it is deserialized into the domain type.
@@ -38,7 +38,7 @@ use uuid::Uuid;
 
 use crate::event::EventData;
 
-/// Re-export [`SchemaVersion`](crate::event::SchemaVersion) from the event module so
+/// Re-export [`SchemaVersion`] from the event module so
 /// that code importing from this module gets a consistent, unambiguous type.
 pub use crate::event::SchemaVersion;
 
@@ -88,7 +88,7 @@ impl<'a> UpcastContext<'a> {
 /// projections.
 pub trait Upcaster: Send + Sync {
     /// The `event_type` (PascalCase, matching
-    /// [`EventData::event_type`](crate::event::EventData::event_type)) this step applies
+    /// [`EventData::event_type`]) this step applies
     /// to.
     fn event_type(&self) -> &str;