matroskin 0.1.0

//! Autonomous actor and its single-owner lifecycle state machine.

pub mod codec;
mod connect;
pub mod state;

pub use state::{ActorStatus, CloseReason, ReconnectPolicy, ReconnectReason};

use bytes::Bytes;
use futures_util::{SinkExt, StreamExt};
use tokio::{
    io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
    net::TcpStream,
    sync::{mpsc, oneshot, watch},
    time,
};
use tokio_util::codec::Framed;
use tracing::instrument;

use self::{
    codec::{WhatsMinerCodec, WhatsMinerFrame},
    connect::{IO_TIMEOUT, connect_and_get_salt, read_command_response_frame},
    state::{ActorState, LifecycleIntent, PendingPlan, Session},
};
use crate::{
    account::Account,
    auth_data::AuthData,
    command::{
        Command, get_log_download::GetLogDownload,
        set_system_update_firmware::SetSystemUpdateFirmware,
    },
    error::{Error, Result},
    password::Password,
    permission::PermissionLevel,
    redacted::Redacted,
    response::{Envelope, parse_response},
};

const HEARTBEAT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60);

type Encode = Box<dyn FnOnce(Option<AuthData>) -> Result<Redacted<Vec<u8>>> + Send>;

struct JsonMessage {
    name: &'static str,
    requires_auth: bool,
    permission: PermissionLevel,
    encode: Encode,
    generation: u64,
    response: oneshot::Sender<Result<Bytes>>,
    validated: oneshot::Receiver<bool>,
}

struct DownloadMessage {
    destination: Box<dyn AsyncWrite + Unpin + Send>,
    generation: u64,
    result: oneshot::Sender<Result<()>>,
}

struct UploadMessage {
    source: Box<dyn AsyncRead + Unpin + Send>,
    len: u64,
    generation: u64,
    result: oneshot::Sender<Result<()>>,
}

/// An independently clonable connection configuration and its actor task.
///
/// Construction and cloning require an active Tokio runtime. Use `Arc<Actor>`
/// to share one running actor; `clone()` creates a new actor in `Created`.
pub struct Actor {
    username: Account,
    password: Password,
    addr: String,
    reconnect_policy: ReconnectPolicy,
    status_rx: watch::Receiver<ActorStatus>,
    generation_rx: watch::Receiver<u64>,
    lifecycle_tx: mpsc::Sender<LifecycleIntent>,
    json_tx: mpsc::Sender<JsonMessage>,
    download_tx: mpsc::Sender<DownloadMessage>,
    upload_tx: mpsc::Sender<UploadMessage>,
}

impl Clone for Actor {
    fn clone(&self) -> Self {
        Self::with_reconnect_policy(
            self.addr.clone(),
            self.username,
            self.password.clone(),
            self.reconnect_policy,
        )
    }
}

impl Actor {
    /// Create an actor in `Created`, without network I/O or automatic reconnect.
    pub fn new(
        addr: impl Into<String>,
        username: impl Into<Account>,
        password: impl Into<Password>,
    ) -> Self {
        Self::with_reconnect_policy(addr, username, password, ReconnectPolicy::None)
    }

    /// Create an independent actor with an explicit reconnect policy.
    pub fn with_reconnect_policy(
        addr: impl Into<String>,
        username: impl Into<Account>,
        password: impl Into<Password>,
        reconnect_policy: ReconnectPolicy,
    ) -> Self {
        let addr = addr.into();
        let username = username.into();
        let password = password.into();
        let (status_tx, status_rx) = watch::channel(ActorStatus::Created);
        let (generation_tx, generation_rx) = watch::channel(0);
        let (lifecycle_tx, lifecycle_rx) = mpsc::channel(PendingPlan::CAPACITY);
        let (json_tx, json_rx) = mpsc::channel(1);
        let (download_tx, download_rx) = mpsc::channel(1);
        let (upload_tx, upload_rx) = mpsc::channel(1);
        tokio::spawn(actor_loop(
            addr.clone(),
            username,
            password.clone(),
            reconnect_policy,
            status_tx,
            generation_tx,
            lifecycle_rx,
            json_rx,
            download_rx,
            upload_rx,
        ));
        Self {
            username,
            password,
            addr,
            reconnect_policy,
            status_rx,
            generation_rx,
            lifecycle_tx,
            json_tx,
            download_tx,
            upload_tx,
        }
    }

