taktwerk 0.6.0

//! `taktwerk record`: pull a recorder's rows and events from a running engine over OPC UA and
//! write them as run files, the same as the engine's own file sink.
//!
//! Reads from the `next_cycle` it got last, at once again while batches come back full, else
//! after a tenth of the engine's buffer (at most 1 s). A jump in the cycle numbers, or events
//! gone from the engine's log, is written to `events.csv` as a `gap` event. A new `engine_run`
//! (the engine restarted) closes the run and opens the next; the engine's `stop` event closes it
//! once the rows up to the end are written. `events.csv` is pushed to disk about once a second,
//! parts only on engine time. A lost connection is retried until SIGINT or SIGTERM, which close
//! the run cleanly.

mod pull;
mod remote;

use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::time::{Duration, Instant, SystemTime};

use anyhow::{Context as _, anyhow};
use taktwerk_core::record::wire::decode_rows;
use taktwerk_core::record::{Event, EventKind, FORMAT, Manifest};
use taktwerk_record::RunWriter;
use tokio::signal::unix::{SignalKind, signal};
use tokio::sync::watch;
use tracing::{info, warn};

use crate::ua::Conn;
use pull::{Summary, Verdict};
use remote::Remote;

/// Pause before reconnecting.
const RETRY: Duration = Duration::from_secs(1);
/// How often `events.csv` is pushed to disk, wall time; parts flush on engine time only.
const EVENTS_FLUSH: Duration = Duration::from_secs(1);
/// How long a disconnect may take.
const CLOSE_TIMEOUT: Duration = Duration::from_secs(2);

/// Arguments of `taktwerk record`.
#[derive(Debug, clap::Args)]
pub struct Args {
    /// The engine's server, e.g. `opc.tcp://192.168.0.10:4840`.
    pub endpoint: String,
    /// Directory of the runs: `<dir>/<project>/<recorder>/<run-id>/`.
    #[arg(long)]
    pub dir: PathBuf,
    /// Recorder id; may be left out when the engine has only one.
    #[arg(long)]
    pub recorder: Option<String>,
    /// Namespace URI of the signal nodes.
    #[arg(long, default_value = "urn:taktwerk")]
    pub namespace: String,
    /// The engine's `system_prefix`.
    #[arg(long, default_value = "taktwerk")]
    pub prefix: String,
}

/// Record until SIGINT or SIGTERM; prints a summary.
///
/// # Errors
/// The runtime cannot start, the recorder cannot be found or files cannot be written.
pub fn run(args: &Args) -> anyhow::Result<ExitCode> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("cannot start the I/O runtime")?;
    rt.block_on(record(args))
}

async fn record(args: &Args) -> anyhow::Result<ExitCode> {
    let (stop_tx, stop) = watch::channel(false);
    let mut sigint = signal(SignalKind::interrupt())?;
    let mut sigterm = signal(SignalKind::terminate())?;
    tokio::spawn(async move {
        tokio::select! {
            _ = sigint.recv() => info!("SIGINT received, closing the run"),
            _ = sigterm.recv() => info!("SIGTERM received, closing the run"),
        }
        let _ = stop_tx.send(true);
    });
    let mut puller = Puller {
        args,
        stop,
        state: None,
        summary: Summary::default(),
        flushed: Instant::now(),
    };
    let result = puller.serve().await;
    let closed = puller.close_run();
    println!("{}", puller.summary);
    result?;
    closed?;
    Ok(ExitCode::SUCCESS)
}

/// Why a session ended.
enum End {
    /// SIGINT or SIGTERM.
    Stopped,
    /// The connection failed; retry.
    Link(String),
}

/// One engine run being recorded.
struct State {
    manifest: Manifest,
    rows: pull::Rows,
    events: pull::Events,
    /// Opened at the first row.
    writer: Option<RunWriter>,
    /// The engine's `stop` event was pulled; the rows up to the end follow.
    stop_seen: bool,
    /// The run is closed; nothing more is written until the engine restarts.
    ended: bool,
}

