onlyne_client/runtime/runloop/
run.rs

1use super::config::{
2    ClientInit, NOT_READY_PAUSE_MS, PULL_HOLD_MS, PULL_LIMIT, PULL_PAUSE_MS, RunState,
3    SHUTDOWN_CLOSE_BUDGET,
4};
5use super::link::{link_loop, transient};
6use super::sessions::{accept_delivery, outcome_loop};
7use crate::session::adapter_socket::AdapterSocket;
8use crate::session::dispatch::{self, ClientLink, DispatchState};
9use anyhow::{Result, anyhow};
10use onlyne_layout::RoleWorkspace;
11use onlyne_proto::{AckArgs, ClientOp, ControlOp, Delivery, PullArgs, PullReply};
12use onlyne_store::ClientStore;
13use std::sync::atomic::Ordering;
14use std::time::Duration;
15use tokio::time::sleep;
16
17/// Run the role runtime until a permanent handshake failure or a dead local
18/// surface stops it.
19///
20/// The acceptor task owns the workspace socket bind, and its failure ends this
21/// run the same way a permanent link failure does: a client that keeps a TLS
22/// link while its socket is unbound reads as connected from the server while
23/// every verb the workspace issues fails, so the exit status and the message on
24/// stderr are the operator's only notice.
25pub async fn run(init: ClientInit) -> Result<()> {
26    let workspace = RoleWorkspace::resolve(&init.workspace);
27    let store = ClientStore::open(workspace.client_db_path())?;
28    let state = RunState::new(&init, store)?;
29    let mut acceptor = tokio::spawn(acceptor(init.clone(), state.clone()));
30    let closing = tokio::spawn(close_on_signal(state.dispatch.clone()));
31    let mut outcomes = tokio::spawn(outcome_loop(state.clone()));
32    let outcome = tokio::select! {
33        link = link_loop(&init, &state) => link,
34        served = &mut acceptor => match served {
35            Ok(Ok(())) => Err(anyhow!("the adapter surface stopped serving")),
36            Ok(Err(error)) => Err(error),
37            Err(error) => Err(anyhow!("adapter acceptor task ended: {error}")),
38        },
39        pumped = &mut outcomes => match pumped {
40            Ok(Ok(())) => Err(anyhow!("the session outcome pump stopped")),
41            Ok(Err(error)) => Err(error),
42            Err(error) => Err(anyhow!("session outcome pump task ended: {error}")),
43        },
44    };
45    acceptor.abort();
46    closing.abort();
47    outcomes.abort();
48    outcome
49}
50
51/// Close live sessions when the operator stops the client.
52///
53/// `SIGTERM` ends the foreground client, and the default disposition would
54/// kill the process with every tab it opened still running: the resources
55/// would outlive the only thing that can address them. Each session closes with
56/// [`onlyne_session::CloseReason::Shutdown`] first, so the backend record and
57/// the plugin-facing tab map end truthfully.
58#[cfg(unix)]
59pub(super) async fn close_on_signal(dispatch: DispatchState) {
60    use tokio::signal::unix::{SignalKind, signal};
61    let mut terminate = match signal(SignalKind::terminate()) {
62        Ok(stream) => stream,
63        Err(error) => {
64            tracing::warn!(error = %error, "SIGTERM handler was not installed");
65            return;
66        }
67    };
68    let mut interrupt = match signal(SignalKind::interrupt()) {
69        Ok(stream) => stream,
70        Err(error) => {
71            tracing::warn!(error = %error, "SIGINT handler was not installed");
72            return;
73        }
74    };
75    tokio::select! {
76        _ = terminate.recv() => tracing::info!("SIGTERM: closing live sessions"),
77        _ = interrupt.recv() => tracing::info!("SIGINT: closing live sessions"),
78    }
79    dispatch::close_all(
80        &dispatch,
81        onlyne_session::CloseReason::Shutdown,
82        SHUTDOWN_CLOSE_BUDGET,
83    );
84    std::process::exit(0);
85}
86
87/// Close live sessions when the operator stops the client.
88///
89/// Windows has no SIGTERM; operators use `onlyne shutdown` for a graceful
90/// daemon stop. Ctrl-C is the console interrupt, and it runs the same
91/// close_all budget the unix SIGINT path uses.
92#[cfg(windows)]
93pub(super) async fn close_on_signal(dispatch: DispatchState) {
94    // `ctrl_c()` installs synchronously and hands back the watch stream; the
95    // await belongs on `recv`, which yields once per console interrupt.
96    let mut interrupt = match tokio::signal::windows::ctrl_c() {
97        Ok(stream) => stream,
98        Err(error) => {
99            tracing::warn!(error = %error, "Ctrl-C handler was not installed");
100            return;
101        }
102    };
103    interrupt.recv().await;
104    tracing::info!("Ctrl-C: closing live sessions");
105    dispatch::close_all(
106        &dispatch,
107        onlyne_session::CloseReason::Shutdown,
108        SHUTDOWN_CLOSE_BUDGET,
109    );
110    std::process::exit(0);
111}
112
113/// Bind the adapter socket and serve it for the life of the process.
114///
115/// The bind is the first act, and its error leaves this task: a socket that
116/// never opened is a workspace whose plugins cannot mount, and only the exit
117/// status says so. Once the listener exists the surface stays up — a failed
118/// `accept` is logged and retried by [`AdapterSocket::accept_loop`] — so this
119/// task ends the run exactly when the local surface could not start.
120pub(super) async fn acceptor(init: ClientInit, state: RunState) -> Result<()> {
121    let workspace = RoleWorkspace::resolve(&init.workspace);
122    let cluster = state
123        .store
124        .config("cluster")
125        .ok()
126        .flatten()
127        .unwrap_or_default();
128    let socket = AdapterSocket {
129        workspace: workspace.root().to_path_buf(),
130        role: init.role.clone(),
131        cluster,
132        server: init.server.clone(),
133        dispatch: state.dispatch.clone(),
134    };
135    let (listener, _endpoint) = socket.bind().await?;
136    socket.accept_loop(listener).await
137}
138
139/// The pull-ack task: drain what the server queued, then settle what the local
140/// side finished.
141pub(super) async fn pull_ack_loop(
142    init: ClientInit,
143    link: ClientLink,
144    state: RunState,
145) -> Result<()> {
146    loop {
147        if !state.accept_new.load(Ordering::SeqCst) {
148            sleep(Duration::from_millis(PULL_PAUSE_MS)).await;
149            continue;
150        }
151        // A role at `max_sessions` stops asking for work it has nowhere to run
152        // (plan §5), and the same pause must not stop it hearing the command that
153        // frees a slot: a full role is exactly the one whose operator wants to
154        // `recycle` or `focus`. Control rows still travel on the ordinary pull
155        // when the role has capacity.
156        let control_only = !state.dispatch.has_capacity();
157        let reply = match link
158            .request(ClientOp::Pull(PullArgs {
159                role: Some(init.role.clone()),
160                limit: PULL_LIMIT,
161                hold_ms: Some(PULL_HOLD_MS),
162                control_only: control_only.then_some(true),
163            }))
164            .await
165        {
166            Ok(reply) => reply,
167            Err(error) if transient(&error) => {
168                sleep(Duration::from_millis(NOT_READY_PAUSE_MS)).await;
169                continue;
170            }
171            Err(error) => return Err(anyhow!(error)),
172        };
173        if !reply.ok {
174            tracing::warn!(error = ?reply.error, "pull refused");
175            sleep(Duration::from_millis(PULL_PAUSE_MS)).await;
176            continue;
177        }
178        let Some(data) = reply.data else { continue };
179        let pulled: PullReply = serde_json::from_value(data)?;
180        for delivery in pulled.deliveries {
181            accept_delivery(&state, &delivery).await;
182        }
183        state.set_cursor(pulled.seq);
184        sleep(Duration::from_millis(PULL_PAUSE_MS)).await;
185    }
186}
187
188/// Apply one delivered control command and settle its row.
189///
190/// The row settles whether or not this role still holds the task it names. A
191/// command whose session already ended has nothing left to act on, and leaving
192/// the row in flight would report an operator's `control` as undelivered.
193pub(super) async fn settle_control(state: &RunState, delivery: &Delivery) {
194    let ack = |accepted: bool, reason: Option<String>| AckArgs {
195        msg_id: delivery.msg_id.clone(),
196        op_id: None,
197        accepted,
198        reason,
199    };
200    let Some(op) = delivery.envelope.control.clone() else {
201        tracing::warn!(msg_id = %delivery.msg_id, "control delivery carried no command");
202        state.dispatch.push_settled(ack(
203            false,
204            Some("control delivery carried no control op".to_string()),
205        ));
206        return;
207    };
208    match dispatch::on_control(&state.dispatch, &op).await {
209        Ok(held) => {
210            tracing::info!(
211                op = op.name(),
212                task = %op.task_id(),
213                held,
214                "control command applied"
215            );
216            state.dispatch.push_settled(ack(true, None));
217            if held && matches!(op, ControlOp::Recycle { .. } | ControlOp::Cancel { .. }) {
218                // A published exit releases the task's in-flight rows. The
219                // command's own row is one of them until its ack is enqueued.
220                if let Err(error) = dispatch::sync_session(&state.dispatch, op.task_id()).await {
221                    tracing::warn!(
222                        error = %error,
223                        task = %op.task_id(),
224                        "a closed control session was not published"
225                    );
226                }
227            }
228        }
229        Err(error) => {
230            tracing::warn!(error = %error, op = op.name(), "control command refused");
231            state
232                .dispatch
233                .push_settled(ack(false, Some(error.to_string())));
234        }
235    }
236}
237
238#[cfg(test)]
239mod tests;
onlyne_client/runtime/runloop/run.rs

onlyne_client/runtime/runloop/
run.rs
