chadtmiller.com/knotless
git that lives in your Atmosphere account
7
//! What the runner tells the appview (`KNOTLESS_APPVIEW`) as it happens,
//! and what it hears from it.
//!
//! A check's record says nothing of a run until it ends, so each run sends
//! the appview what its steps print as they print it, every second while
//! someone follows it there and every few otherwise
//! (`dev.knotless.putRunOutput`). Only whole lines go, each with the run's
//! secrets masked as its log's are, so no secret is ever sent split where
//! masking would miss it, and with the steps that ended and the one going.
//! What fails to send is sent again next time, up to as much as the appview
//! keeps; a run never waits on the appview, and the log is what is kept.
//!
//! The appview is told of each write in a private repository's space,
//! which Jetstream never carries, and passes that on to the runners the
//! space's owner invited (`dev.knotless.awaitSpaceWrites`): a runner hears
//! of a push there at once, and asks the space itself only now and then.
use crate::check::{self, Step};
use anyhow::{Context, Result};
use knotless::env::Ctx;
use knotless::identity::Identity;
use knotless::session;
use knotless::xrpc::{Auth, Xrpc};
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::io::{Read, Seek, SeekFrom};
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};
use tokio::sync::{Mutex, OnceCell, watch};
pub use knotless::invites::{OUTPUT, WRITES};
/// How often output is sent while someone follows the run, and while
/// no one does.
pub const WATCHED: Duration = Duration::from_secs(1);
pub const UNWATCHED: Duration = Duration::from_secs(5);
/// How long a run's last output may take to send before it is filed.
pub const LAST: Duration = Duration::from_secs(5);
/// How often a private repository's space is asked itself whether it was
/// written to while the appview passes that on, for what it missed.
pub const WOKEN_POLL: Duration = Duration::from_secs(60);
/// How often a space is asked to tell the appview of its writes: the
/// authority keeps the ask for a day.
pub const REGISTER: Duration = Duration::from_secs(12 * 60 * 60);
/// How long after the appview could not be asked for writes it is asked
/// again.
pub const RETRY: Duration = Duration::from_secs(30);
/// How long a service token is asked for, and used.
const TOKEN_FOR: Duration = Duration::from_secs(30 * 60);
const TOKEN_USED: Duration = Duration::from_secs(25 * 60);
/// The most of a line held back for its end: past it, the line is let go
/// unsent rather than sent in part.
const LINE_MAX: usize = 1_000_000;
/// The most of a log read at once: the rest waits for the next send.
const READ_MAX: u64 = 256 << 10;
/// The most text sent at once, as `dev.knotless.putRunOutput` takes it.
const PIECE: usize = 16_000;
/// The most text held while the appview does not take it, as much as it
/// keeps of a run: past it, the oldest is let go.
const UNSENT_MAX: usize = 1_000_000;
/// The appview, as this runner's account calls it.
pub struct Appview {
ctx: Ctx,
identity: Identity,
xrpc: Xrpc,
url: String,
did: OnceCell<String>,
tokens: Mutex<HashMap<&'static str, (Auth, Instant)>>,
}
/// The spaces written to since a cursor, as the appview tells them.
#[derive(Deserialize, Debug)]
pub struct Writes {
pub cursor: String,
#[serde(default)]
pub writes: Vec<Written>,
/// Set when the cursor was of no place the appview remembers.
#[serde(default)]
pub missed: bool,
}
#[derive(Deserialize, Debug)]
pub struct Written {
pub space: String,
}
impl Appview {
/// The appview `ctx` names, as `identity` calls it, if it names one.
pub fn new(ctx: &Ctx, identity: &Identity) -> Option<Arc<Appview>> {
let url = ctx.env.appview.clone()?;
Some(Arc::new(Appview {
ctx: ctx.clone(),
identity: identity.clone(),
xrpc: Xrpc::new(ctx.http.clone(), &url),
url,
did: OnceCell::new(),
tokens: Mutex::default(),
}))
}
/// Its DID, as its document says, read once.
pub async fn did(&self) -> Result<&str> {
let did = self
.did
.get_or_try_init(|| knotless::moderation::appview_did(&self.ctx, &self.url));
Ok(did.await?.as_str())
}
/// A token for calling `lxm`, asked for once in a while.
async fn auth(&self, lxm: &'static str) -> Result<Auth> {
let mut tokens = self.tokens.lock().await;
if let Some((auth, _)) = tokens.get(lxm).filter(|(_, at)| at.elapsed() < TOKEN_USED) {
return Ok(auth.clone());
}
let me = session::ensure(&self.ctx, &self.identity).await?;
let now = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)?;
let exp = (now + TOKEN_FOR).as_secs();
let did = self.did().await?;
let token = knotless::auth::lasting_token(&self.ctx, &me, did, lxm, exp).await?;
let auth = Auth::Bearer(token);
tokens.insert(lxm, (auth.clone(), Instant::now()));
Ok(auth)
}
/// Sends `body`, a run's output since it last sent. Answers whether
/// anyone follows the run.
pub async fn put_output(&self, body: &Value) -> Result<bool> {
#[derive(Deserialize)]
struct Out {
watched: bool,
}
let auth = self.auth(OUTPUT).await?;
let out: Out = self.xrpc.procedure(OUTPUT, Some(body), Some(&auth)).await?;
Ok(out.watched)
}
/// The spaces written to since `cursor`, waiting a while for one.
pub async fn await_writes(&self, cursor: Option<&str>) -> Result<Writes> {
let auth = self.auth(WRITES).await?;
let params: Vec<(&str, &str)> = cursor.map(|c| ("cursor", c)).into_iter().collect();
self.xrpc.query(WRITES, ¶ms, Some(&auth)).await
}
/// The service a space's authority tells this appview of writes at:
/// its DID, and the entry in its document.
pub async fn service(&self) -> Result<String> {
Ok(format!("{}{}", self.did().await?, knotless::space::APPVIEW))
}
}
/// A run's log as it is read: where reading it got to, and the line
/// begun but not ended.
#[derive(Default)]
pub struct Tail {
read: u64,
line: Vec<u8>,
}
impl Tail {
/// Takes in what the log has past where it was read to, and returns
/// the lines ended, masked, to send; with `ended`, the rest too. A
/// line longer than [`LINE_MAX`] is let go unsent.
pub fn take(&mut self, more: &[u8], secrets: &[String], ended: bool) -> String {
self.read += more.len() as u64;
self.line.extend_from_slice(more);
let end = match ended {
true => self.line.len(),
false => self
.line
.iter()
.rposition(|b| *b == b'\n' || *b == b'\r')
.map_or(0, |i| i + 1),
};
let whole: Vec<u8> = self.line.drain(..end).collect();
if self.line.len() > LINE_MAX {
self.line.clear();
}
String::from_utf8_lossy(&check::redact(&whole, secrets)).into_owned()
}
}
/// How a run stands as its steps go: those that ended, and the one going.
#[derive(Default, Clone, PartialEq, Debug)]
pub struct Progress {
pub steps: Vec<Step>,
pub going: Option<String>,
}
/// What a run has for the appview, held until it is taken in.
#[derive(Default)]
struct Outbox {
tail: Tail,
/// Where `text` starts in all the run sent.
at: u64,
/// What was read and not yet taken in.
text: String,
/// How the run stands now, and as the appview last took it.
now: Progress,
told: Progress,
over: bool,
/// Set when a send of this round failed: the rest waits for the next.
refused: bool,
/// Set once the last send is taken in, or fails with the run over.
done: bool,
}
impl Outbox {
/// Starts a round: `more` of the log, how the run stands, and whether
/// it is over.
fn round(&mut self, more: &[u8], secrets: &[String], now: Progress, over: bool) {
let text = self.tail.take(more, secrets, over);
self.text.push_str(&text);
let cut = ceil(&self.text, self.text.len().saturating_sub(UNSENT_MAX));
self.text.drain(..cut);
self.at += cut as u64;
(self.now, self.over, self.refused) = (now, over, false);
}
/// The next send of this round, if it has one: a piece of the text,
/// the last with how the run stands, and on the run's very last that
/// it ended. How the run stands goes every time, for an appview that
/// lost it in a restart.
fn next(&self, check: &str, started: &str) -> Option<Value> {
let due = !self.text.is_empty() || self.now != self.told || self.over;
if self.refused || self.done || !due {
return None;
}
let end = floor(&self.text, PIECE);
let whole = end == self.text.len();
let piece = &self.text[..end];
let ended = self.over && whole;
let mut body = json!({ "check": check, "startedAt": started, "offset": self.at, "text": piece, "ended": ended });
if whole {
body["steps"] = serde_json::to_value(&self.now.steps).unwrap_or_default();
body["going"] = json!(self.now.going);
}
Some(body)
}
/// That `body`, from [`Outbox::next`], was `taken` in or not.
fn answered(&mut self, body: &Value, taken: bool) {
let sent = body["text"].as_str().map_or(0, str::len) * usize::from(taken);
self.text.drain(..sent);
self.at += sent as u64;
if taken && body.get("steps").is_some() {
self.told = self.now.clone();
}
self.refused = !taken;
self.done = body["ended"] == true || (!taken && self.over);
}
}
/// Sends what a run writes to `log`, its check `uri`'s, which started at
/// `started`, to `appview` as it writes it, with how its steps go, until
/// `ended` says it did: then the rest, and that it ended.
pub async fn send(
appview: Arc<Appview>,
log: PathBuf,
uri: String,
started: String,
secrets: Vec<String>,
mut progress: watch::Receiver<Progress>,
mut ended: watch::Receiver<bool>,
) {
let mut outbox = Outbox::default();
let mut every = UNWATCHED;
loop {
// Its sender gone, as when the run failed to start, it ended too.
let over = *ended.borrow_and_update() || ended.has_changed().is_err();
let more = read_from(&log, outbox.tail.read).unwrap_or_default();
let now = progress.borrow_and_update().clone();
outbox.round(&more, &secrets, now, over);
while let Some(body) = outbox.next(&uri, &started) {
let answer = appview.put_output(&body).await;
every = match answer.as_ref().is_ok_and(|watched| *watched) {
true => WATCHED,
false => UNWATCHED,
};
outbox.answered(&body, answer.is_ok());
}
if outbox.done {
return;
}
tokio::select! {
() = tokio::time::sleep(every) => {}
_ = ended.changed() => {}
_ = progress.changed() => {}
}
}
}
/// The nearest end of a character in `text` at `at` or before it.
fn floor(text: &str, at: usize) -> usize {
let mut at = at.min(text.len());
while !text.is_char_boundary(at) {
at -= 1;
}
at
}
/// The nearest end of a character in `text` at `at` or after it.
fn ceil(text: &str, at: usize) -> usize {
let mut at = at.min(text.len());
while !text.is_char_boundary(at) {
at += 1;
}
at
}
/// What the file `path` holds past `from`, [`READ_MAX`] at most.
fn read_from(path: &PathBuf, from: u64) -> Result<Vec<u8>> {
let mut file = std::fs::File::open(path).context("the run's log")?;
file.seek(SeekFrom::Start(from))?;
let mut more = vec![];
file.take(READ_MAX).read_to_end(&mut more)?;
Ok(more)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn whole_lines_go_masked() {
let secrets = vec!["s3cret".to_string()];
let mut tail = Tail::default();
assert_eq!(tail.take(b"one\ntw", &secrets, false), "one\n");
assert_eq!(tail.take(b"o s3c", &secrets, false), "");
// A secret split as it was written goes masked once its line ends.
assert_eq!(tail.take(b"ret\rthree", &secrets, false), "two ***\r");
assert_eq!(tail.take(b"", &secrets, true), "three");
assert_eq!(tail.read, 20);
// A line too long to hold is let go.
let mut tail = Tail::default();
let long = vec![b'x'; LINE_MAX + 1];
assert_eq!(tail.take(&long, &secrets, false), "");
assert_eq!(tail.take(b"next\n", &secrets, false), "next\n");
}
#[test]
fn text_is_cut_at_whole_characters() {
assert_eq!(floor("abécd", 3), 2);
assert_eq!(floor("ab", 3), 2);
assert_eq!(ceil("abécd", 3), 4);
}
#[test]
fn what_is_not_taken_goes_again() {
let step = |name: &str| Step {
name: name.into(),
status: "success".into(),
exit_code: Some(0),
duration_ms: 1,
};
let going = |name: &str, steps: Vec<Step>| Progress {
steps,
going: Some(name.into()),
};
let mut outbox = Outbox::default();
// Nothing new is nothing to send.
outbox.round(b"", &[], Progress::default(), false);
assert_eq!(outbox.next("c", "t"), None);
// A step going is sent, with what it printed; refused, it waits.
outbox.round(b"==> a\nhi\n", &[], going("a", vec![]), false);
let body = outbox.next("c", "t").unwrap();
assert_eq!(body["text"], "==> a\nhi\n");
assert_eq!(body["going"], "a");
outbox.answered(&body, false);
assert_eq!(outbox.next("c", "t"), None);
assert!(!outbox.done);
// Next round it goes again, from where it was, with what came since.
outbox.round(b"more\n", &[], going("b", vec![step("a")]), false);
let body = outbox.next("c", "t").unwrap();
assert_eq!(
(&body["offset"], &body["text"]),
(&json!(0), &json!("==> a\nhi\nmore\n"))
);
assert_eq!(body["steps"][0]["name"], "a");
outbox.answered(&body, true);
assert_eq!(outbox.next("c", "t"), None);
// More than a piece goes in pieces, how the run stands on the last.
let long = "x".repeat(PIECE) + "\n";
outbox.round(long.as_bytes(), &[], going("b", vec![step("a")]), false);
let first = outbox.next("c", "t").unwrap();
assert_eq!((&first["offset"], first.get("steps")), (&json!(14), None));
outbox.answered(&first, true);
let last = outbox.next("c", "t").unwrap();
assert_eq!(last["text"], "\n");
outbox.answered(&last, true);
// Too much held is let go from its start, whole characters.
let lines = format!("aé\n{}\n\n", "b\n".repeat(UNSENT_MAX / 2 - 2));
assert_eq!(lines.len(), UNSENT_MAX + 2);
outbox.round(lines.as_bytes(), &[], Progress::default(), false);
assert!(outbox.text.starts_with("\nb\n"), "{:?}", &outbox.text[..4]);
let ahead = (lines.len() - outbox.text.len()) as u64;
assert_eq!(outbox.at, 14 + PIECE as u64 + 1 + ahead);
outbox.text.clear();
// Over, the last send says so; refused then, it is let go.
outbox.round(b"", &[], Progress::default(), true);
let body = outbox.next("c", "t").unwrap();
assert_eq!(body["ended"], true);
outbox.answered(&body, false);
assert!(outbox.done);
let mut taken = Outbox::default();
taken.round(b"", &[], Progress::default(), true);
let body = taken.next("c", "t").unwrap();
taken.answered(&body, true);
assert!(taken.done);
}
}