chadtmiller.com/knotless
git that lives in your Atmosphere account
7
//! `dev.knotless.check` records, in the runner's own account: one per
//! workflow, ref and commit, replaced as each run of it goes. A private
//! repository's are in the runner's repo in its space, as is all that
//! says anything of it.
use anyhow::Result;
use knotless::crypto::PublicKey;
use knotless::env::Ctx;
use knotless::identity::{self, Identity};
use knotless::records::{BlobRef, MinutesRecord, nsid};
use knotless::remote::{self, Remote};
use knotless::sealed;
use knotless::session;
use knotless::space;
use knotless::xrpc::{Auth, Xrpc, XrpcError};
use serde_json::json;
use std::time::Duration;
pub const NSID: &str = nsid::CHECK;
/// At most this much of a run's output is kept: its tail.
pub const LOG_MAX: usize = 1_000_000;
pub use knotless::keys::check_rkey as rkey;
pub use knotless::records::{CheckRecord as Check, CheckStep as Step, status};
/// Values shorter than this are left in a log: masking them would mask
/// ordinary output too.
pub const SECRET_MIN: usize = 4;
/// `log` with each of `secrets`, and each line of one that has several,
/// replaced by `***`, longest first so a secret inside another leaves no
/// part of it showing.
pub fn redact(log: &[u8], secrets: &[String]) -> Vec<u8> {
let mut masks: Vec<&str> = secrets
.iter()
.flat_map(|s| std::iter::once(s.as_str()).chain(s.lines().map(str::trim)))
.filter(|s| s.len() >= SECRET_MIN)
.collect();
masks.sort_by_key(|s| std::cmp::Reverse(s.len()));
masks.dedup();
let mut out = log.to_vec();
for mask in masks {
out = replace(&out, mask.as_bytes());
}
out
}
fn replace(haystack: &[u8], needle: &[u8]) -> Vec<u8> {
let mut out = Vec::with_capacity(haystack.len());
let mut rest = haystack;
while let Some(at) = rest.windows(needle.len()).position(|w| w == needle) {
out.extend_from_slice(&rest[..at]);
out.extend_from_slice(b"***");
rest = &rest[at + needle.len()..];
}
out.extend_from_slice(rest);
out
}
/// The log of a run that takes another's, `earlier`, as its own: what it
/// `said` first, then as much of the end of `earlier` as fits within
/// [`LOG_MAX`] with it, so it stands alone once that run's check is gone.
pub fn copied(said: &str, earlier: &[u8]) -> Vec<u8> {
let room = LOG_MAX.saturating_sub(said.len());
let kept = &earlier[earlier.len().saturating_sub(room)..];
[said.as_bytes(), kept].concat()
}
/// The most checks one write deletes: the reference PDS takes 200
/// writes in one `applyWrites`.
pub const DELETES: usize = 200;
/// The last [`LOG_MAX`] bytes of `log`.
pub fn tail(log: &[u8]) -> &[u8] {
&log[log.len().saturating_sub(LOG_MAX)..]
}
/// The wait before a write the PDS failed is tried again, doubled each
/// time, while it is at most [`WRITE_RETRY_MAX`]: four tries in all.
const WRITE_RETRY: Duration = Duration::from_millis(500);
const WRITE_RETRY_MAX: Duration = Duration::from_secs(2);
/// The most a write waits, all told, for rate limits that refuse it to
/// let writes in again: a run holds its job meanwhile.
const LIMITED_MAX: Duration = Duration::from_secs(10 * 60);
/// Whether `err` may pass if tried again: the PDS failed or did not
/// answer, rather than refusing what was asked.
fn passing(err: &anyhow::Error) -> bool {
XrpcError::of(err).is_none_or(|e| e.status >= 500 || e.status == 429)
}
/// How long a write refused by a rate limit, as `err` was, waits for it to
/// let writes in again, at `now` (seconds since the epoch), having waited
/// `limited` so far: `None` if `err` is not one that says when, or that is
/// past [`LIMITED_MAX`].
fn limited_for(err: &anyhow::Error, limited: Duration, now: u64) -> Option<Duration> {
let reset = XrpcError::of(err).filter(|e| e.status == 429)?.reset?;
let wait = Duration::from_secs(reset.saturating_sub(now)).max(WRITE_RETRY);
(limited + wait <= LIMITED_MAX).then_some(wait)
}
/// Seconds since the epoch.
fn epoch_secs() -> u64 {
let since = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
since.unwrap_or_default().as_secs()
}
/// Writes checks as the runner's account.
pub struct Reporter {
ctx: Ctx,
pub identity: Identity,
xrpc: Xrpc,
/// Where its writes go: its account's PDS, or a test's stand-in.
writes: Xrpc,
}
impl Reporter {
/// The account this process is logged in as.
pub async fn new(ctx: &Ctx) -> Result<Reporter> {
let me = session::me(ctx).await?;
let identity = identity::resolve(ctx, &me.did).await?;
let xrpc = Xrpc::new(ctx.http.clone(), &identity.pds);
Ok(Reporter {
writes: xrpc.clone(),
xrpc,
ctx: ctx.clone(),
identity,
})
}
/// Points its writes at `pds` rather than its account's PDS, as a
/// test's stand-in in front of it does, over `ctx`'s connections: its
/// reads go as ever.
pub fn through(mut self, ctx: &Ctx, pds: &str) -> Reporter {
self.writes = Xrpc::new(ctx.http.clone(), pds);
self
}
/// The URI of this account's check `rkey` of the repository `of`
/// reads: in full, in its space, for a private one.
pub fn uri_of(&self, of: &Remote, rkey: &str) -> String {
let at = knotless::aturi::AtUri::public(&self.identity.did, NSID, rkey);
at.resolved(of.space_ref().as_ref()).to_string()
}
/// Files `check` under `rkey`, of the repository `of` reads.
pub async fn put(&self, of: &Remote, rkey: &str, check: &Check) -> Result<()> {
let mut record = serde_json::to_value(check)?;
record["$type"] = NSID.into();
let mut body = json!({
"repo": self.identity.did,
"collection": NSID,
"rkey": rkey,
"record": record,
});
let nsid = match &of.space {
Some(space) => {
body["space"] = space.uri.clone().into();
"com.atproto.space.putRecord"
}
None => "com.atproto.repo.putRecord",
};
self.write(nsid, &body).await
}
/// Files `minutes`, of its owner's public repositories, under the
/// owner's DID.
pub async fn put_minutes(&self, owner: &str, minutes: &MinutesRecord) -> Result<()> {
let mut record = serde_json::to_value(minutes)?;
record["$type"] = nsid::MINUTES.into();
let body = json!({
"repo": self.identity.did,
"collection": nsid::MINUTES,
"rkey": owner,
"record": record,
});
self.write("com.atproto.repo.putRecord", &body).await
}
/// Publishes `key` as the key owners seal its secrets to, unless its
/// record already holds it.
pub async fn put_key(&self, key: &PublicKey) -> Result<()> {
let key = key.to_did_key();
let params = [
("repo", self.identity.did.as_str()),
("collection", sealed::KEY_NSID),
("rkey", sealed::KEY_RKEY),
];
let held =
self.xrpc
.query::<serde_json::Value>("com.atproto.repo.getRecord", ¶ms, None);
let held = held.await.ok().map(|r| r["value"]["key"].clone());
if held.is_some_and(|k| k == key.as_str()) {
return Ok(());
}
let body = json!({
"repo": self.identity.did,
"collection": sealed::KEY_NSID,
"rkey": sealed::KEY_RKEY,
"record": {
"$type": sealed::KEY_NSID,
"key": key,
"createdAt": knotless::records::now(),
},
});
self.write("com.atproto.repo.putRecord", &body).await
}
/// Calls the procedure `nsid` with `body`, as [`Reporter::retried`].
async fn write(&self, nsid: &str, body: &serde_json::Value) -> Result<()> {
self.retried(|auth| async move {
let wrote =
self.writes
.procedure::<serde::de::IgnoredAny>(nsid, Some(body), Some(&auth));
wrote.await.map(drop)
})
.await
}
/// What `call` returns, as the account, tried again a few times while
/// the PDS fails it or does not answer in time, as one busy for a
/// minute does, and once a rate limit refusing it lets writes in again:
/// a write given up on leaves a check that never ends, or a run that
/// passed filed as an error.
async fn retried<T, F: Future<Output = Result<T>>>(
&self,
call: impl Fn(Auth) -> F,
) -> Result<T> {
let (mut wait, mut limited) = (WRITE_RETRY, Duration::ZERO);
loop {
let auth = session::ensure(&self.ctx, &self.identity).await?.auth();
let err = match call(auth).await {
Err(err) => err,
done => return done,
};
let until = match limited_for(&err, limited, epoch_secs()) {
Some(until) => {
limited += until;
until
}
None if wait <= WRITE_RETRY_MAX && passing(&err) => {
wait *= 2;
wait / 2
}
None => return Err(err),
};
crate::telemetry::retried(&err, until);
tokio::time::sleep(until).await;
}
}
/// The checks keyed under `prefix` that this account filed, by key,
/// of the repository `of` reads: one that does not read as a check is
/// left out.
pub async fn checks(&self, of: &Remote, prefix: &str) -> Result<Vec<(String, Check)>> {
let mine = self.mine(of).await?;
let found = mine.scan_all::<serde_json::Value>(NSID, prefix).await?;
Ok(found
.into_iter()
.filter_map(|s| Some((s.rkey, serde_json::from_value::<Check>(s.value).ok()?)))
.collect())
}
/// The checks keyed under `prefix` that this account left queued or
/// running, of the repository `of` reads.
pub async fn running(&self, of: &Remote, prefix: &str) -> Result<Vec<(String, Check)>> {
let all = self.checks(of, prefix).await?;
let running = all.into_iter().filter(|(_, c)| status::going(&c.status));
Ok(running.collect())
}
/// Deletes the checks this account filed under `rkeys`, of the
/// repository `of` reads, [`DELETES`] to a write.
pub async fn delete(&self, of: &Remote, rkeys: &[String]) -> Result<()> {
let mut mine = self.mine(of).await?;
mine.xrpc = self.writes.clone();
for some in rkeys.chunks(DELETES) {
let writes: Vec<_> = some.iter().map(|r| remote::delete(NSID, r)).collect();
let (mine, writes) = (&mine, &writes);
self.retried(|auth| async move { mine.apply(writes.clone(), &auth).await })
.await?;
}
Ok(())
}
/// Whether this account filed any check keyed under `prefix`, of the
/// repository `of` reads.
pub async fn any(&self, of: &Remote, prefix: &str) -> Result<bool> {
let mine = self.mine(of).await?;
let page = mine.scan::<serde_json::Value>(NSID, prefix, None).await?;
Ok(!page.items.is_empty())
}
/// The check this account filed under `rkey`, of the repository `of`
/// reads, if there is one.
pub async fn get(&self, of: &Remote, rkey: &str) -> Result<Option<Check>> {
let mine = self.mine(of).await?;
let found = mine.get_record::<Check>(NSID, rkey).await?;
Ok(found.map(|s| s.value))
}
/// This account's records about the repository `of` reads: in its
/// repo in the space, for a private one, read with a space credential
/// as a member's are, even its own.
async fn mine(&self, of: &Remote) -> Result<Remote> {
let space = match &of.space {
Some(space) => {
let joined = space::joined(&self.ctx, &space.uri).await?;
Some(joined.at(&self.identity.did))
}
None => None,
};
let mine = Remote {
xrpc: self.xrpc.clone(),
identity: self.identity.clone(),
url: of.url.clone(),
space,
};
Ok(mine)
}
/// The log `log` of a check this account filed of the repository `of`
/// reads.
pub async fn log_of(&self, of: &Remote, log: &BlobRef) -> Result<Vec<u8>> {
let mine = self.mine(of).await?;
mine.get_blob(&log.link.cid).await
}
pub async fn upload_log(&self, log: &[u8]) -> Result<BlobRef> {
self.upload(tail(log).to_vec(), knotless::records::LOG_MIME)
.await
}
pub async fn upload(&self, bytes: Vec<u8>, mime: &str) -> Result<BlobRef> {
let bytes = &bytes;
self.retried(
|auth| async move { self.writes.upload_blob(bytes.clone(), mime, &auth).await },
)
.await
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn copies_keep_what_they_said_first() {
assert_eq!(
copied("==> passed\n", b"==> test\nok\n"),
b"==> passed\n==> test\nok\n"
);
let long = vec![b'x'; LOG_MAX];
let log = copied("==> passed\n", &long);
assert_eq!(log.len(), LOG_MAX);
assert!(log.starts_with(b"==> passed\nxx"));
}
#[test]
fn keys_and_tails() {
let k = rkey(
"did:plc:a",
"3l3qo2vuowo2a",
"refs/heads/main",
"abc",
None,
"ci",
);
assert_eq!(k.split('.').count(), 5);
let pull = rkey(
"did:plc:a",
"3l3qo2vuowo2a",
"at://p",
"abc",
Some("def"),
"ci",
);
assert!(pull.ends_with(".abc.def.ci"), "{pull}");
assert!(k.starts_with(&knotless::keys::check_prefix("did:plc:a", "3l3qo2vuowo2a")));
assert!(k.ends_with(".ci"));
assert_ne!(
k,
rkey(
"did:plc:b",
"3l3qo2vuowo2a",
"refs/heads/main",
"abc",
None,
"ci"
)
);
assert_eq!(tail(b"abc"), b"abc");
let secrets = [
"s3cret".to_string(),
"abc".to_string(),
"-----BEGIN\n keybody1\n-----END".to_string(),
];
let log = b"s3crets3cret abc\n-----BEGIN\n keybody1\n-----END\nkeybody1 again\n";
assert_eq!(
String::from_utf8(redact(log, &secrets)).unwrap(),
"****** abc\n***\n*** again\n"
);
assert_eq!(redact(b"nothing", &[]), b"nothing");
let big = vec![b'x'; LOG_MAX + 5];
assert_eq!(tail(&big).len(), LOG_MAX);
}
#[test]
fn what_is_tried_again() {
let refused = |status| {
anyhow::Error::from(XrpcError {
status,
error: "E".into(),
message: "m".into(),
reset: None,
})
};
assert!(passing(&refused(500)));
assert!(passing(&refused(429)));
assert!(!passing(&refused(400)));
assert!(passing(&anyhow::anyhow!("operation timed out")));
}
#[test]
fn rate_limits_are_waited_out() {
let limited = |status, reset| {
anyhow::Error::from(XrpcError {
status,
error: "RateLimitExceeded".into(),
message: "m".into(),
reset,
})
};
let none = Duration::ZERO;
let wait = |err: &anyhow::Error, so_far| limited_for(err, so_far, 1_000);
assert_eq!(
wait(&limited(429, Some(1_030)), none),
Some(Duration::from_secs(30))
);
// A reset already past is tried again soon, not at once.
assert_eq!(wait(&limited(429, Some(990)), none), Some(WRITE_RETRY));
// One that says nothing of when, or is not a limit, is backed off.
assert_eq!(wait(&limited(429, None), none), None);
assert_eq!(wait(&limited(503, Some(1_030)), none), None);
assert_eq!(wait(&anyhow::anyhow!("timed out"), none), None);
// Past what it waits all told, it gives up.
assert_eq!(wait(&limited(429, Some(1_000 + 3_600)), none), None);
let most = LIMITED_MAX - Duration::from_secs(10);
assert_eq!(wait(&limited(429, Some(1_030)), most), None);
assert!(epoch_secs() > 1_700_000_000);
}
}