    /// Current lifecycle projection, without transport or secrets.
    pub fn status(&self) -> ActorStatus {
        self.status_rx.borrow().clone()
    }

    /// Subscribe to current-value lifecycle status, not an event history.
    pub fn watch_status(&self) -> watch::Receiver<ActorStatus> {
        self.status_rx.clone()
    }

    /// Explicit target address, including port.
    pub fn addr(&self) -> &str {
        &self.addr
    }

    /// Configured account used for the handshake and every write operation.
    pub fn username(&self) -> Account {
        self.username
    }

    /// Configured automatic reconnection policy.
    pub fn reconnect_policy(&self) -> ReconnectPolicy {
        self.reconnect_policy
    }

    /// Submit a connect intent without awaiting network completion.
    pub fn connect(&self) -> Result<()> {
        self.lifecycle_intent(LifecycleIntent::Connect)
    }

    /// Submit a shutdown intent; active streaming consumes and ignores it.
    pub fn shutdown(&self) -> Result<()> {
        self.lifecycle_intent(LifecycleIntent::Shutdown)
    }

    fn lifecycle_intent(&self, intent: LifecycleIntent) -> Result<()> {
        self.lifecycle_tx
            .try_send(intent)
            .map_err(|error| match error {
                mpsc::error::TrySendError::Full(_) => Error::LifecycleQueueFull,
                mpsc::error::TrySendError::Closed(_) => Error::ActorClosed,
            })
    }

    /// Submit an owned command, waiting for the current session and mailbox capacity.
    ///
    /// Closing invalidates queued work. Cancelling this future does not undo
    /// network I/O; an unconfirmed response invalidates the session.
    #[instrument(level = "trace", skip_all, fields(command = C::CMD_NAME))]
    pub async fn send<C>(&self, command: C) -> Result<crate::response::Response<C::Response>>
    where
        C: Command + Send + 'static,
    {
        let (permit, generation) = reserve(&self.json_tx, &self.generation_rx).await?;
        let (response, receive) = oneshot::channel();
        let (validate, validated) = oneshot::channel();
        permit.send(JsonMessage {
            name: C::CMD_NAME,
            requires_auth: C::LEVEL.requires_auth(),
            permission: C::MINIMAL_PERMISSIONS,
            encode: Box::new(move |auth| command.encode(auth)),
            generation,
            response,
            validated,
        });
        let raw = receive.await.map_err(|_| Error::ActorClosed)??;
        let result = C::parse_response(&raw);
        let synchronized = matches!(&result, Ok(_) | Err(Error::Device { .. }));
        let _ = validate.send(synchronized);
        result
    }

    /// Download exactly the declared archive bytes to an owned asynchronous writer.
    pub async fn download_log<W>(&self, destination: W) -> Result<()>
    where
        W: AsyncWrite + Unpin + Send + 'static,
    {
        let (permit, generation) = reserve(&self.download_tx, &self.generation_rx).await?;
        let (result, receive) = oneshot::channel();
        permit.send(DownloadMessage {
            destination: Box::new(destination),
            generation,
            result,
        });
        receive.await.map_err(|_| Error::ActorClosed)?
    }

    /// Upload exactly `len` bytes; the source need not reach EOF after those bytes.
    /// Successful confirmation discards the old session because the device reboots.
    pub async fn upload_firmware<R>(&self, source: R, len: u64) -> Result<()>
    where
        R: AsyncRead + Unpin + Send + 'static,
    {
        let (permit, generation) = reserve(&self.upload_tx, &self.generation_rx).await?;
        let (result, receive) = oneshot::channel();
        permit.send(UploadMessage {
            source: Box::new(source),
            len,
            generation,
            result,
        });
        receive.await.map_err(|_| Error::ActorClosed)?
    }
}

// Closing invalidates outstanding permits as well as capacity waiters. The
// generation stamped on a message also covers a send racing with drain/reopen.
async fn reserve<'a, T>(
    sender: &'a mpsc::Sender<T>,
    generation: &watch::Receiver<u64>,
) -> Result<(mpsc::Permit<'a, T>, u64)> {
    let mut changes = generation.clone();
    let current = *changes.borrow_and_update();
    let permit = tokio::select! {
        result = sender.reserve() => result.map_err(|_| Error::ActorClosed)?,
        _ = changes.changed() => return Err(Error::ActorClosed),
    };
    if *changes.borrow() != current {
        return Err(Error::ActorClosed);
    }
    Ok((permit, current))
}

