//! Loopback HTTP client for packsetd.
//!
//! Reads `PACKSET_URL` or `INSIDE_MEMORY_URL`. search/get against packsetd; no SQLite.
//! Does not open LMDB.
use serde::{Deserialize, Serialize};
use std::env;
use std::error::Error as StdError;
use std::path::PathBuf;
use std::time::Duration;
/// How long one request may take: `PACKSET_TIMEOUT_MS`, else thirty seconds.
/// A write that waits behind others on a busy seat is late, not failed. The
/// same budget is sent as `X-Packset-Deadline`, and the writer drops a
/// request that is still waiting when it passes.
fn timeout() -> Duration {
std::env::var("PACKSET_TIMEOUT_MS")
.ok()
.and_then(|v| v.trim().parse::<u64>().ok())
.filter(|ms| *ms > 0)
.map_or(Duration::from_secs(30), Duration::from_millis)
}
/// The wait before a down or busy writer is asked again, tripled each time,
/// and never longer than five seconds. The first waits are 50, 150 and 450 ms.
const BUSY_WAIT: Duration = Duration::from_millis(50);
/// Each wait is this many times the last.
const BUSY_GROWTH: u32 = 3;
/// The longest pause between asks.
const BUSY_WAIT_CAP: Duration = Duration::from_secs(5);
/// Whether another ask still ends inside `budget`.
///
/// `waited` is the sum of the waits already taken, not the clock. A round
/// trip that is slow on one host does not spend the next wait, so every
/// host asks the same number of times for one budget.
fn ask_again(waited: Duration, wait: Duration, budget: Duration) -> bool {
waited.saturating_add(wait) < budget
}
/// How many times [`send`] asks inside `budget` when the writer sends no
/// `Retry-After`. The first ask costs no wait. Each later ask costs the
/// next backoff, and a wait that would pass the budget is not taken.
#[cfg(test)]
fn attempts_in(budget: Duration) -> u32 {
let mut waited = Duration::ZERO;
let mut wait = BUSY_WAIT;
let mut attempts = 1u32;
while ask_again(waited, wait, budget) {
waited = waited.saturating_add(wait);
wait = (wait * BUSY_GROWTH).min(BUSY_WAIT_CAP);
attempts = attempts.saturating_add(1);
}
attempts
}
/// What [`send`] could not turn into a response.
enum Attempt {
/// The writer is down or still busy after the budget. One sentence.
Plain(String),
/// Any other failure, for the caller to read.
Ureq(Box<ureq::Error>),
}
/// How long to sleep, and how much of the budget that wait spends. The
/// charge is the full wait, so every host stops at the same attempt. The
/// sleep is half of that plus a fraction of the rest, so a herd does not
/// wake together. `Retry-After` raises the wait when it is longer.
struct Pause {
sleep: Duration,
charge: Duration,
}
fn planned_wait(backoff: Duration, retry_after: Option<Duration>) -> Duration {
match retry_after {
Some(after) if after > backoff => after,
_ => backoff,
}
}
/// Half of `wait`, plus up to the other half. `salt` picks the fraction.
fn jittered(wait: Duration, salt: u64) -> Duration {
let ms = u64::try_from(wait.as_millis()).unwrap_or(u64::MAX).max(1);
let half = ms / 2;
let extra = if half == 0 { 0 } else { salt % (half + 1) };
Duration::from_millis(half + extra)
}
/// The next wait, when it still ends inside `budget`.
fn pause_for(
backoff: Duration,
retry_after: Option<Duration>,
waited: Duration,
budget: Duration,
salt: u64,
) -> Option<Pause> {
let charge = planned_wait(backoff, retry_after);
if !ask_again(waited, charge, budget) {
return None;
}
let room = budget.saturating_sub(waited).max(Duration::from_millis(1));
Some(Pause {
sleep: jittered(charge, salt)
.min(room)
.max(Duration::from_millis(1)),
charge,
})
}
fn retry_after_of(header: Option<&str>) -> Option<Duration> {
let raw = header?.trim();
if raw.is_empty() {
return None;
}
raw.parse::<u64>().ok().map(Duration::from_secs)
}
fn jitter_salt(asked: u32) -> u64 {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|elapsed| u64::from(elapsed.subsec_nanos()))
.unwrap_or(0);
nanos ^ u64::from(asked)
}
/// Send `req`, with `body` when there is one, inside `budget`. A connection
/// refused, a 429, or a 503 is asked again for every wait that fits in the
/// budget. A 503 or 429 that carries `Retry-After` waits at least that long.
/// The waits that are charged are what the budget spends. A slow round trip
/// does not drop an ask. The writer answers 503 before any of the request
/// runs, so a write sent again is not stored twice. Each ask carries
/// `X-Packset-Deadline` at the end of this budget. The first refusal of a
/// loopback writer starts the local daemon once, when this seat has a home
/// and a `packsetd` binary.
fn send(
req: ureq::Request,
body: Option<&serde_json::Value>,
budget: Duration,
) -> Result<ureq::Response, Attempt> {
let deadline = std::time::SystemTime::now() + budget;
let deadline_ms = deadline
.duration_since(std::time::UNIX_EPOCH)
.map(|elapsed| elapsed.as_millis().to_string())
.unwrap_or_default();
let mut waited = Duration::ZERO;
let mut wait = BUSY_WAIT;
let mut asked = 0u32;
let mut started_daemon = false;
loop {
asked = asked.saturating_add(1);
let left = budget.saturating_sub(waited).max(Duration::from_millis(1));
let mut attempt = req.clone().timeout(left);
if !deadline_ms.is_empty() {
attempt = attempt.set("X-Packset-Deadline", &deadline_ms);
}
if started_daemon {
if let Some(token) = token() {
attempt = attempt.set("Authorization", &format!("Bearer {token}"));
}
}
let answered = match body {
Some(json) => attempt.send_json(json),
None => attempt.call(),
};
match answered {
Err(ureq::Error::Status(code, response)) if code == 503 || code == 429 => {
let after = retry_after_of(response.header("Retry-After"));
match pause_for(wait, after, waited, budget, jitter_salt(asked)) {
Some(pause) => {
std::thread::sleep(pause.sleep);
waited = waited.saturating_add(pause.charge);
wait = (wait * BUSY_GROWTH).min(BUSY_WAIT_CAP);
}
None => return Err(Attempt::Plain(busy_line(response))),
}
}
Err(err) if connect_refused(&err) => {
match pause_for(wait, None, waited, budget, jitter_salt(asked)) {
Some(pause) => {
if !started_daemon {
started_daemon = autostart(req.url());
}
std::thread::sleep(pause.sleep);
waited = waited.saturating_add(pause.charge);
wait = (wait * BUSY_GROWTH).min(BUSY_WAIT_CAP);
}
None => {
return Err(Attempt::Plain("packsetd is not running.".to_string()));
}
}
}
other => return other.map_err(|err| Attempt::Ureq(Box::new(err))),
}
}
}
/// The writer's own reason, as one sentence.
fn busy_line(response: ureq::Response) -> String {
let text = response.into_string().unwrap_or_default();
let reason = serde_json::from_str::<serde_json::Value>(&text)
.ok()
.and_then(|value| {
value
.get("error")
.and_then(|reason| reason.as_str())
.map(str::to_string)
})
.unwrap_or(text);
let reason = reason.trim();
if reason.is_empty() || reason.eq_ignore_ascii_case("busy") {
"packsetd is busy.".to_string()
} else if reason.ends_with('.') {
format!("packsetd is busy: {reason}")
} else {
format!("packsetd is busy: {reason}.")
}
}
/// Whether `err` is a TCP refusal. A reset or a timeout is a different failure.
fn connect_refused(err: &ureq::Error) -> bool {
let ureq::Error::Transport(transport) = err else {
return false;
};
if transport
.source()
.and_then(|source| source.downcast_ref::<std::io::Error>())
.is_some_and(|io| io.kind() == std::io::ErrorKind::ConnectionRefused)
{
return true;
}
err.to_string().contains("Connection refused")
}
fn into_error(url: &str, err: Attempt) -> Error {
match err {
Attempt::Plain(text) => Error::Unavailable(text),
Attempt::Ureq(err) => refused(url, err),
}
}
fn path_seg(id: &str) -> String {
let mut out = String::with_capacity(id.len());
for b in id.bytes() {
match b {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
out.push(b as char)
}
_ => out.push_str(&format!("%{b:02X}")),
}
}
out
}
/// The port a writer listens on when nothing names one. The command line,
/// the server and this client agree on it, so a seat needs no variable set.
pub const DEFAULT_PORT: u16 = 8761;
/// Load `~/.config/ljos/env` (KEY=VALUE) when the process has not set
/// those keys. `ljos`, `packset`, and `packset-mcp` then share one pack.
pub fn load_seat_env() {
let Some(home) = env::var_os("HOME") else {
return;
};
let path = PathBuf::from(home).join(".config/ljos/env");
let Ok(text) = std::fs::read_to_string(path) else {
return;
};
for line in text.lines() {
let line = line.trim();
if line.is_empty() || line.starts_with('#') {
continue;
}
let Some((k, v)) = line.split_once('=') else {
continue;
};
let k = k.trim();
if k.is_empty() || env::var_os(k).is_some() {
continue;
}
env::set_var(k, v.trim());
}
}
/// The workspace the seat's memory lives in: `PACKSET_WORKSPACE` after
/// loading the seat env file, else `seat`. Not `default`, and not the
/// working directory's git remote: those two are how one harness's
/// remember missed the other's sitting.
#[must_use]
pub fn resolved_workspace() -> String {
load_seat_env();
env::var("PACKSET_WORKSPACE")
.ok()
.map(|w| w.trim().to_string())
.filter(|w| !w.is_empty())
.unwrap_or_else(|| "seat".to_string())
}
/// `PACKSET_PORT` (`LJOS_MEM_PORT` is an alias), else [`DEFAULT_PORT`].
#[must_use]
pub fn default_port() -> u16 {
env::var("PACKSET_PORT")
.or_else(|_| env::var("LJOS_MEM_PORT"))
.ok()
.and_then(|raw| raw.trim().parse().ok())
.unwrap_or(DEFAULT_PORT)
}
/// The token file a writer leaves in its pack home: `PACKSET_TOKEN_FILE`,
/// else `token` under `PACKSET_HOME` (`LJOS_HOME`,
/// `LJOS_INSIDE_MEMORY_HOME`), else under the default home
/// ([`packset_core::home::resolve`]). packsetd resolves its home the same
/// way.
#[must_use]
pub fn token_path() -> Option<PathBuf> {
if let Some(file) = env::var_os("PACKSET_TOKEN_FILE").filter(|v| !v.is_empty()) {
return Some(PathBuf::from(file));
}
packset_core::home::resolve().map(|home| home.join(TOKEN_FILE))
}
/// The name of the token file in a pack home.
pub const TOKEN_FILE: &str = "token";
/// The token this seat shows the writer: `PACKSET_TOKEN`, else the first
/// line of [`token_path`]. `None` when there is neither, and the writer then
/// answers 401 to everything but `/health`.
#[must_use]
pub fn token() -> Option<String> {
if let Ok(t) = env::var("PACKSET_TOKEN") {
let t = t.trim().to_string();
if !t.is_empty() {
return Some(t);
}
}
let text = std::fs::read_to_string(token_path()?).ok()?;
let t = text.lines().next()?.trim().to_string();
(!t.is_empty()).then_some(t)
}
/// Start the local writer once. `true` when a process was spawned.
///
/// A home has to be named (`PACKSET_HOME` or an older alias). The binary is
/// `PACKSET_BIN`, else `packsetd` beside this program, else on `PATH`.
fn autostart(url: &str) -> bool {
use std::sync::atomic::{AtomicBool, Ordering};
static STARTED: AtomicBool = AtomicBool::new(false);
if STARTED.load(Ordering::Acquire) {
return false;
}
let Some(port) = loopback_port(url) else {
return false;
};
load_seat_env();
let Some(home) = packset_core::home::named_home() else {
return false;
};
let Some(bin) = daemon_binary() else {
return false;
};
if STARTED.swap(true, Ordering::AcqRel) {
return false;
}
spawn_daemon(&bin, &home, port)
}
/// `packsetd`, if this seat has one.
fn daemon_binary() -> Option<PathBuf> {
if let Some(named) = env::var_os("PACKSET_BIN").filter(|value| !value.is_empty()) {
let path = PathBuf::from(named);
if path.is_file() {
return Some(path);
}
}
if let Some(beside) = env::current_exe()
.ok()
.and_then(|exe| exe.parent().map(|dir| dir.join("packsetd")))
.filter(|path| path.is_file())
{
return Some(beside);
}
let paths = env::var_os("PATH")?;
env::split_paths(&paths)
.map(|dir| dir.join("packsetd"))
.find(|path| path.is_file())
}
/// Spawn `bin` on `port` for `home`. The child leaves the caller's session
/// on Unix so the client's exit does not take the writer with it.
fn spawn_daemon(bin: &std::path::Path, home: &std::path::Path, port: u16) -> bool {
use std::process::{Command, Stdio};
let mut cmd = Command::new(bin);
cmd.arg("--home")
.arg(home)
.arg("--port")
.arg(port.to_string())
.stdin(Stdio::null());
if let Ok(log) = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(home.join("packsetd.log"))
{
if let Ok(errs) = log.try_clone() {
cmd.stdout(Stdio::from(log)).stderr(Stdio::from(errs));
}
} else {
cmd.stdout(Stdio::null()).stderr(Stdio::null());
}
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
// SAFETY: `pre_exec` runs in the forked child before exec. `setsid`
// there detaches the writer from this client's session.
unsafe {
cmd.pre_exec(|| {
extern "C" {
fn setsid() -> i32;
}
if setsid() == -1 {
return Err(std::io::Error::last_os_error());
}
Ok(())
});
}
}
cmd.spawn().is_ok()
}
/// `request` with this seat's token as a bearer header. A caller that speaks
/// to the writer outside [`PacksetClient`] wraps its request in this.
///
/// The token stays home when another user's process holds the writer's
/// loopback port: anyone can answer `/health` with `packsetd ok`, and a
/// listener that got there first would otherwise collect the token.
pub fn authorize(request: ureq::Request) -> ureq::Request {
if foreign_listener(request.url()) {
return request;
}
match token() {
Some(t) => request.set("Authorization", &format!("Bearer {t}")),
None => request,
}
}
/// The port of a loopback `url`, `None` for any other host.
///
/// The foreign-listener check is Linux only. Starting a local writer uses
/// the same parse on every host.
fn loopback_port(url: &str) -> Option<u16> {
let rest = url
.strip_prefix("http://")
.or_else(|| url.strip_prefix("https://"))?;
let authority = rest.split(['/', '?', '#']).next()?;
let (host, port) = match authority.rsplit_once(':') {
Some((host, port)) if !port.contains(']') => (host, port.parse().ok()?),
_ => (authority, 80),
};
matches!(host, "127.0.0.1" | "localhost" | "[::1]").then_some(port)
}
/// The owners of the sockets listening on `port`, from one `/proc/net/tcp`
/// or `tcp6` table: local address is the second column, state the fourth,
/// uid the eighth.
#[cfg(target_os = "linux")]
fn listen_uids(table: &str, port: u16) -> Vec<u32> {
table
.lines()
.skip(1)
.filter_map(|line| {
let fields: Vec<&str> = line.split_whitespace().collect();
let local = fields.get(1)?;
if fields.get(3) != Some(&"0A") {
return None;
}
let hex = local.rsplit(':').next()?;
if u16::from_str_radix(hex, 16).ok()? != port {
return None;
}
fields.get(7)?.parse().ok()
})
.collect()
}
/// Whether every socket listening on the loopback port of `url` belongs to
/// a user other than this one. Unknown is not foreign: off Linux, or when
/// the tables cannot be read, the token goes as before.
fn foreign_listener(url: &str) -> bool {
#[cfg(target_os = "linux")]
{
use std::os::unix::fs::MetadataExt;
let Some(port) = loopback_port(url) else {
return false;
};
let Ok(me) = std::fs::metadata("/proc/self").map(|m| m.uid()) else {
return false;
};
let owners: Vec<u32> = ["/proc/net/tcp", "/proc/net/tcp6"]
.iter()
.filter_map(|t| std::fs::read_to_string(t).ok())
.flat_map(|text| listen_uids(&text, port))
.collect();
!owners.is_empty() && owners.iter().all(|uid| *uid != me)
}
#[cfg(not(target_os = "linux"))]
{
let _ = url;
false
}
}
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[error("packset url missing")]
NoUrl,
#[error("http: {0}")]
Http(#[from] Box<ureq::Error>),
#[error("io: {0}")]
Io(#[from] std::io::Error),
#[error("json: {0}")]
Json(#[from] serde_json::Error),
#[error("bad response: {0}")]
Bad(String),
/// The writer was down or busy for the whole budget. One sentence.
#[error("{0}")]
Unavailable(String),
}
#[derive(Debug, Clone)]
pub struct PacksetClient {
base: String,
/// A workspace pinned by the caller; `None` reads the environment and
/// the working directory.
workspace: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Hit {
pub id: Option<String>,
pub text: String,
#[serde(default)]
pub score: f64,
#[serde(default)]
pub kind: String,
/// When the memory was written, the writer's clock, RFC 3339. Absent
/// on card paragraphs, which have no clock.
#[serde(default)]
pub ts: Option<String>,
/// How many of the panel's ballots named this hit, and how many ran.
/// Two of three is agreement; one of three is one scorer's opinion.
/// The atom's entities: `seat:<name>` names the seat that wrote it,
/// `persona:<name>` a persona's own claim, `habit:<name>` a reading.
#[serde(default)]
pub entities: Vec<String>,
#[serde(default)]
pub ballots: Option<u32>,
#[serde(default)]
pub of: Option<u32>,
/// The cross-encoder score when the second stage scored this hit.
/// Absent, or null, when it did not.
#[serde(default)]
pub rerank_score: Option<f64>,
/// Labels on the claim, `flag:credential-shaped` and the caller's own.
/// Empty when the hit has none, including a writer that omits the field.
#[serde(default)]
pub labels: Vec<String>,
}
/// A refusal, carrying the reason the writer gave in its body.
fn refused(url: &str, e: Box<ureq::Error>) -> Error {
match *e {
ureq::Error::Status(code, response) => {
let text = response.into_string().unwrap_or_default();
let reason = serde_json::from_str::<serde_json::Value>(&text)
.ok()
.and_then(|v| v.get("error").and_then(|r| r.as_str()).map(str::to_string))
.unwrap_or(text);
let reason = reason.trim();
if code == 401 {
return Error::Bad(format!(
"{url}: 401: {reason}; this seat's token does not match the writer's, \
so it is another user's writer or PACKSET_HOME names another pack"
));
}
if reason.is_empty() {
Error::Bad(format!("{url}: status code {code}"))
} else {
Error::Bad(format!("{url}: {code}: {reason}"))
}
}
other => Error::Http(Box::new(other)),
}
}
impl PacksetClient {
pub fn new(base: impl Into<String>) -> Self {
let mut base = base.into();
while base.ends_with('/') {
base.pop();
}
Self {
base,
workspace: None,
}
}
/// Pin the workspace this client speaks for, ahead of `PACKSET_WORKSPACE`
/// and the working directory. A seat that is one memory across every
/// repository it works in sets this once.
#[must_use]
pub fn with_workspace(mut self, workspace: impl Into<String>) -> Self {
let workspace = workspace.into();
self.workspace = (!workspace.is_empty()).then_some(workspace);
self
}
/// The writer the seat talks to, with nothing set: `PACKSET_URL`
/// (`INSIDE_MEMORY_URL` is an alias), else the loopback port the command
/// line starts a writer on, `PACKSET_PORT` (`LJOS_MEM_PORT`) or 8761.
/// `PACKSET_URL=off` is the one way to have no pack.
pub fn from_env() -> Result<Self, Error> {
load_seat_env();
let url = env::var("PACKSET_URL")
.or_else(|_| env::var("INSIDE_MEMORY_URL"))
.ok()
.filter(|url| !url.is_empty());
match url {
Some(url) if url == "off" => Err(Error::NoUrl),
Some(url) => Ok(Self::new(url)),
None => Ok(Self::new(format!("http://127.0.0.1:{}", default_port()))),
}
}
pub fn base(&self) -> &str {
&self.base
}
pub fn workspace(&self) -> String {
if let Some(w) = &self.workspace {
return w.clone();
}
if let Ok(w) = env::var("PACKSET_WORKSPACE") {
if !w.is_empty() {
return w;
}
}
let cwd = env::var("HOSTSEAT_WORKSPACE")
.ok()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.map(std::path::PathBuf::from)
.or_else(|| env::current_dir().ok())
.unwrap_or_else(|| std::path::PathBuf::from("."));
self.workspace_for_cwd(&cwd)
}
/// Workspace id from `/v1/identity` for `cwd`, or `dir:<abs>` if that call fails.
pub fn workspace_for_cwd(&self, cwd: &std::path::Path) -> String {
let abs = cwd.canonicalize().unwrap_or_else(|_| cwd.to_path_buf());
let url = format!("{}/v1/identity", self.base);
let body = send(
authorize(ureq::get(&url)).query("cwd", abs.to_string_lossy().as_ref()),
None,
timeout(),
)
.ok()
.and_then(|r| r.into_string().ok());
if let Some(body) = body {
if let Ok(val) = serde_json::from_str::<serde_json::Value>(&body) {
if let Some(ws) = val.get("workspace").and_then(|v| v.as_str()) {
if !ws.is_empty() {
return ws.to_string();
}
}
}
}
format!("dir:{}", abs.display())
}
pub fn health(&self) -> Result<String, Error> {
let url = format!("{}/health", self.base);
let body = send(authorize(ureq::get(&url)), None, timeout())
.map_err(|e| into_error(&url, e))?
.into_string()?;
Ok(body)
}
/// Queue depth, shed counts and the current limits. `line` is `ok`, or
/// `busy, retrying` when work is waiting.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn stats(&self) -> Result<serde_json::Value, Error> {
let url = format!("{}/stats", self.base);
let body: serde_json::Value = send(authorize(ureq::get(&url)), None, timeout())
.map_err(|e| into_error(&url, e))?
.into_json()?;
Ok(body)
}
pub fn get_atom(&self, workspace: &str, id: &str) -> Result<serde_json::Value, Error> {
let encoded = path_seg(id);
let url = format!("{}/v1/atoms/{encoded}", self.base);
let resp = match send(
authorize(ureq::get(&url)).query("workspace", workspace),
None,
timeout(),
) {
Ok(resp) => resp,
Err(Attempt::Ureq(e)) if matches!(*e, ureq::Error::Status(404, _)) => {
return Err(Error::Bad(format!("no atom {id}")));
}
Err(e) => return Err(into_error(&url, e)),
};
Ok(resp.into_json()?)
}
pub fn list_atoms(&self, workspace: &str) -> Result<Vec<serde_json::Value>, Error> {
self.atoms_as_of(workspace, None)
}
/// Live-now atoms, or the ones that were live at `as_of`.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn atoms_as_of(
&self,
workspace: &str,
as_of: Option<&str>,
) -> Result<Vec<serde_json::Value>, Error> {
let url = format!("{}/v1/atoms", self.base);
let mut req = authorize(ureq::get(&url)).query("workspace", workspace);
if let Some(at) = as_of {
req = req.query("as_of", at);
}
let mut body: serde_json::Value = send(req, None, timeout())
.map_err(|e| into_error(&url, e))?
.into_json()?;
let atoms = body
.get_mut("atoms")
.map(serde_json::Value::take)
.unwrap_or(serde_json::Value::Array(vec![]));
Ok(serde_json::from_value(atoms)?)
}
/// [`Self::atoms_as_of`] without the dense vectors, for a reader of
/// texts, review clocks or rules: the vectors are most of the bytes. A
/// writer older than the `embedding=omit` query sends them anyway.
///
/// # Errors
///
/// The pack not answering, or an answer that is not atoms.
pub fn atoms_without_vectors(
&self,
workspace: &str,
as_of: Option<&str>,
) -> Result<Vec<serde_json::Value>, Error> {
let url = format!("{}/v1/atoms", self.base);
let mut req = authorize(ureq::get(&url))
.query("workspace", workspace)
.query("embedding", "omit");
if let Some(at) = as_of {
req = req.query("as_of", at);
}
let mut body: serde_json::Value = send(req, None, timeout())
.map_err(|e| into_error(&url, e))?
.into_json()?;
let atoms = body
.get_mut("atoms")
.map(serde_json::Value::take)
.unwrap_or(serde_json::Value::Array(vec![]));
Ok(serde_json::from_value(atoms)?)
}
pub fn search(&self, workspace: &str, q: &str, limit: u32) -> Result<Vec<Hit>, Error> {
self.search_as_of(workspace, q, limit, None)
}
/// Ranked hits, optionally over the atoms that were live at `as_of`.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn search_as_of(
&self,
workspace: &str,
q: &str,
limit: u32,
as_of: Option<&str>,
) -> Result<Vec<Hit>, Error> {
self.search_opts(workspace, q, limit, as_of, false)
}
/// Ranked hits, optionally dated and optionally through the measured
/// cross-encoder stage.
///
/// Off by default. On, the writer spends a forward pass per candidate and
/// the request waits for that rather than the usual five-second budget.
///
/// # Errors
///
/// The request's, or a body that is not a hit list.
pub fn search_opts(
&self,
workspace: &str,
q: &str,
limit: u32,
as_of: Option<&str>,
rerank: bool,
) -> Result<Vec<Hit>, Error> {
self.search_ranked(workspace, q, limit, as_of, rerank, None)
}
/// [`Self::search_opts`] with a rank. `Some("relevance")` asks for the
/// top k by lexical, BM25 and cosine, with no half-life and no
/// retrievability. `None` leaves the shipped order. An older writer
/// ignores the parameter; `GET /v1/status` lists `search.ranks` when
/// this writer honours it.
///
/// # Errors
///
/// The request's, including a rank the writer refuses, or a body that
/// is not a hit list.
pub fn search_ranked(
&self,
workspace: &str,
q: &str,
limit: u32,
as_of: Option<&str>,
rerank: bool,
rank: Option<&str>,
) -> Result<Vec<Hit>, Error> {
let url = format!("{}/v1/search", self.base);
let budget = if rerank {
Duration::from_secs(60).max(timeout())
} else {
timeout()
};
let mut req = authorize(ureq::get(&url))
.query("workspace", workspace)
.query("q", q)
.query("limit", &limit.to_string());
if let Some(at) = as_of {
req = req.query("as_of", at);
}
if rerank {
req = req.query("rerank", "1");
}
if let Some(rank) = rank.map(str::trim).filter(|s| !s.is_empty()) {
req = req.query("rank", rank);
}
let body: serde_json::Value = send(req, None, budget)
.map_err(|e| into_error(&url, e))?
.into_json()?;
let hits = body
.get("hits")
.cloned()
.unwrap_or(serde_json::Value::Array(vec![]));
Ok(serde_json::from_value(hits)?)
}
/// Seat home, atom counts by kind, pin, index and embedder.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn status(&self, workspace: Option<&str>) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/status", self.base);
let mut req = authorize(ureq::get(&url));
if let Some(workspace) = workspace {
req = req.query("workspace", workspace);
}
Ok(send(req, None, timeout())
.map_err(|e| into_error(&url, e))?
.into_json()?)
}
/// The set a workspace is pinned to.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn pin(&self, workspace: &str) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/pin", self.base);
Ok(send(
authorize(ureq::get(&url)).query("workspace", workspace),
None,
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?)
}
/// Pin a workspace to a set.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn set_pin(&self, workspace: &str, name: &str) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/pin", self.base);
Ok(send(
authorize(ureq::put(&url)),
Some(&serde_json::json!({ "workspace": workspace, "name": name })),
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?)
}
/// The deed accessions a workspace's live atoms cite, sorted.
///
/// The accession is the only identifier crossing the tracker, the pack and
/// the deed store, so this is what `deedar evidence -` reads.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn accessions(&self, workspace: &str) -> Result<Vec<String>, Error> {
let url = format!("{}/v1/accessions", self.base);
let body: serde_json::Value = send(
authorize(ureq::get(&url)).query("workspace", workspace),
None,
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?;
let found = body
.get("accessions")
.cloned()
.unwrap_or(serde_json::Value::Array(vec![]));
Ok(serde_json::from_value(found)?)
}
/// Every live atom in a workspace.
///
/// The bodies, not the join keys: this is what a handover carries when
/// somebody is given what the seat learned rather than only what it cites.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn atoms(&self, workspace: &str) -> Result<Vec<serde_json::Value>, Error> {
self.atoms_as_of(workspace, None)
}
/// The live atoms of one kind in a workspace: personas, trust rows,
/// habits, without the rest of the pack.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn atoms_of_kind(
&self,
workspace: &str,
kind: &str,
) -> Result<Vec<serde_json::Value>, Error> {
let url = format!("{}/v1/atoms", self.base);
let body: serde_json::Value = send(
authorize(ureq::get(&url))
.query("workspace", workspace)
.query("kind", kind),
None,
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?;
Ok(body
.get("atoms")
.and_then(serde_json::Value::as_array)
.cloned()
.unwrap_or_default())
}
/// The live atoms in a workspace that cite one deed accession.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn citers(
&self,
workspace: &str,
accession: &str,
) -> Result<Vec<serde_json::Value>, Error> {
let url = format!("{}/v1/citers", self.base);
let body: serde_json::Value = send(
authorize(ureq::get(&url))
.query("workspace", workspace)
.query("accession", accession),
None,
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?;
let found = body
.get("atoms")
.cloned()
.unwrap_or(serde_json::Value::Array(vec![]));
Ok(serde_json::from_value(found)?)
}
/// Tombstone one atom. The daemon keeps the record and drops the index
/// entry, so a forgotten atom stops being recalled without the pack losing
/// the fact that it once held it.
///
/// `why` is the deed accession that withdrew the claim, or the plain-text
/// reason when the seat has no accession. An accession is stored as
/// `retracted_by`. Plain text is stored as `why_text`, and the answer
/// carries one `warning` line. The retraction and what it retracted read
/// back together either way.
///
/// # Errors
///
/// [`Error::Bad`] when the workspace does not hold that atom, else the
/// request's or a body that is not JSON.
pub fn delete_atom(
&self,
workspace: &str,
id: &str,
why: Option<&str>,
) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/atoms/delete", self.base);
let mut body = serde_json::json!({
"workspace": workspace,
"id": id,
});
if let Some(accession) = why {
body["why"] = serde_json::Value::String(accession.to_string());
}
let resp = match send(authorize(ureq::post(&url)), Some(&body), timeout()) {
Ok(resp) => resp,
Err(Attempt::Ureq(e)) if matches!(*e, ureq::Error::Status(404, _)) => {
return Err(Error::Bad(format!("no atom {id}")));
}
Err(e) => return Err(into_error(&url, e)),
};
Ok(resp.into_json()?)
}
/// Move one atom along the review clock: recalled, or lapsed.
pub fn grade(
&self,
workspace: &str,
id: &str,
recalled: bool,
) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/grade", self.base);
let body: serde_json::Value = send(
authorize(ureq::post(&url)),
Some(&serde_json::json!({
"workspace": workspace,
"id": id,
"recalled": recalled,
})),
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?;
Ok(body)
}
/// The link graph's communities, largest first.
/// The claims the link graph turns on, highest first.
pub fn hubs(&self, workspace: &str, limit: usize) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/hubs", self.base);
let body: serde_json::Value = send(
authorize(ureq::get(&url))
.query("workspace", workspace)
.query("limit", &limit.to_string()),
None,
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?;
Ok(body)
}
pub fn islands(&self, workspace: &str) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/islands", self.base);
let body: serde_json::Value = send(
authorize(ureq::get(&url)).query("workspace", workspace),
None,
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?;
Ok(body)
}
/// Claims that fired together: their links gain weight.
pub fn fire(&self, workspace: &str, ids: &[String]) -> Result<serde_json::Value, Error> {
self.fire_as(workspace, ids, None)
}
/// [`Self::fire`] through a persona's lens: the weights move under its
/// name and the seat's stand.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn fire_as(
&self,
workspace: &str,
ids: &[String],
lens: Option<&str>,
) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/fire", self.base);
let body: serde_json::Value = send(
authorize(ureq::post(&url)),
Some(
&serde_json::json!({"workspace": workspace, "ids": ids, "as": lens.unwrap_or("")}),
),
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?;
Ok(body)
}
/// Consolidate the workspace: every claim that replaces an earlier one
/// closes it (the write-time rule, run over what is held). `apply`
/// false reports the pairs and writes nothing.
pub fn consolidate(&self, workspace: &str, apply: bool) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/consolidate", self.base);
let body: serde_json::Value = send(
authorize(ureq::post(&url)),
Some(&serde_json::json!({"workspace": workspace, "apply": apply})),
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?;
Ok(body)
}
/// Sweep a workspace for neglect: reviews left due past twice their
/// interval lapse, and the third miss forgets a never-recalled lesson.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn sweep(&self, workspace: &str) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/sweep", self.base);
let body: serde_json::Value = send(
authorize(ureq::post(&url)),
Some(&serde_json::json!({"workspace": workspace})),
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?;
Ok(body)
}
/// Drop a scratch workspace's live atoms, no deed; its tombstones stay.
/// Retraction stays [`Self::delete_atom`]; this is what a per-run
/// scratch workspace calls on its way out.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn forget_workspace(&self, workspace: &str) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/forget", self.base);
let body: serde_json::Value = send(
authorize(ureq::post(&url)),
Some(&serde_json::json!({"workspace": workspace})),
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?;
Ok(body)
}
/// The memories a cue activates, strongest first; with `fire`, the top
/// of them fire together.
pub fn activate(
&self,
workspace: &str,
q: &str,
limit: u32,
fire: bool,
) -> Result<serde_json::Value, Error> {
self.activate_as(workspace, q, limit, fire, None)
}
/// [`Self::activate`] through a persona's lens: the spread follows the
/// weights that persona wrote, and a fire writes them.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn activate_as(
&self,
workspace: &str,
q: &str,
limit: u32,
fire: bool,
lens: Option<&str>,
) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/activate", self.base);
let mut req = authorize(ureq::get(&url))
.query("workspace", workspace)
.query("q", q)
.query("limit", &limit.to_string())
.query("fire", if fire { "1" } else { "0" });
if let Some(name) = lens.filter(|n| !n.trim().is_empty()) {
req = req.query("as", name);
}
let body: serde_json::Value = send(req, None, timeout())
.map_err(|e| into_error(&url, e))?
.into_json()?;
Ok(body)
}
/// The live atoms of one named set in a workspace: a persona's own
/// conclusions, a pinned slice.
///
/// # Errors
///
/// The request's, or a body that is not JSON.
pub fn atoms_in_set(
&self,
workspace: &str,
set: &str,
) -> Result<Vec<serde_json::Value>, Error> {
let url = format!("{}/v1/pack", self.base);
let body: serde_json::Value = send(
authorize(ureq::get(&url))
.query("workspace", workspace)
.query("set", set),
None,
timeout(),
)
.map_err(|e| into_error(&url, e))?
.into_json()?;
Ok(body
.get("atoms")
.and_then(serde_json::Value::as_array)
.cloned()
.unwrap_or_default())
}
pub fn post_atom(&self, atom: &serde_json::Value) -> Result<serde_json::Value, Error> {
let url = format!("{}/v1/atoms", self.base);
let body: serde_json::Value = send(authorize(ureq::post(&url)), Some(atom), timeout())
.map_err(|e| into_error(&url, e))?
.into_json()?;
Ok(body)
}
}
#[cfg(test)]
mod tests {
use super::*;
// Both workspace tests below rewrite process-global environment, so
// they hold this lock: without it the runner's parallel threads
// interleave one test's set-and-restore with the other's read, and
// the suite fails one run in five on a polluted read.
static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
#[test]
fn only_a_loopback_url_has_a_port_to_check() {
assert_eq!(loopback_port("http://127.0.0.1:8761/health"), Some(8761));
assert_eq!(loopback_port("http://localhost:18462"), Some(18462));
assert_eq!(loopback_port("http://[::1]:8761/v1/search?q=a"), Some(8761));
assert_eq!(loopback_port("http://127.0.0.1/health"), Some(80));
assert_eq!(loopback_port("http://example.org:8761/"), None);
assert_eq!(loopback_port("http://[::1]/"), Some(80));
}
#[cfg(target_os = "linux")]
#[test]
fn the_owner_of_a_listening_port_is_read_off_the_table() {
let table = " sl local_address rem_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode
0: 0100007F:223A 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1001 0 4242 1 0 100 0 0 10 0
1: 0100007F:223A 0100007F:9C40 01 00000000:00000000 00:00000000 00000000 1000 0 4343 1 0 20 4 30 10 -1
2: 0100007F:1F90 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 4444 1 0 100 0 0 10 0
";
assert_eq!(listen_uids(table, 8762), vec![1001], "only the listener");
assert_eq!(listen_uids(table, 8080), vec![1000]);
assert!(listen_uids(table, 9).is_empty());
}
#[cfg(target_os = "linux")]
#[test]
fn a_port_this_user_holds_is_not_foreign() {
let held = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = held.local_addr().unwrap().port();
assert!(!foreign_listener(&format!(
"http://127.0.0.1:{port}/health"
)));
drop(held);
assert!(
!foreign_listener(&format!("http://127.0.0.1:{port}/health")),
"nobody listening is not foreign"
);
}
#[test]
fn resolved_workspace_reads_ljos_env_not_default() {
let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let dir = std::env::temp_dir().join(format!("packset-ljos-env-{}", std::process::id()));
std::fs::create_dir_all(dir.join(".config/ljos")).unwrap();
std::fs::write(
dir.join(".config/ljos/env"),
"PACKSET_WORKSPACE=git:example.com/seat/notes\n",
)
.unwrap();
let old_home = env::var("HOME").ok();
let old_ws = env::var("PACKSET_WORKSPACE").ok();
unsafe {
env::remove_var("PACKSET_WORKSPACE");
env::set_var("HOME", &dir);
}
let got = resolved_workspace();
unsafe {
match old_home {
Some(h) => env::set_var("HOME", h),
None => env::remove_var("HOME"),
}
match old_ws {
Some(w) => env::set_var("PACKSET_WORKSPACE", w),
None => env::remove_var("PACKSET_WORKSPACE"),
}
}
assert_eq!(got, "git:example.com/seat/notes");
}
#[test]
fn resolved_workspace_without_env_is_seat_not_default() {
let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let dir = std::env::temp_dir().join(format!("packset-no-ljos-env-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let old_home = env::var("HOME").ok();
let old_ws = env::var("PACKSET_WORKSPACE").ok();
unsafe {
env::remove_var("PACKSET_WORKSPACE");
env::set_var("HOME", &dir);
}
let got = resolved_workspace();
unsafe {
match old_home {
Some(h) => env::set_var("HOME", h),
None => env::remove_var("HOME"),
}
match old_ws {
Some(w) => env::set_var("PACKSET_WORKSPACE", w),
None => env::remove_var("PACKSET_WORKSPACE"),
}
}
assert_eq!(got, "seat");
assert_ne!(got, "default");
}
/// A writer that answers busy to its first `busy` requests and counts every
/// request. `retry_after` is the `Retry-After` value, in seconds, when
/// the writer sends one.
fn busy_for(
busy: usize,
retry_after: Option<u64>,
) -> (
PacksetClient,
std::sync::Arc<std::sync::atomic::AtomicUsize>,
) {
use std::io::{BufRead, BufReader, Read, Write};
use std::sync::atomic::Ordering;
let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap();
let client = PacksetClient::new(format!("http://{}", listener.local_addr().unwrap()));
let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let counted = std::sync::Arc::clone(&seen);
std::thread::spawn(move || {
for stream in listener.incoming().flatten() {
let mut reader = BufReader::new(&stream);
let mut length = 0;
let mut line = String::new();
while reader.read_line(&mut line).is_ok_and(|n| n > 2) {
if let Some(v) = line.to_ascii_lowercase().strip_prefix("content-length:") {
length = v.trim().parse().unwrap_or(0);
}
line.clear();
}
let mut body = vec![0u8; length];
let _ = reader.read_exact(&mut body);
let (status, text) = if counted.fetch_add(1, Ordering::SeqCst) < busy {
("503 Service Unavailable", r#"{"error":"busy"}"#)
} else {
("200 OK", r#"{"ok":true}"#)
};
let retry = retry_after
.map(|secs| format!("Retry-After: {secs}\r\n"))
.unwrap_or_default();
let _ = (&stream).write_all(
format!(
"HTTP/1.1 {status}\r\n{retry}Content-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}",
text.len()
)
.as_bytes(),
);
}
});
(client, seen)
}
#[test]
fn a_busy_writer_is_asked_again_for_a_read_and_a_write() {
use std::sync::atomic::Ordering;
let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let (client, seen) = busy_for(2, None);
assert_eq!(client.status(None).unwrap()["ok"], true);
assert_eq!(seen.load(Ordering::SeqCst), 3);
let (client, seen) = busy_for(1, None);
let stored = client
.post_atom(&serde_json::json!({"text": "one write", "workspace": "w"}))
.unwrap();
assert_eq!(stored["ok"], true);
assert_eq!(
seen.load(Ordering::SeqCst),
2,
"the write was not sent again"
);
}
#[test]
fn a_writer_that_stays_busy_fails_in_one_sentence() {
use std::sync::atomic::Ordering;
let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let old = env::var_os("PACKSET_TIMEOUT_MS");
unsafe { env::set_var("PACKSET_TIMEOUT_MS", "700") };
let (client, seen) = busy_for(usize::MAX, None);
let err = client.status(None).unwrap_err().to_string();
unsafe {
match old {
Some(value) => env::set_var("PACKSET_TIMEOUT_MS", value),
None => env::remove_var("PACKSET_TIMEOUT_MS"),
}
}
assert_eq!(err, "packsetd is busy.", "{err}");
assert!(
!err.contains("os error") && !err.contains("status code"),
"{err}"
);
// The count is the waits that fit in 700 ms, not how long this host
// took to read each 503.
assert_eq!(
seen.load(Ordering::SeqCst),
usize::try_from(attempts_in(Duration::from_millis(700))).unwrap(),
"{err}"
);
}
#[test]
fn a_refused_connection_is_retried_and_then_one_sentence() {
use std::time::Instant;
let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let dir = std::env::temp_dir().join(format!("packset-down-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let saved = save_env(&[
"HOME",
"PACKSET_HOME",
"LJOS_HOME",
"LJOS_INSIDE_MEMORY_HOME",
"PACKSET_BIN",
"PACKSET_TIMEOUT_MS",
]);
unsafe {
env::set_var("HOME", &dir);
env::remove_var("PACKSET_HOME");
env::remove_var("LJOS_HOME");
env::remove_var("LJOS_INSIDE_MEMORY_HOME");
env::remove_var("PACKSET_BIN");
env::set_var("PACKSET_TIMEOUT_MS", "250");
}
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
drop(listener);
let client = PacksetClient::new(format!("http://127.0.0.1:{port}"));
let started = Instant::now();
let err = client.status(None).unwrap_err().to_string();
let took = started.elapsed();
restore_env(saved);
let _ = std::fs::remove_dir_all(&dir);
assert_eq!(err, "packsetd is not running.", "{err}");
assert!(took >= Duration::from_millis(40), "no backoff: {took:?}");
assert!(took < Duration::from_secs(2), "{took:?}");
}
#[test]
fn a_down_local_writer_is_started_once() {
let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let dir = std::env::temp_dir().join(format!(
"packset-autostart-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let home = dir.join("home");
std::fs::create_dir_all(&home).unwrap();
let log = dir.join("starts");
let stub = dir.join("packsetd-stub");
std::fs::write(
&stub,
format!(
"#!/bin/sh\necho start >> \"{}\"\necho pid $$ >> \"{}\"\nexec python3 - \"$@\" << 'PY'\nimport sys\nargs = sys.argv[1:]\nport = int(args[args.index('--port') + 1])\nhome = args[args.index('--home') + 1]\nfrom http.server import BaseHTTPRequestHandler, HTTPServer\nclass Handler(BaseHTTPRequestHandler):\n def do_GET(self):\n body = b'{{\"ok\":true}}'\n self.send_response(200)\n self.send_header('Content-Type', 'application/json')\n self.send_header('Content-Length', str(len(body)))\n self.end_headers()\n self.wfile.write(body)\n def log_message(self, *args):\n pass\nHTTPServer(('127.0.0.1', port), Handler).serve_forever()\nPY\n",
log.display(),
log.display(),
),
)
.unwrap();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut mode = std::fs::metadata(&stub).unwrap().permissions();
mode.set_mode(0o755);
std::fs::set_permissions(&stub, mode).unwrap();
}
let saved = save_env(&[
"HOME",
"PACKSET_HOME",
"LJOS_HOME",
"LJOS_INSIDE_MEMORY_HOME",
"PACKSET_BIN",
"PACKSET_TIMEOUT_MS",
"PACKSET_TOKEN",
]);
unsafe {
env::set_var("HOME", &dir);
env::set_var("PACKSET_HOME", &home);
env::remove_var("LJOS_HOME");
env::remove_var("LJOS_INSIDE_MEMORY_HOME");
env::remove_var("PACKSET_TOKEN");
env::set_var("PACKSET_BIN", &stub);
env::set_var("PACKSET_TIMEOUT_MS", "5000");
}
let reap = Reap(log.clone());
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
drop(listener);
let client = PacksetClient::new(format!("http://127.0.0.1:{port}"));
let body = client.status(None).expect("the stub should answer");
let text = std::fs::read_to_string(&log).unwrap_or_default();
drop(reap);
restore_env(saved);
let _ = std::fs::remove_dir_all(&dir);
assert_eq!(body["ok"], true, "{body}");
assert_eq!(
text.lines().filter(|line| *line == "start").count(),
1,
"{text}"
);
}
fn save_env(keys: &[&str]) -> Vec<(String, Option<std::ffi::OsString>)> {
keys.iter()
.map(|key| ((*key).to_string(), env::var_os(key)))
.collect()
}
fn restore_env(saved: Vec<(String, Option<std::ffi::OsString>)>) {
for (key, value) in saved {
unsafe {
match value {
Some(value) => env::set_var(key, value),
None => env::remove_var(key),
}
}
}
}
/// Kills every pid the autostart stub wrote, including when the test fails.
struct Reap(std::path::PathBuf);
impl Drop for Reap {
fn drop(&mut self) {
let Ok(text) = std::fs::read_to_string(&self.0) else {
return;
};
for line in text.lines() {
let Some(pid) = line.strip_prefix("pid ") else {
continue;
};
let _ = std::process::Command::new("kill")
.args(["-9", pid.trim()])
.status();
}
}
}
/// Room for the first wait and not the second: 50 ms fits in 100 ms,
/// and 50 + 150 ms does not.
const BUDGET: Duration = Duration::from_millis(100);
/// A 100 ms budget leaves no room for the 150 ms wait. The time handed
/// in is the waits already taken. A round trip does not move it.
#[test]
fn the_retries_stop_where_the_budget_does() {
let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let old = env::var_os("PACKSET_TIMEOUT_MS");
unsafe { env::set_var("PACKSET_TIMEOUT_MS", BUDGET.as_millis().to_string()) };
let budget = timeout();
unsafe {
match old {
Some(v) => env::set_var("PACKSET_TIMEOUT_MS", v),
None => env::remove_var("PACKSET_TIMEOUT_MS"),
}
}
assert_eq!(budget, BUDGET);
let second = BUSY_WAIT * BUSY_GROWTH;
assert!(
ask_again(Duration::ZERO, BUSY_WAIT, budget),
"the first wait did not fit"
);
assert!(!ask_again(BUSY_WAIT, second, budget), "the second wait fit");
assert!(
!ask_again(budget - BUSY_WAIT, BUSY_WAIT, budget),
"a first round trip that spent the room was followed by a wait"
);
}
/// 50 ms fits in 51 and not in 50. 50 + 150 + 450 fits in 700, and the
/// next wait, 1350 ms, does not. The same numbers on every host.
#[test]
fn the_attempt_count_is_the_waits_that_fit_the_budget() {
assert_eq!(attempts_in(Duration::from_millis(50)), 1);
assert_eq!(attempts_in(Duration::from_millis(51)), 2);
assert_eq!(attempts_in(Duration::from_millis(100)), 2);
assert_eq!(attempts_in(Duration::from_millis(700)), 4);
assert_eq!(attempts_in(Duration::from_secs(30)), 10);
}
/// `Retry-After` longer than the budget is not waited. The shorter
/// backoff is not used in its place.
#[test]
fn a_retry_after_past_the_budget_is_not_waited() {
let budget = Duration::from_millis(700);
assert!(pause_for(
BUSY_WAIT,
Some(Duration::from_secs(1)),
Duration::ZERO,
budget,
0
)
.is_none());
let inside = pause_for(BUSY_WAIT, None, Duration::ZERO, budget, 0).unwrap();
assert_eq!(inside.charge, BUSY_WAIT);
assert!(inside.sleep >= BUSY_WAIT / 2 && inside.sleep <= BUSY_WAIT);
let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let old = env::var_os("PACKSET_TIMEOUT_MS");
unsafe { env::set_var("PACKSET_TIMEOUT_MS", "700") };
let (client, seen) = busy_for(usize::MAX, Some(1));
let err = client.status(None).unwrap_err().to_string();
unsafe {
match old {
Some(value) => env::set_var("PACKSET_TIMEOUT_MS", value),
None => env::remove_var("PACKSET_TIMEOUT_MS"),
}
}
assert_eq!(err, "packsetd is busy.", "{err}");
assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "{err}");
}
/// One second fits in a three-second budget, so the second ask waits
/// for it. Jitter keeps the wait under that second and over half of it.
#[test]
fn a_retry_after_inside_the_budget_is_waited() {
use std::time::Instant;
let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let old = env::var_os("PACKSET_TIMEOUT_MS");
unsafe { env::set_var("PACKSET_TIMEOUT_MS", "3000") };
let (client, seen) = busy_for(1, Some(1));
let started = Instant::now();
let body = client.status(None).unwrap();
let took = started.elapsed();
unsafe {
match old {
Some(value) => env::set_var("PACKSET_TIMEOUT_MS", value),
None => env::remove_var("PACKSET_TIMEOUT_MS"),
}
}
assert_eq!(body["ok"], true, "{body}");
assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 2);
assert!(
took >= Duration::from_millis(400),
"Retry-After was not waited: {took:?}"
);
assert!(took < Duration::from_secs(3), "{took:?}");
}
#[test]
fn a_request_carries_the_end_of_its_budget() {
use std::io::{BufRead, BufReader, Write};
let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap();
let client = PacksetClient::new(format!("http://{}", listener.local_addr().unwrap()));
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let (stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(&stream);
let mut deadline = String::new();
let mut line = String::new();
while reader.read_line(&mut line).is_ok_and(|n| n > 2) {
let lower = line.to_ascii_lowercase();
if let Some(value) = lower.strip_prefix("x-packset-deadline:") {
deadline = value.trim().to_string();
}
line.clear();
}
let _ = tx.send(deadline);
let text = r#"{"ok":true}"#;
let _ = (&stream).write_all(
format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}",
text.len()
)
.as_bytes(),
);
});
let before = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis();
let body = client.status(None).unwrap();
let header = rx.recv_timeout(Duration::from_secs(2)).unwrap();
let stamp: u128 = header.parse().unwrap();
assert_eq!(body["ok"], true, "{body}");
assert!(stamp > before, "{stamp} <= {before}");
assert!(stamp < before + 60_000, "{stamp} is not this budget");
}
}