cloudpub-client 3.6.906

use crate::config::{ClientConfig, ClientOpts};
use crate::shell::SubProcess;
use anyhow::{bail, Context, Result};
use async_trait::async_trait;
use cloudpub_common::fair_channel::FairSender;
use cloudpub_common::protocol::message::Message;
use cloudpub_common::protocol::{Break, ErrorInfo, ErrorKind, ServerEndpoint};
use cloudpub_common::utils::is_tcp_port_available;
use parking_lot::RwLock;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio::time::{self, Duration, Instant};
use tracing::{debug, error};

/// ClientEndpoint intentionally has no PartialEq (see common/build.rs);
/// its deterministic prost encoding stands in for config equality.
fn encode_client_config(endpoint: &ServerEndpoint) -> Vec<u8> {
    use cloudpub_common::prost::Message as ProstMessage;
    endpoint
        .client
        .as_ref()
        .map(|c| c.encode_to_vec())
        .unwrap_or_default()
}

#[async_trait]
pub trait Plugin: Send + Sync {
    /// Name of the plugin
    fn name(&self) -> &'static str;

    /// Setup the plugin environment
    async fn setup(
        &self,
        config: &Arc<RwLock<ClientConfig>>,
        opts: &ClientOpts,
        command_rx: &mut mpsc::Receiver<Message>,
        result_tx: &mpsc::Sender<Message>,
    ) -> Result<()>;

    /// Publish a service using this plugin
    async fn publish(
        &self,
        endpoint: &ServerEndpoint,
        config: &Arc<RwLock<ClientConfig>>,
        opts: &ClientOpts,
        result_tx: &mpsc::Sender<Message>,
    ) -> Result<SubProcess>;
}

pub struct PluginHandle {
    guid: String,
    /// prost-encoded ClientEndpoint this handle was spawned for, used to
    /// recognize duplicate EndpointStart deliveries for the same config
    client_config: Vec<u8>,
    server: Arc<RwLock<Option<SubProcess>>>,
    cancel_tx: mpsc::Sender<Message>,
    task: JoinHandle<()>,
}