struct ExitStatus<'a>(&'a watch::Sender<ActorStatus>);

impl Drop for ExitStatus<'_> {
    fn drop(&mut self) {
        self.0.send_if_modified(|status| {
            if status.is_closed() {
                false
            } else {
                *status = ActorStatus::Closed {
                    reason: CloseReason::Disconnected,
                };
                true
            }
        });
    }
}

#[allow(clippy::too_many_arguments)]
async fn actor_loop(
    addr: String,
    account: Account,
    password: Password,
    policy: ReconnectPolicy,
    status: watch::Sender<ActorStatus>,
    generation: watch::Sender<u64>,
    mut lifecycle: mpsc::Receiver<LifecycleIntent>,
    mut json: mpsc::Receiver<JsonMessage>,
    mut downloads: mpsc::Receiver<DownloadMessage>,
    mut uploads: mpsc::Receiver<UploadMessage>,
) {
    let _exit = ExitStatus(&status);
    let mut state = ActorState::Created;
    let mut plan = PendingPlan::default();
    let mut attempt = 0usize;
    let mut retry_deadline: Option<Option<time::Instant>> = None;
    let mut last_activity = time::Instant::now();
    let mut active_download: Option<DownloadMessage> = None;
    let mut active_upload: Option<UploadMessage> = None;

    loop {
        sync_actor_status(&status, &state);
        if matches!(
            state,
            ActorState::Created
                | ActorState::Closed { .. }
                | ActorState::Connected(_)
                | ActorState::Reconnecting { .. }
        ) {
            plan.drain_from_receiver(&mut lifecycle, &state);
            if let Some(intent) = plan.pop() {
                state = match (intent, state) {
                    (LifecycleIntent::Connect, ActorState::Created | ActorState::Closed { .. }) => {
                        attempt = 0;
                        retry_deadline = None;
                        ActorState::Connecting
                    }
                    (LifecycleIntent::Shutdown, ActorState::Connected(session)) => {
                        ActorState::Disconnecting(session)
                    }
                    (LifecycleIntent::Shutdown, _) => {
                        close_operations(&generation, &mut json, &mut downloads, &mut uploads);
                        retry_deadline = None;
                        ActorState::Closed {
                            reason: CloseReason::Shutdown,
                        }
                    }
                    (LifecycleIntent::Connect, current) => current,
                };
                continue;
            }
        }
        state = match state {
            ActorState::Created => match lifecycle.recv().await {
                Some(intent) => {
                    plan.push(intent, &ActorState::Created);
                    ActorState::Created
                }
                None => return,
            },
            ActorState::Closed { reason } => {
                // Intents already queued by connect() take precedence over work
                // submitted immediately after it. Connected arbitration is unbiased.
                tokio::select! {
                    biased;
                    intent = lifecycle.recv() => match intent {
                        Some(intent) => plan.push(intent, &ActorState::Closed { reason: reason.clone() }),
                        None => return,
                    },
                    msg = json.recv() => match msg {
                        Some(msg) => { let _ = msg.response.send(Err(Error::ActorClosed)); }
                        None => return,
                    },
                    msg = downloads.recv() => match msg {
                        Some(msg) => { let _ = msg.result.send(Err(Error::ActorClosed)); }
                        None => return,
                    },
                    msg = uploads.recv() => match msg {
                        Some(msg) => { let _ = msg.result.send(Err(Error::ActorClosed)); }
                        None => return,
                    },
                }
                ActorState::Closed { reason }
            }
            ActorState::Connecting => {
                let connected = connect_and_get_salt(&addr, account).await;
                plan.drain_from_receiver(&mut lifecycle, &ActorState::Connecting);
                retry_deadline = None;
                match connected {
                    Ok((transport, salt, permissions)) => {
                        attempt = 0;
                        last_activity = time::Instant::now();
                        ActorState::Connected(Session {
                            transport,
                            salt,
                            permissions,
                        })
                    }
                    Err(error) => reconnect_or_close(
                        &policy,
                        attempt.saturating_add(1),
                        (&error).into(),
                        &generation,
                        &mut json,
                        &mut downloads,
                        &mut uploads,
                    ),
                }
            }
            ActorState::Reconnecting {
                attempt: next_attempt,
                reason,
            } => {
                let delay = policy
                    .delay_for_attempt(next_attempt)
                    .expect("reconnect state requires an available attempt");
                let deadline =
                    *retry_deadline.get_or_insert_with(|| time::Instant::now().checked_add(delay));
                tokio::select! {
                    _ = wait_deadline(deadline) => {
                        attempt = next_attempt;
                        retry_deadline = None;
                        ActorState::Connecting
                    }
                    intent = lifecycle.recv() => match intent {
                        Some(intent) => {
                            let current = ActorState::Reconnecting { attempt: next_attempt, reason };
                            plan.push(intent, &current);
                            current
                        }
                        None => return,
                    },
                }
            }
            ActorState::Disconnecting(session) => {
                close_operations(&generation, &mut json, &mut downloads, &mut uploads);
                drop(session);
                retry_deadline = None;
                ActorState::Closed {
                    reason: CloseReason::Shutdown,
                }
            }
            ActorState::Downloading(session) => {
                let msg = active_download
                    .take()
                    .expect("download message accompanies downloading state");
                let operation = download_session(session, msg.destination);
                let outcome = stream_to_boundary(operation, &mut lifecycle).await;
                let Some(outcome) = outcome else {
                    return;
                };
                last_activity = time::Instant::now();
                match outcome {
                    StreamOutcome::Keep(session, result) => {
                        let _ = msg.result.send(result);
                        ActorState::Connected(session)
                    }
                    StreamOutcome::Lost(reason, error) => {
                        let _ = msg.result.send(Err(error));
                        reconnect_or_close(
                            &policy,
                            1,
                            reason,
                            &generation,
                            &mut json,
                            &mut downloads,
                            &mut uploads,
                        )
                    }
                }
            }
            ActorState::Uploading(session) => {
                let msg = active_upload
                    .take()
                    .expect("upload message accompanies uploading state");
                let operation = upload_session(session, msg.source, msg.len, account, &password);
                let outcome = stream_to_boundary(operation, &mut lifecycle).await;
                let Some(outcome) = outcome else {
                    return;
                };
                last_activity = time::Instant::now();
                match outcome {
                    StreamOutcome::Keep(session, result) => {
                        let succeeded = result.is_ok();
                        let _ = msg.result.send(result);
                        if succeeded {
                            drop(session);
                            reconnect_or_close(
                                &policy,
                                1,
                                ReconnectReason::FirmwareUpdate,
                                &generation,
                                &mut json,
                                &mut downloads,
                                &mut uploads,
                            )
                        } else {
                            ActorState::Connected(session)
                        }
                    }
                    StreamOutcome::Lost(reason, error) => {
                        let _ = msg.result.send(Err(error));
                        reconnect_or_close(
                            &policy,
                            1,
                            reason,
                            &generation,
                            &mut json,
                            &mut downloads,
                            &mut uploads,
                        )
                    }
                }
            }
            ActorState::Connected(mut session) => {
                tokio::select! {
                    intent = lifecycle.recv() => match intent {
                        Some(intent) => {
                            let current = ActorState::Connected(session);
                            plan.push(intent, &current);
                            current
                        }
                        None => return,
                    },
                    message = json.recv() => match message {
                        Some(msg) => {
                            if msg.generation != *generation.borrow() {
                                let _ = msg.response.send(Err(Error::ActorClosed));
                                ActorState::Connected(session)
                            } else {
                                match handle_json(&mut session, msg, account, &password).await {
                                    Ok(activity) => {
                                        if activity { last_activity = time::Instant::now(); }
                                        ActorState::Connected(session)
                                    }
                                    Err(reason) => {
                                        drop(session);
                                        reconnect_or_close(&policy, 1, reason, &generation, &mut json, &mut downloads, &mut uploads)
                                    }
                                }
                            }
                        }
                        None => return,
                    },
                    message = downloads.recv() => match message {
                        Some(msg) => {
                            let rejected = if msg.generation != *generation.borrow() { Some(Error::ActorClosed) }
                                else { permission_error(&session, GetLogDownload::CMD_NAME, GetLogDownload::MINIMAL_PERMISSIONS) };
                            if let Some(error) = rejected {
                                let _ = msg.result.send(Err(error));
                                ActorState::Connected(session)
                            } else {
                                active_download = Some(msg);
                                ActorState::Downloading(session)
                            }
                        }
                        None => return,
                    },
                    message = uploads.recv() => match message {
                        Some(msg) => {
                            let rejected = if msg.generation != *generation.borrow() { Some(Error::ActorClosed) }
                                else { permission_error(&session, SetSystemUpdateFirmware::CMD_NAME, SetSystemUpdateFirmware::MINIMAL_PERMISSIONS) };
                            if let Some(error) = rejected {
                                let _ = msg.result.send(Err(error));
                                ActorState::Connected(session)
                            } else {
                                active_upload = Some(msg);
                                ActorState::Uploading(session)
                            }
                        }
                        None => return,
                    },
                    frame = session.transport.next() => {
                        let result = match frame {
                            Some(Ok(WhatsMinerFrame::Heartbeat)) => Ok(()),
                            Some(Ok(WhatsMinerFrame::Data(bytes))) => {
                                match serde_json::from_slice::<Envelope<'_>>(&bytes) {
                                    Ok(envelope) if envelope.desc == "get.miner.report" => Ok(()),
                                    Ok(_) => Err(Error::Protocol("unexpected unsolicited response".into())),
                                    Err(error) => Err(Error::Json(error)),
                                }
                            }
                            Some(Err(error)) => Err(error),
                            None => Err(Error::ConnectionClosed),
                        };
                        match result {
                            Ok(()) => { last_activity = time::Instant::now(); ActorState::Connected(session) }
                            Err(error) => {
                                drop(session);
                                reconnect_or_close(&policy, 1, (&error).into(), &generation, &mut json, &mut downloads, &mut uploads)
                            }
                        }
                    },
                    _ = time::sleep_until(last_activity + HEARTBEAT_INTERVAL) => {
                        match io_step(session.transport.send(WhatsMinerFrame::Heartbeat)).await {
                            Ok(()) => { last_activity = time::Instant::now(); ActorState::Connected(session) }
                            Err(error) => {
                                drop(session);
                                reconnect_or_close(&policy, 1, (&error).into(), &generation, &mut json, &mut downloads, &mut uploads)
                            }
                        }
                    },
                }
            }
        };
    }
}