impl State {
    fn new(manifest: Manifest) -> Self {
        Self {
            rows: pull::Rows::new(manifest.every),
            events: pull::Events::default(),
            manifest,
            writer: None,
            stop_seen: false,
            ended: false,
        }
    }
}

struct Puller<'a> {
    args: &'a Args,
    stop: watch::Receiver<bool>,
    state: Option<State>,
    summary: Summary,
    /// When `events.csv` was last pushed to disk.
    flushed: Instant,
}

/// `f`, or `None` once a stop is signalled.
async fn or_stop<T>(stop: &mut watch::Receiver<bool>, f: impl Future<Output = T>) -> Option<T> {
    if *stop.borrow() {
        return None;
    }
    tokio::select! {
        v = f => Some(v),
        _ = stop.wait_for(|s| *s) => None,
    }
}

impl Puller<'_> {
    /// Connect, record, reconnect, until stopped or a fatal error.
    async fn serve(&mut self) -> anyhow::Result<()> {
        let endpoint = self.args.endpoint.clone();
        let mut announced = false;
        loop {
            let Some(conn) =
                or_stop(&mut self.stop, Conn::open(&endpoint, "taktwerk-record")).await
            else {
                return Ok(());
            };
            let end = match conn {
                Ok(conn) => {
                    let opened = Remote::open(
                        conn,
                        &self.args.namespace,
                        &self.args.prefix,
                        self.args.recorder.as_deref(),
                    )
                    .await;
                    match opened {
                        Ok(remote) => {
                            info!(%endpoint, "connected");
                            announced = false;
                            let end = self.session(&remote).await;
                            close(remote.into_conn()).await;
                            end?
                        }
                        Err((conn, e, fatal)) => {
                            close(*conn).await;
                            if fatal {
                                return Err(anyhow!(e));
                            }
                            End::Link(e)
                        }
                    }
                }
                Err(e) => End::Link(e),
            };
            match end {
                End::Stopped => return Ok(()),
                End::Link(e) => {
                    if announced {
                        tracing::debug!("{e}; retrying");
                    } else {
                        warn!("{e}; retrying");
                        announced = true;
                    }
                }
            }
            if or_stop(&mut self.stop, tokio::time::sleep(RETRY))
                .await
                .is_none()
            {
                return Ok(());
            }
        }
    }

    /// Record over one connection.
    async fn session(&mut self, remote: &Remote) -> anyhow::Result<End> {
        let Some(manifest) = self.manifest(remote).await? else {
            return Ok(End::Stopped);
        };
        let manifest = match manifest {
            Ok(m) => m,
            Err(e) => return Ok(End::Link(e)),
        };
        self.adopt(manifest)?;
        loop {
            match self.pull_rows(remote).await? {
                Step::Stopped => return Ok(End::Stopped),
                Step::Link(e) => return Ok(End::Link(e)),
                Step::Restarted => {
                    info!("the engine restarted");
                    let Some(manifest) = self.manifest(remote).await? else {
                        return Ok(End::Stopped);
                    };
                    match manifest {
                        Ok(m) => self.adopt(m)?,
                        Err(e) => return Ok(End::Link(e)),
                    }
                }
                Step::Pulled { more } => {
                    let finishing = self.state.as_ref().is_some_and(|s| s.stop_seen && !s.ended);
                    if finishing {
                        // These rows were read after the stop event: the ring is final.
                        if !more {
                            self.end_run()?;
                        }
                        continue;
                    }
                    match self.pull_events(remote).await? {
                        Step::Stopped => return Ok(End::Stopped),
                        Step::Link(e) => return Ok(End::Link(e)),
                        Step::Restarted | Step::Pulled { .. } => {}
                    }
                    self.flush_events()?;
                    let stop_seen = self.state.as_ref().is_some_and(|s| s.stop_seen && !s.ended);
                    if !more && !stop_seen {
                        let period = pull::poll_period(
                            self.state.as_ref().map_or(10.0, |s| s.manifest.buffer_s),
                        );
                        if or_stop(&mut self.stop, tokio::time::sleep(period))
                            .await
                            .is_none()
                        {
                            return Ok(End::Stopped);
                        }
                    }
                }
            }
            if remote.ended() {
                return Ok(End::Link("connection lost".to_owned()));
            }
        }
    }

    /// The remote manifest; `None` when stopped.
    async fn manifest(
        &mut self,
        remote: &Remote,
    ) -> anyhow::Result<Option<Result<Manifest, String>>> {
        let Some(m) = or_stop(&mut self.stop, remote.manifest()).await else {
            return Ok(None);
        };
        if let Ok(m) = &m
            && m.format != FORMAT
        {
            return Err(anyhow!(
                "the engine speaks recorder format {}, this taktwerk {FORMAT}",
                m.format
            ));
        }
        Ok(Some(m))
    }

    /// Continue the current run when `manifest` is of the same engine run, else close it and
    /// start the next.
    fn adopt(&mut self, manifest: Manifest) -> anyhow::Result<()> {
        if self
            .state
            .as_ref()
            .is_some_and(|s| s.manifest.engine_run == manifest.engine_run)
        {
            return Ok(());
        }
        self.close_run()?;
        info!(
            project = %manifest.project,
            recorder = %manifest.recorder,
            engine_run = %manifest.engine_run,
            "recording"
        );
        self.state = Some(State::new(manifest));
        Ok(())
    }

    /// Push `events.csv` to disk when [`EVENTS_FLUSH`] has passed.
    fn flush_events(&mut self) -> anyhow::Result<()> {
        if self.flushed.elapsed() < EVENTS_FLUSH {
            return Ok(());
        }
        self.flushed = Instant::now();
        if let Some(w) = self.state.as_mut().and_then(|s| s.writer.as_mut()) {
            w.flush_events()
                .with_context(|| format!("flush {}", w.dir().display()))?;
        }
        Ok(())
    }

    /// Close the open run, if any.
    fn close_run(&mut self) -> anyhow::Result<()> {
        let Some(state) = self.state.take() else {
            return Ok(());
        };
        close_writer(state.writer, &mut self.summary)
    }

    /// The engine stopped and every row is written: close the run, keep the state so a
    /// reconnect to the same engine run writes nothing more.
    fn end_run(&mut self) -> anyhow::Result<()> {
        let Some(state) = self.state.as_mut() else {
            return Ok(());
        };
        state.ended = true;
        info!("the engine stopped");
        close_writer(state.writer.take(), &mut self.summary)
    }

    async fn pull_rows(&mut self, remote: &Remote) -> anyhow::Result<Step> {
        let (from, width) = match &self.state {
            Some(s) => (s.rows.from(), s.manifest.width()),
            None => return Ok(Step::Link("no manifest".to_owned())),
        };
        let asked = pull::rows_per_call(width);
        let Some(batch) = or_stop(&mut self.stop, remote.read_rows(from, asked)).await else {
            return Ok(Step::Stopped);
        };
        let batch = match batch {
            Ok(b) => b,
            Err(e) => return Ok(Step::Link(e)),
        };
        let Some(state) = self.state.as_mut() else {
            return Ok(Step::Link("no manifest".to_owned()));
        };
        if batch.engine_run != state.manifest.engine_run {
            return Ok(Step::Restarted);
        }
        let mut failed = None;
        let summary = &mut self.summary;
        let (root, source) = (&self.args.dir, &self.args.endpoint);
        let n = decode_rows(&batch.rows, width, |cycle, words| {
            if failed.is_some() {
                return;
            }
            if let Err(e) = write_row(state, summary, root, source, cycle, words) {
                failed = Some(e);
            }
        })
        .map_err(|e| anyhow!("read_rows: {e}"))?;
        if let Some(e) = failed {
            return Err(e);
        }
        state.rows.advance(batch.next_cycle);
        tracing::debug!("pulled {n} rows, next cycle {}", state.rows.from());
        Ok(Step::Pulled {
            more: pull::more(n, asked),
        })
    }

    async fn pull_events(&mut self, remote: &Remote) -> anyhow::Result<Step> {
        loop {
            let from = match &self.state {
                Some(s) if s.writer.is_some() => s.events.from(),
                _ => return Ok(Step::Pulled { more: false }),
            };
            let read = remote.read_events(from, pull::MAX_EVENTS);
            let Some(batch) = or_stop(&mut self.stop, read).await else {
                return Ok(Step::Stopped);
            };
            let batch = match batch {
                Ok(b) => b,
                Err(e) => return Ok(Step::Link(e)),
            };
            let Some(state) = self.state.as_mut() else {
                return Ok(Step::Pulled { more: false });
            };
            if batch.engine_run != state.manifest.engine_run {
                return Ok(Step::Restarted);
            }
            let Some(writer) = state.writer.as_mut() else {
                return Ok(Step::Pulled { more: false });
            };
            if let Some((first, last)) = state.events.read(batch.oldest_seq, batch.next_seq) {
                let cycle = batch.events.first().map_or(state.rows.from(), |e| e.cycle);
                let detail = format!("events {first}..={last} lost");
                warn!("{detail}");
                writer.write_event(&Event {
                    seq: 0,
                    cycle,
                    kind: EventKind::Gap,
                    source: "events".to_owned(),
                    detail,
                })?;
                self.summary.gaps += 1;
            }
            for e in &batch.events {
                writer.write_event(e)?;
                state.stop_seen |= e.kind == EventKind::Stop;
            }
            if !pull::more(batch.events.len(), pull::MAX_EVENTS) {
                return Ok(Step::Pulled { more: false });
            }
        }
    }
}

