Method run_inner

nodedb/src/control/server/session.rs:272–359

(
        &mut self,
        idle_timeout_secs: u64,
        absolute_timeout_secs: u64,
    )

    }

    async fn run_inner(
        &mut self,
        idle_timeout_secs: u64,
        absolute_timeout_secs: u64,
    ) -> crate::Result<()> {
        loop {
            // Hard-revoke check: bus consumer sent kill signal.
            if self.is_killed() {
                let msg = r#"{"status":"error","sqlstate":"57P01","error":"session revoked by administrator"}"#;
                let resp_len = (msg.len() as u32).to_be_bytes();
                let _ = self.stream.write_all(&resp_len).await;
                let _ = self.stream.write_all(msg.as_bytes()).await;
                return Ok(());
            }

            // Enforce absolute session lifetime (SQLSTATE 57P01 "admin shutdown").
            if absolute_timeout_secs > 0
                && self.connected_at.elapsed().as_secs() >= absolute_timeout_secs
            {
                debug!(
                    "session absolute timeout ({}s), closing connection",
                    absolute_timeout_secs
                );
                let msg = r#"{"status":"error","sqlstate":"57P01","error":"session timeout: absolute lifetime exceeded"}"#;
                let resp_len = (msg.len() as u32).to_be_bytes();
                let _ = self.stream.write_all(&resp_len).await;
                let _ = self.stream.write_all(msg.as_bytes()).await;
                return Ok(());
            }

            // Read length prefix with idle timeout.
            let mut len_buf = [0u8; 4];
            let read_result: std::io::Result<usize> = if idle_timeout_secs > 0 {
                match tokio::time::timeout(
                    Duration::from_secs(idle_timeout_secs),
                    self.stream.read_exact(&mut len_buf),
                )
                .await
                {
                    Ok(result) => result,
                    Err(_) => {
                        debug!("session idle timeout ({}s)", idle_timeout_secs);
                        return Ok(());
                    }
                }
            } else {
                self.stream.read_exact(&mut len_buf).await
            };
            match read_result {
                Ok(_) => {}
                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                    debug!("client disconnected");
                    return Ok(());
                }
                Err(e) => return Err(e.into()),
            }

            let payload_len = u32::from_be_bytes(len_buf);