impl PluginHandle {
    pub fn spawn(
        plugin: Arc<dyn Plugin>,
        mut endpoint: ServerEndpoint,
        config: Arc<RwLock<ClientConfig>>,
        opts: ClientOpts,
        to_server_tx: FairSender<Message>,
    ) -> Self {
        let guid = endpoint.guid.clone();
        let client_config = encode_client_config(&endpoint);
        let guid_for_task = guid.clone();
        let server = Arc::new(RwLock::new(None));
        let server_task = server.clone();

        // Per-process cancel and event channels
        let (cancel_tx, mut cancel_rx) = mpsc::channel::<Message>(1);
        let (proc_event_tx, mut proc_event_rx) = mpsc::channel::<Message>(1024);

        let task = tokio::spawn(async move {
            // Forward per-process events to the main client event channel
            tokio::spawn({
                let mut endpoint = endpoint.clone();
                let to_server_tx = to_server_tx.clone();
                let guid_for_task = guid_for_task.clone();
                async move {
                    use tokio::time::{Duration, Instant};

                    let mut last_progress_time = Instant::now() - Duration::from_secs(1); // Allow first progress immediately

                    while let Some(mut msg) = proc_event_rx.recv().await {
                        let should_send = match &mut msg {
                            Message::Progress(progress_info) => {
                                progress_info.guid = guid_for_task.clone();

                                let now = Instant::now();
                                // Always send 0% and 100% progress messages, or throttle to 1 per second
                                if progress_info.current == 0
                                    || progress_info.current >= progress_info.total
                                    || now.duration_since(last_progress_time)
                                        >= Duration::from_secs(1)
                                {
                                    last_progress_time = now;
                                    true
                                } else {
                                    false
                                }
                            }
                            Message::Error(info) => {
                                // CLOUDPUB-1505: keep the failure reason for
                                // the trailing offline status below - the one
                                // the server stores as the publication's error.
                                endpoint.error = info.message.clone();
                                true
                            }
                            _ => true, // Always send non-progress messages
                        };

                        if should_send {
                            to_server_tx.send(msg.clone()).await.ok();
                        }
                    }

                    endpoint.status = Some("offline".to_string());
                    to_server_tx
                        .send(Message::EndpointStatus(endpoint.clone()))
                        .await
                        .ok();
                }
            });

            // Initial lifecycle status: waiting
            endpoint.status = Some("waiting".to_string());
            let _ = to_server_tx
                .send(Message::EndpointStatus(endpoint.clone()))
                .await;

            let res: Result<()> = async {
                // Long running setup: cancellable via cancel_rx
                plugin
                    .setup(&config, &opts, &mut cancel_rx, &proc_event_tx)
                    .await
                    .context("Failed to setup plugin")?;

                let server_process = plugin
                    .publish(&endpoint, &config, &opts, &proc_event_tx)
                    .await
                    .context("Failed to publish plugin service")?;

                // Wait until the port is bound
                let now = Instant::now();
                while is_tcp_port_available("127.0.0.1", server_process.port)
                    .await
                    .context("Check port availability")?
                {
                    if server_process.result.read().is_err() {
                        return Ok(()); // Error already reported by subprocess
                    }

                    if now.elapsed() > Duration::from_secs(60) {
                        bail!("{}", crate::t!("error-start-server"));
                    }
                    debug!(
                        "Waiting for server to start on port {}",
                        server_process.port
                    );
                    time::sleep(Duration::from_secs(1)).await;
                }

                *server_task.write() = Some(server_process);
                Ok(())
            }
            .await;

            match res {
                Ok(()) => {
                    endpoint.status = Some("online".into());
                    let _ = to_server_tx.send(Message::EndpointStatus(endpoint)).await;
                }
                Err(e) => {
                    error!("Error handling endpoint {}: {:#}", &guid_for_task, e);
                    // Through the forwarder, not straight to the server: the
                    // forwarder owns the trailing offline EndpointStatus, and
                    // only a message it has seen can put the reason into it
                    // (CLOUDPUB-1505). {:#} keeps the whole anyhow chain;
                    // to_string() would carry only the outermost context.
                    let _ = proc_event_tx
                        .send(Message::Error(ErrorInfo {
                            kind: ErrorKind::PublishFailed.into(),
                            message: format!("{:#}", e),
                            guid: guid_for_task,
                        }))
                        .await;
                }
            }
        });

        Self {
            guid,
            client_config,
            server,
            cancel_tx,
            task,
        }
    }

    /// True when this handle already serves exactly this endpoint
    /// configuration: a duplicate EndpointStart (the server re-sends them
    /// after every EndpointStartAll and on repeated registrations) must
    /// not bounce a running publication.
    pub fn same_config(&self, endpoint: &ServerEndpoint) -> bool {
        self.client_config == encode_client_config(endpoint)
    }

    /// The publication is being set up or its subprocess is alive. A
    /// finished task with no healthy subprocess means setup failed or the
    /// server died - a fresh spawn is warranted then.
    pub fn is_healthy(&self) -> bool {
        if !self.task.is_finished() {
            return true;
        }
        match self.server.read().as_ref() {
            Some(process) => process.result.read().is_ok(),
            None => false,
        }
    }

    pub fn port(&self) -> Option<u16> {
        self.server.read().as_ref().map(|s| s.port)
    }

    pub fn guid(&self) -> &str {
        &self.guid
    }

    pub fn send_break(&self) {
        let _ = self.cancel_tx.try_send(Message::Break(Break {
            guid: self.guid.clone(),
        }));
    }
}

