fix(orchestrator-core): F-Q-006 + F-Q-010 InboxRegistry expect on poison
InboxRegistry::register returned (Uuid, Receiver) even when the inner
Mutex was poisoned — `if let Ok(mut g) = self.inner.lock()` swallowed
the error, the sender was never inserted, the later deliver(id, …)
saw no entry and returned Abandoned, and the caller's `await rx`
blocked until the orchestrator's outer timeout.
Sister registries (RouteTable, AppDomainTable) already .expect()
panic on poison. Make InboxRegistry::{register, cancel, deliver} loud
in the same way.
A poisoned Mutex means a prior panic inside a critical section — the
process is unrecoverable; silently swallowing it just defers the
crash and corrupts the inbox protocol on the way.
AUDIT.md anchors: F-Q-006 (register), F-Q-010 (consistency policy).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -42,13 +42,21 @@ impl InboxRegistry {
|
|||||||
/// Allocate a new inbox id and register the sender side. The
|
/// Allocate a new inbox id and register the sender side. The
|
||||||
/// caller awaits the returned `Receiver`; the dispatcher delivers
|
/// caller awaits the returned `Receiver`; the dispatcher delivers
|
||||||
/// the outcome via `deliver(id, …)`.
|
/// the outcome via `deliver(id, …)`.
|
||||||
|
///
|
||||||
|
/// F-Q-006: panic on poison — a poisoned mutex means a prior panic
|
||||||
|
/// inside the lock-holding code path, so the process is unrecoverable.
|
||||||
|
/// Silently swallowing it (the old behaviour) leaked a dead id: the
|
||||||
|
/// sender never got inserted, the later `deliver(id, …)` saw no
|
||||||
|
/// entry and returned Abandoned, and the caller's `await rx` blocked
|
||||||
|
/// until the orchestrator's outer timeout. Sister registries
|
||||||
|
/// (RouteTable, AppDomainTable) already `.expect()` here — settle
|
||||||
|
/// on the same loud policy.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn register(&self) -> (Uuid, oneshot::Receiver<InboxResult>) {
|
pub fn register(&self) -> (Uuid, oneshot::Receiver<InboxResult>) {
|
||||||
let id = Uuid::new_v4();
|
let id = Uuid::new_v4();
|
||||||
let (tx, rx) = oneshot::channel();
|
let (tx, rx) = oneshot::channel();
|
||||||
if let Ok(mut g) = self.inner.lock() {
|
let mut g = self.inner.lock().expect("inbox registry poisoned");
|
||||||
g.insert(id, tx);
|
g.insert(id, tx);
|
||||||
}
|
|
||||||
(id, rx)
|
(id, rx)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -56,10 +64,8 @@ impl InboxRegistry {
|
|||||||
/// Drops the sender so any future `deliver` returns `Abandoned`.
|
/// Drops the sender so any future `deliver` returns `Abandoned`.
|
||||||
/// Returns `true` if the receiver was still registered.
|
/// Returns `true` if the receiver was still registered.
|
||||||
pub fn cancel(&self, id: Uuid) -> bool {
|
pub fn cancel(&self, id: Uuid) -> bool {
|
||||||
self.inner
|
let mut g = self.inner.lock().expect("inbox registry poisoned");
|
||||||
.lock()
|
g.remove(&id).is_some()
|
||||||
.map(|mut g| g.remove(&id).is_some())
|
|
||||||
.unwrap_or(false)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -72,9 +78,9 @@ impl Default for InboxRegistry {
|
|||||||
#[async_trait]
|
#[async_trait]
|
||||||
impl InboxResolver for InboxRegistry {
|
impl InboxResolver for InboxRegistry {
|
||||||
async fn deliver(&self, inbox_id: Uuid, result: InboxResult) -> InboxDeliveryOutcome {
|
async fn deliver(&self, inbox_id: Uuid, result: InboxResult) -> InboxDeliveryOutcome {
|
||||||
let Ok(mut g) = self.inner.lock() else {
|
// F-Q-006 + F-Q-010: panic on poison, matching the registry's
|
||||||
return InboxDeliveryOutcome::Abandoned;
|
// register/cancel policy and the sister registries.
|
||||||
};
|
let mut g = self.inner.lock().expect("inbox registry poisoned");
|
||||||
let Some(tx) = g.remove(&inbox_id) else {
|
let Some(tx) = g.remove(&inbox_id) else {
|
||||||
return InboxDeliveryOutcome::Abandoned;
|
return InboxDeliveryOutcome::Abandoned;
|
||||||
};
|
};
|
||||||
|
|||||||
Reference in New Issue
Block a user