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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
//! Session runtime-work projection reads and durable lifecycle events.
use crate::{SessionError, SessionManager, db};
use bcode_session_models::{
ClientId, RuntimeWorkStatus, SessionEvent, SessionEventKind, SessionId, WorkId,
};
impl SessionManager {
/// Return active runtime-work rows through the session actor's DB connection.
///
/// # Errors
///
/// Returns [`SessionError::NotFound`] when the session does not exist, or
/// [`SessionError::ProjectionStale`] when the DB projection is not current.
pub async fn active_runtime_work(
&self,
session_id: SessionId,
) -> Result<Vec<db::RuntimeWorkProjection>, SessionError> {
let phase = bcode_metrics::startup::phase("session.active_work.acquire_handle");
let result = self.session_handle(session_id).await;
phase.finish_result(&result);
let handle = result?;
let phase = bcode_metrics::startup::phase("session.active_work.actor_query");
let result = handle.active_runtime_work().await;
phase.finish_result(&result);
result
}
/// Probe active work without sleeping on database-open contention.
///
/// Intended for best-effort recovery discovery. Contention remains an error;
/// callers must defer rather than interpret it as an empty work set.
/// Existing actor connections and projection freshness checks are unchanged.
///
/// # Errors
/// Returns normal active-work errors, including immediate database-open contention.
pub async fn probe_active_runtime_work(
&self,
session_id: SessionId,
) -> Result<Vec<db::RuntimeWorkProjection>, SessionError> {
let phase = bcode_metrics::startup::phase("session.active_work.acquire_handle");
let result = self.session_handle(session_id).await;
phase.finish_result(&result);
let handle = result?;
let phase = bcode_metrics::startup::phase("session.active_work.actor_query");
let result = handle.active_runtime_work_with_policy(true).await;
phase.finish_result(&result);
result
}
/// Return latest runtime-work rows from the DB read model.
///
/// # Errors
///
/// Returns [`SessionError::NotFound`] when the session does not exist, or
/// [`SessionError::ProjectionStale`] when the DB projection is not current.
pub async fn runtime_work_history(
&self,
session_id: SessionId,
limit: usize,
) -> Result<Vec<db::RuntimeWorkProjection>, SessionError> {
self.ensure_session_loaded(session_id).await?;
let store = self
.store
.as_ref()
.ok_or(SessionError::DbUnavailable(session_id))?;
// Bounded read path: the local handle is dropped at the end of this function, which
// releases the backend connection. It must not close, because closing checkpoints the WAL
// and would make a normal read mutate canonical storage.
let db = db::SessionDb::open_existing_turso_in_root(session_id, &store.root_path()).await?;
let expected_last_sequence = db.last_event_sequence().await?.unwrap_or(0);
let checkpoint = db
.materialized_projection_checkpoint(db::MaterializedProjection::RuntimeWork)
.await?;
if checkpoint.is_some_and(|checkpoint| checkpoint >= expected_last_sequence) {
return Ok(db.runtime_work_history(limit).await?);
}
Err(SessionError::ProjectionStale {
session_id,
projection: "runtime_work",
checkpoint,
expected: expected_last_sequence,
})
}
/// Append a runtime-work started event to a session.
///
/// # Errors
///
/// Returns an error when the session does not exist or the event cannot be persisted.
pub async fn append_runtime_work_started(
&self,
session_id: SessionId,
event: SessionEventKind,
) -> Result<SessionEvent, SessionError> {
self.append_event(session_id, event).await
}
/// Append a runtime-work cancellation request event to a session.
///
/// # Errors
///
/// Returns an error when the session does not exist or the event cannot be persisted.
pub async fn append_runtime_work_cancel_requested(
&self,
session_id: SessionId,
work_id: WorkId,
requested_at_ms: Option<u64>,
client_id: Option<ClientId>,
) -> Result<SessionEvent, SessionError> {
self.append_event(
session_id,
SessionEventKind::RuntimeWorkCancelRequested {
work_id,
requested_at_ms,
client_id,
},
)
.await
}
/// Append a runtime-work finished event to a session.
///
/// # Errors
///
/// Returns an error when the session does not exist or the event cannot be persisted.
pub async fn append_runtime_work_finished(
&self,
session_id: SessionId,
work_id: WorkId,
status: RuntimeWorkStatus,
finished_at_ms: Option<u64>,
message: Option<String>,
) -> Result<SessionEvent, SessionError> {
self.append_event(
session_id,
SessionEventKind::RuntimeWorkFinished {
work_id,
status,
finished_at_ms,
message,
},
)
.await
}
}