async fn wait_deadline(deadline: Option<time::Instant>) {
    match deadline {
        Some(deadline) => time::sleep_until(deadline).await,
        None => std::future::pending().await,
    }
}

async fn stream_to_boundary<F: Future>(
    operation: F,
    lifecycle: &mut mpsc::Receiver<LifecycleIntent>,
) -> Option<F::Output> {
    tokio::pin!(operation);
    loop {
        tokio::select! {
            result = &mut operation => {
                // Consume intents queued in the same poll as stream completion.
                for _ in 0..PendingPlan::CAPACITY { if lifecycle.try_recv().is_err() { break; } }
                return Some(result);
            }
            intent = lifecycle.recv() => { intent?; },
        }
    }
}

fn sync_actor_status(tx: &watch::Sender<ActorStatus>, state: &ActorState) {
    let next = state.status();
    tx.send_if_modified(|current| {
        if *current == next {
            false
        } else {
            *current = next;
            true
        }
    });
}

// fn reason_for_error(error: &Error) -> ReconnectReason {}

#[allow(clippy::too_many_arguments)]
fn reconnect_or_close(
    policy: &ReconnectPolicy,
    attempt: usize,
    reason: ReconnectReason,
    generation: &watch::Sender<u64>,
    json: &mut mpsc::Receiver<JsonMessage>,
    downloads: &mut mpsc::Receiver<DownloadMessage>,
    uploads: &mut mpsc::Receiver<UploadMessage>,
) -> ActorState {
    if policy.delay_for_attempt(attempt).is_some() {
        ActorState::Reconnecting { attempt, reason }
    } else {
        close_operations(generation, json, downloads, uploads);
        let reason = if attempt > 1 {
            CloseReason::ReconnectFailed {
                attempts: attempt - 1,
            }
        } else {
            CloseReason::Disconnected
        };
        ActorState::Closed { reason }
    }
}