impl Drop for PluginHandle {
    fn drop(&mut self) {
        // Abort main task and drop cancel_tx to notify setup to stop
        self.task.abort();
        // cancel_tx dropped here; receiver sees end-of-stream and should terminate
        let _ = self.cancel_tx.try_send(Message::Break(Break {
            guid: self.guid.clone(),
        }));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use anyhow::anyhow;
    use cloudpub_common::fair_channel::fair_channel;
    use tokio::time::timeout;

    // BUG (CLOUDPUB-1505): a publication whose plugin setup fails comes back
    // as offline with an EMPTY error, so the dashboard shows "clicked -
    // nothing happened" while the agent log alone holds the reason.
    // HYPOTHESIS: PluginHandle::spawn reports the failure twice, both times
    // without the cause reaching the status the server stores:
    //  - Message::Error carries e.to_string(), which is only the outermost
    //    context ("Failed to setup plugin"), not the anyhow chain;
    //  - the forwarder task sends the final EndpointStatus(offline) from its
    //    own endpoint clone that never learned the failure, and the server
    //    (control.rs) faithfully stores that empty error.
    // SOLUTION: format the ErrorInfo message with {:#} so the whole chain
    // travels, route it through the forwarder so the trailing offline status
    // inherits the reason, and assert the LAST status the server would see
    // is offline WITH the cause.
    // RESULT: the user sees the real reason ("404 Not Found") in the row.
    // STATUS: FIXED - see CLOUDPUB-1505
    struct SetupFails;

    #[async_trait]
    impl Plugin for SetupFails {
        fn name(&self) -> &'static str {
            "setup_fails"
        }

        async fn setup(
            &self,
            _config: &Arc<RwLock<ClientConfig>>,
            _opts: &ClientOpts,
            _command_rx: &mut mpsc::Receiver<Message>,
            _result_tx: &mpsc::Sender<Message>,
        ) -> Result<()> {
            // Two levels below the context spawn() adds - the shape of the
            // real 404 chain from the stand, minus the localized wrapping.
            Err(anyhow!("HTTP request failed with status: 404 Not Found")
                .context("web server package download failed"))
        }

        async fn publish(
            &self,
            _endpoint: &ServerEndpoint,
            _config: &Arc<RwLock<ClientConfig>>,
            _opts: &ClientOpts,
            _result_tx: &mpsc::Sender<Message>,
        ) -> Result<SubProcess> {
            unreachable!("setup failed before publish was reached")
        }
    }

    #[tokio::test]
    async fn failed_plugin_setup_reports_reason_in_offline_status() {
        let (to_server_tx, mut to_server_rx) = fair_channel::<Message>(16);
        let endpoint = ServerEndpoint {
            guid: "guid-1505".to_string(),
            ..Default::default()
        };

        let handle = PluginHandle::spawn(
            Arc::new(SetupFails),
            endpoint,
            Arc::new(RwLock::new(ClientConfig::default())),
            ClientOpts::default(),
            to_server_tx,
        );

        let mut error_message: Option<String> = None;
        let mut last_status: Option<(Option<String>, String)> = None;
        // FairReceiver has no sender tracking: recv() would wait forever once
        // the tasks are done, so drain until a quiet window says they are.
        loop {
            match timeout(Duration::from_secs(1), to_server_rx.recv()).await {
                Ok(Some(msg)) => match msg {
                    Message::Error(info) => error_message = Some(info.message),
                    Message::EndpointStatus(ep) => {
                        last_status = Some((ep.status.clone(), ep.error.clone()))
                    }
                    other => debug!("unexpected message: {:?}", other),
                },
                Ok(None) => break,
                Err(_) => break,
            }
        }

        let error_message = error_message.expect("no Message::Error reached the server channel");
        assert!(
            error_message.contains("Failed to setup plugin")
                && error_message.contains("web server package download failed")
                && error_message.contains("404 Not Found"),
            "Error message must carry the full chain, got: {}",
            error_message
        );

        let (status, status_error) =
            last_status.expect("no EndpointStatus reached the server channel");
        assert_eq!(
            status.as_deref(),
            Some("offline"),
            "expected the final offline status, got status={:?} error={}",
            status,
            status_error
        );
        assert!(
            status_error.contains("404 Not Found"),
            "the offline status must carry the failure reason, got: {}",
            status_error
        );

        drop(handle);
    }
}