/// What one read did.
enum Step {
    Stopped,
    Link(String),
    Restarted,
    Pulled { more: bool },
}

/// Write the row of `cycle`: open the run at the first, log a gap before a jump.
fn write_row(
    state: &mut State,
    summary: &mut Summary,
    root: &Path,
    source: &str,
    cycle: u64,
    words: &[u64],
) -> anyhow::Result<()> {
    let verdict = state.rows.row(cycle);
    if verdict == Verdict::Skip || state.ended {
        return Ok(());
    }
    if state.writer.is_none() {
        let writer = RunWriter::create(
            root,
            state.manifest.clone(),
            source,
            cycle,
            SystemTime::now(),
        )
        .with_context(|| format!("open a run under {}", root.display()))?;
        info!(run = %writer.dir().display(), "run opened");
        state.writer = Some(writer);
        summary.runs += 1;
    }
    let Some(writer) = state.writer.as_mut() else {
        return Ok(());
    };
    if let Verdict::AfterGap(lost) = verdict {
        let detail = lost.detail();
        warn!(recorder = %state.manifest.recorder, "{detail}");
        writer.write_event(&Event {
            seq: 0,
            cycle: lost.first,
            kind: EventKind::Gap,
            source: state.manifest.recorder.clone(),
            detail,
        })?;
        summary.gaps += 1;
    }
    writer.write_row(cycle, words)?;
    summary.rows += 1;
    Ok(())
}

/// Close `writer`, if any, and count its parts.
fn close_writer(writer: Option<RunWriter>, summary: &mut Summary) -> anyhow::Result<()> {
    if let Some(writer) = writer {
        let dir = writer.dir().to_owned();
        let stats = writer
            .close(SystemTime::now())
            .with_context(|| format!("close {}", dir.display()))?;
        summary.parts += stats.part;
        info!(run = %dir.display(), rows = stats.rows, "run closed");
    }
    Ok(())
}

async fn close(conn: Conn) {
    let _ = tokio::time::timeout(CLOSE_TIMEOUT, conn.close()).await;
}