1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
//! One immutable Root execution's actual per-session/account publication.
use super::AccountEventSink;
use bamboo_agent_core::AgentEvent;
use bamboo_domain::{RootActorRuntimeWrite, Storage};
use std::sync::Arc;
#[derive(Clone)]
pub struct RootActorEventPublication {
storage: Arc<dyn Storage>,
account_sink: Arc<AccountEventSink>,
owner: RootActorRuntimeWrite,
}
impl RootActorEventPublication {
pub fn new(
storage: Arc<dyn Storage>,
account_sink: Arc<AccountEventSink>,
owner: RootActorRuntimeWrite,
) -> Self {
Self {
storage,
account_sink,
owner,
}
}
pub fn matches_execution(&self, session_id: &str, run_id: &str) -> bool {
self.owner.fence.actor_id == session_id && self.owner.fence.run_id == run_id
}
/// Numeric observations and a process-local opaque identifier, never the
/// owner capability or lease-owner value.
pub(crate) fn diagnostic_identity(&self) -> (u64, u64, String) {
(
self.owner.fence.lease_epoch,
self.owner.fence.attempt,
crate::execution::workspace_diagnostics::fence_fingerprint(&self.owner.fence),
)
}
pub async fn publish(
&self,
session_id: &str,
event: &AgentEvent,
publish: Box<dyn FnOnce() -> bool + Send>,
) -> std::io::Result<()> {
let storage = self.storage.clone();
let sink = self.account_sink.clone();
let owner_for_queue = self.owner.clone();
let event = event.clone();
let session_id = session_id.to_owned();
let (queued, receipt) = tokio::sync::oneshot::channel();
self.storage
.publish_root_actor_runtime_event(
&self.owner,
Box::new(move |check_current| {
check_current()?;
if !publish() {
return Err(std::io::Error::other(
"Root per-session event publication was retired",
));
}
check_current()?;
let confirmation = if event.is_durable_change() {
Some(
sink.record_root_actor(
storage,
owner_for_queue,
Some(&session_id),
&event,
)
.ok_or_else(|| {
std::io::Error::other(
"Root account event queue rejected publication",
)
})?,
)
} else {
None
};
let _ = queued.send(confirmation);
Ok(())
}),
)
.await?;
// The account writer needs the same physical Root guards. Await only
// after the per-session publication job has released those guards.
let confirmation = receipt
.await
.map_err(|_| std::io::Error::other("Root account admission receipt closed"))?;
if let Some(confirmation) = confirmation {
if !tokio::time::timeout(std::time::Duration::from_secs(30), confirmation)
.await
.ok()
.and_then(Result::ok)
.unwrap_or(false)
{
// The account writer reports a boolean receipt, so an owner
// conflict there otherwise loses its typed cause. Revalidate
// without publishing or writing runtime state before returning the
// generic receipt failure; a lost owner must still interrupt
// its exact live transport through the event forwarder.
if let Err(error) = self
.storage
.publish_root_actor_runtime_event(
&self.owner,
Box::new(|check_current| check_current()),
)
.await
{
if error
.get_ref()
.is_some_and(|cause| cause.is::<bamboo_domain::SessionAuthorityConflict>())
{
return Err(error);
}
}
return Err(std::io::Error::other(
"Root account final publication was not confirmed",
));
}
}
Ok(())
}
}