fn close_operations(
    generation: &watch::Sender<u64>,
    json: &mut mpsc::Receiver<JsonMessage>,
    downloads: &mut mpsc::Receiver<DownloadMessage>,
    uploads: &mut mpsc::Receiver<UploadMessage>,
) {
    generation.send_modify(|generation| *generation = generation.wrapping_add(1));
    while let Ok(msg) = json.try_recv() {
        let _ = msg.response.send(Err(Error::ActorClosed));
    }
    while let Ok(msg) = downloads.try_recv() {
        let _ = msg.result.send(Err(Error::ActorClosed));
    }
    while let Ok(msg) = uploads.try_recv() {
        let _ = msg.result.send(Err(Error::ActorClosed));
    }
}

fn permission_error(
    session: &Session,
    name: &'static str,
    required: PermissionLevel,
) -> Option<Error> {
    (!session.permissions.can_execute(required)).then_some(Error::NoPermission {
        command: name,
        required,
        actual: session.permissions.0,
    })
}

async fn handle_json(
    session: &mut Session,
    message: JsonMessage,
    account: Account,
    password: &Password,
) -> std::result::Result<bool, ReconnectReason> {
    let JsonMessage {
        name,
        requires_auth,
        permission,
        encode,
        response,
        validated,
        ..
    } = message;
    if let Some(error) = permission_error(session, name, permission) {
        let _ = response.send(Err(error));
        return Ok(false);
    }
    let prepared = (|| {
        let auth = if requires_auth {
            Some(AuthData::for_command_name(
                name,
                account,
                password.as_ref(),
                session.salt.as_str(),
            )?)
        } else {
            None
        };
        encode(auth)
    })();
    let mut encoded = match prepared {
        Ok(encoded) => encoded,
        Err(error) => {
            let _ = response.send(Err(error));
            return Ok(false);
        }
    };
    let payload = Bytes::from(std::mem::take(&mut encoded.0));
    let raw = match process_command(&mut session.transport, name, payload).await {
        Ok(raw) => raw,
        Err(error) => {
            let _ = response.send(Err(Error::ConnectionLost));
            return Err((&error).into());
        }
    };
    if response.send(Ok(raw)).is_err() {
        return Err(ReconnectReason::ProtocolError(
            "response receiver cancelled".into(),
        ));
    }
    // A suspended caller must not keep the transport indefinitely while the
    // actor waits for its synchronous typed-response validation.
    match time::timeout(IO_TIMEOUT, validated).await {
        Ok(Ok(true)) => Ok(true),
        Ok(_) => Err(ReconnectReason::ProtocolError(
            "invalid or unconfirmed response".into(),
        )),
        Err(_) => Err(ReconnectReason::Timeout),
    }
}

async fn io_step<T, E: Into<Error>>(
    future: impl Future<Output = std::result::Result<T, E>>,
) -> Result<T> {
    time::timeout(IO_TIMEOUT, future)
        .await
        .map_err(|_| Error::Timeout)?
        .map_err(Into::into)
}

async fn process_command(
    transport: &mut Framed<TcpStream, WhatsMinerCodec>,
    name: &'static str,
    payload: Bytes,
) -> Result<Bytes> {
    io_step(transport.send(WhatsMinerFrame::Data(payload))).await?;
    read_command_response_frame(transport, name).await
}

enum StreamOutcome {
    Keep(Session, Result<()>),
    Lost(ReconnectReason, Error),
}

impl From<Error> for StreamOutcome {
    fn from(value: Error) -> Self {
        StreamOutcome::Lost((&value).into(), Error::ConnectionLost)
    }
}

async fn download_session(
    mut session: Session,
    mut destination: Box<dyn AsyncWrite + Unpin + Send>,
) -> StreamOutcome {
    let payload = match GetLogDownload::encode_request() {
        Ok(payload) => Bytes::from(payload),
        Err(error) => return StreamOutcome::Keep(session, Err(error)),
    };
    let raw = match process_command(&mut session.transport, GetLogDownload::CMD_NAME, payload).await
    {
        Ok(raw) => raw,
        Err(error) => return error.into(),
    };
    let response =
        match parse_response::<crate::command::get_log_download::GetLogDownloadResponse>(&raw) {
            Ok(response) => response,
            Err(error @ Error::Device { .. }) => return StreamOutcome::Keep(session, Err(error)),
            Err(error) => return error.into(),
        };
    let mut remaining = response.msg.logsize;
    let mut parts = session.transport.into_parts();
    let buffered = (parts.read_buf.len() as u64).min(remaining) as usize;
    if buffered != 0 {
        let chunk = parts.read_buf.split_to(buffered);
        if let Err(error) = io_step(destination.write_all(&chunk)).await {
            return error.into();
        }
        remaining -= buffered as u64;
    }
    let mut buffer = [0; 16 * 1024];
    while remaining != 0 {
        let capacity = remaining.min(buffer.len() as u64) as usize;
        let count = match io_step(parts.io.read(&mut buffer[..capacity])).await {
            Ok(0) => return Error::ConnectionClosed.into(),
            Ok(count) => count,
            Err(error) => return error.into(),
        };
        if let Err(error) = io_step(destination.write_all(&buffer[..count])).await {
            return error.into();
        }
        remaining -= count as u64;
    }
    if let Err(error) = io_step(destination.flush()).await {
        return error.into();
    }
    session.transport = Framed::from_parts(parts);
    StreamOutcome::Keep(session, Ok(()))
}

async fn upload_session(
    mut session: Session,
    mut source: Box<dyn AsyncRead + Unpin + Send>,
    len: u64,
    account: Account,
    password: &Password,
) -> StreamOutcome {
    let size = match u32::try_from(len) {
        Ok(size) => size,
        Err(_) => {
            return StreamOutcome::Keep(
                session,
                Err(Error::ResponseTooLarge(
                    usize::try_from(len).unwrap_or(usize::MAX),
                )),
            );
        }
    };
    let payload = AuthData::for_command_name(
        SetSystemUpdateFirmware::CMD_NAME,
        account,
        password.as_ref(),
        session.salt.as_str(),
    )
    .and_then(SetSystemUpdateFirmware::encode_request);
    let payload = match payload {
        Ok(payload) => Bytes::from(payload),
        Err(error) => return StreamOutcome::Keep(session, Err(error)),
    };
    let raw = match process_command(
        &mut session.transport,
        SetSystemUpdateFirmware::CMD_NAME,
        payload,
    )
    .await
    {
        Ok(raw) => raw,
        Err(error) => return error.into(),
    };
    match parse_response::<String>(&raw) {
        Ok(response) if response.msg == "ready" => {}
        Ok(_) => {
            return Error::Protocol("unexpected firmware readiness response".into()).into();
        }
        Err(error @ Error::Device { .. }) => return StreamOutcome::Keep(session, Err(error)),
        Err(error) => return error.into(),
    }
    let mut parts = session.transport.into_parts();
    // Server-to-client traffic remains framed during an upload. Preserve
    // coalesced heartbeats/reports for the final response reader.
    if !parts.write_buf.is_empty() {
        return Error::Protocol("unflushed request before raw upload".into()).into();
    }
    if let Err(error) = io_step(parts.io.write_all(&size.to_le_bytes())).await {
        return error.into();
    }
    let mut remaining = len;
    let mut buffer = [0; 16 * 1024];
    while remaining != 0 {
        let capacity = remaining.min(buffer.len() as u64) as usize;
        let count = match io_step(source.read(&mut buffer[..capacity])).await {
            Ok(0) => {
                return Error::Protocol("firmware source shorter than declared length".into())
                    .into();
            }
            Ok(count) => count,
            Err(error) => return error.into(),
        };
        if let Err(error) = io_step(parts.io.write_all(&buffer[..count])).await {
            return error.into();
        }
        remaining -= count as u64;
    }
    if let Err(error) = io_step(parts.io.flush()).await {
        return error.into();
    }
    session.transport = Framed::from_parts(parts);
    let raw = match read_command_response_frame(
        &mut session.transport,
        SetSystemUpdateFirmware::CMD_NAME,
    )
    .await
    {
        Ok(raw) => raw,
        Err(error) => return error.into(),
    };
    match parse_response::<serde::de::IgnoredAny>(&raw) {
        Ok(_) => StreamOutcome::Keep(session, Ok(())),
        Err(error @ Error::Device { .. }) => StreamOutcome::Keep(session, Err(error)),
        Err(error) => error.into(),
    }
}