johnandersen777.bsky.social/atlibs-socialweb-computer-requester
socialweb.computer market requester and policy modules for atlibs, with its web pages and lexicons
0
0
// The market side: offerings -> request (vm -> rfp -> submitRfp) -> bids -> policy -> accept.
// No submitBid: this requester serves no XRPC, so bids are read from bidder repos.
use abi::records::temp::compute::Vm;
use abi::records::temp::market::{Accept, EvaluatePolicy, Rfp, SubmitAccept, SubmitRfp, evaluate_policy};
use abi::repo::create_record;
use abi::{AtUri, Call, Fan, Handle, LexiconType, On, Run, TypedStrongRef};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json};
use crate::calls::{attest, bid_uri, created, read, signed, typed};
use crate::ids::{CallId, Mode};
use crate::input::{Args, MAX_PEERS, Out, Route, Step, bidders, did, list, optional_route, peer, routes, strong};
use crate::nsid::{BID, BIDS_FREE, COMPUTE_VM, EVALUATE_POLICY, OFFERING, STRONG_REF, SUBMIT_ACCEPT, SUBMIT_RFP, USER_DATA};
const DEFAULT_DISK: &str = "50G";
/// compute.vm's mem and disk are sizes, "<n>M" or "<n>G" (a bidder refuses anything else after it
/// accepts); a number is GiB.
/// An atproto datetime: YYYY-MM-DDTHH:MM:SS, optional fraction, then Z or ±HH:MM.
fn datetime(t: &str) -> bool {
let b = t.as_bytes();
let digits = |r: std::ops::Range<usize>| r.clone().all(|i| b.get(i).is_some_and(u8::is_ascii_digit));
let head = b.len() >= 20 && digits(0..4) && b[4] == b'-' && digits(5..7) && b[7] == b'-' && digits(8..10) && b[10] == b'T' && digits(11..13) && b[13] == b':' && digits(14..16) && b[16] == b':' && digits(17..19);
let mut i = 19;
if head && b[i] == b'.' {
i += 1;
while b.get(i).is_some_and(u8::is_ascii_digit) {
i += 1;
}
}
head && (&t[i..] == "Z" || (t.len() == i + 6 && (b[i] == b'+' || b[i] == b'-') && digits(i + 1..i + 3) && b[i + 3] == b':' && digits(i + 4..i + 6)))
}
pub fn size(value: &Value, key: &str) -> Result<Value, String> {
let text = match value {
Value::Number(n) => n.as_u64().filter(|n| *n > 0).map(|n| format!("{n}G")),
Value::String(s) => s.strip_suffix(['M', 'G']).filter(|n| !n.is_empty() && n.bytes().all(|b| b.is_ascii_digit()) && n.len() <= 7).map(|_| s.clone()),
_ => None,
};
text.map(Value::String).ok_or(format!("{key} must be a size like 4G or 512M, or a number of GiB"))
}
// ---- offerings: bidder endpoints (old: runComputeContract step 4) ----------------------------
#[derive(Default, Serialize, Deserialize)]
pub struct Offerings {
bidders: Vec<String>,
reads: Fan,
}
pub fn offers_compute(offering: &Value) -> bool {
offering["appliesTo"].as_array().is_some_and(|a| a.contains(&json!(COMPUTE_VM))) && offering["endpoints"].as_array().is_some_and(|e| !e.is_empty()) && offering["withdrawnAt"].is_null()
}
/// Each bidder's compute offering (getRecord at self) as { did, endpoints }; pass them as request's bidders.
pub fn offerings(run: Run<Args>) -> Result<Handle, String> {
let (mut step, on) = Step::<Offerings>::new(run)?;
step.args.signed_in()?;
match on {
On::Start => {
let bidders = bidders(&step.args)?;
let reads = bidders.iter().enumerate().map(|(i, bidder)| (CallId::Offerings(i), read(&AtUri::new(bidder, OFFERING, "self"))));
step.out.fan_out(&mut step.state.reads, reads);
step.state.bidders = bidders;
}
On::Result(id @ CallId::Offerings(_), result) => {
let Some(results) = step.state.reads.land(&id, result) else { return step.handle() };
let (mut endpoints, mut errors) = (Vec::<Value>::new(), Vec::new());
for (bidder, result) in step.state.bidders.iter().zip(results) {
match result {
Ok(output) => {
if let Some(peer) = Some(&output["value"]).filter(|v| offers_compute(v)).and_then(|v| peer(bidder, v)) {
if !endpoints.iter().any(|e| e["did"] == peer["did"]) {
endpoints.push(peer);
}
}
}
Err(error) => errors.push(json!({ "bidder": bidder, "error": error })),
}
}
step.out.finish(json!({ "offerings": endpoints, "errors": errors }));
}
_ => {}
}
step.handle()
}
// ---- request: vm -> rfp -> submitRfp ------------------------------------------------------------
/// [userData hook ->] createRecord vm -> attest rfp -> createRecord rfp, the rfp naming the vm, the input's
/// policies and onBehalfOf. Records it creates go on `kept`. No user_data: the hook builds it from shell,
/// iroh, webrtc_ssh, ssh_key and tunnels (default: the user-data module).
#[derive(Default, Serialize, Deserialize)]
#[serde(default, rename_all = "camelCase")]
pub struct RfpChain {
policies: Vec<TypedStrongRef>,
on_behalf_of: Option<TypedStrongRef>,
unsigned: Value,
pub user_data: String,
pub vm: Option<TypedStrongRef>,
pub rfp: Option<TypedStrongRef>,
}
impl RfpChain {
// disk defaults like socialweb-computer-kcp compute-request-xrpc (DEFAULT_VM_DISK).
pub fn start(args: &Args, on_behalf_of: Option<TypedStrongRef>, out: &mut Out) -> Result<Self, String> {
let policies = match args.arg("policies") {
Value::Null => vec![],
Value::Array(list) => list.iter().map(|p| strong(p, "policies")).collect::<Result<_, _>>()?,
_ => return Err("policies must be a list of strong refs".into()),
};
let mut chain = Self { policies, on_behalf_of, ..Self::default() };
match args.arg("user_data") {
Value::Null => {
let input: Map<String, Value> = ["shell", "iroh", "webrtc_ssh", "ssh_key", "tunnels"].into_iter().filter(|k| !args.arg(k).is_null()).map(|k| (k.to_string(), args.arg(k).clone())).collect();
out.call(CallId::UserData, Call::new("", USER_DATA).input(Value::Object(input)));
}
value => {
chain.user_data = value.as_str().filter(|s| !s.trim().is_empty()).ok_or("user_data must be cloud-init text, or omitted")?.into();
chain.create_vm(args, out)?;
}
}
Ok(chain)
}
fn create_vm(&self, args: &Args, out: &mut Out) -> Result<(), String> {
let role = args.arg("role").as_str().map(str::trim).filter(|s| !s.is_empty()).unwrap_or("compute");
// what the bidder runs, always said: a bidder serving vm and deno must not have to guess
let kind = match args.arg("kind").as_str().unwrap_or("vm") {
kind @ ("vm" | "deno") => kind,
_ => return Err("kind must be vm or deno".into()),
};
let given = |key: &str| Some(args.arg(key).clone()).filter(|v| !v.is_null());
let cpus = given("cpus").map(|c| c.as_u64().filter(|n| (1..=256).contains(n)).map(|n| json!(n)).ok_or("cpus must be an integer, 1 to 256")).transpose()?;
let vm = Vm {
cpus,
mem: given("mem").map(|m| size(&m, "mem")).transpose()?,
disk: given("disk").map_or(Ok(json!(DEFAULT_DISK)), |d| size(&d, "disk"))?,
network: given("network"),
location: None,
role: role.into(),
user_data: self.user_data.clone(),
created_at: args.created_at()?,
};
let mut record = typed(vm);
record["kind"] = json!(kind);
// when the bidder ends the VM (it ends one without, a day after the accept)
if let Some(at) = given("expires_at") {
record["expiresAt"] = json!(at.as_str().filter(|t| datetime(t)).ok_or("expires_at must be an RFC 3339 datetime, e.g. 2026-11-10T22:00:00Z")?);
}
out.call(CallId::Vm, create_record(args.me(), Vm::lexicon_type(), None, record));
Ok(())
}
pub fn step(&mut self, args: &Args, id: CallId, result: Result<Value, String>, out: &mut Out, kept: &mut Vec<String>) -> Result<Option<TypedStrongRef>, String> {
match id {
CallId::UserData => {
let output = result.map_err(|e| format!("{USER_DATA}: {e}"))?;
let user_data = output["user_data"].as_str().filter(|s| !s.trim().is_empty());
self.user_data = user_data.ok_or(format!("user_data is required: no {USER_DATA} hook built it"))?.into();
self.create_vm(args, out)?;
}
CallId::Vm => {
let vm = created(result)?;
kept.push(vm.uri.clone());
let rfp = Rfp {
domain: "compute".into(),
payload: vm.clone(),
submit_bid: None,
policies: Some(self.policies.clone()).filter(|p| !p.is_empty()),
on_behalf_of: self.on_behalf_of.clone(),
created_at: args.created_at()?,
};
self.unsigned = typed(rfp);
out.call(CallId::AttestRfp, attest(args.me(), Some(&self.unsigned)));
self.vm = Some(vm);
}
CallId::AttestRfp => out.call(CallId::Rfp, create_record(args.me(), Rfp::lexicon_type(), None, signed(result, &self.unsigned)?)),
CallId::Rfp => {
let rfp = created(result)?;
kept.push(rfp.uri.clone());
self.rfp = Some(rfp.clone());
return Ok(Some(rfp));
}
_ => {}
}
Ok(None)
}
}
pub fn submit_rfp(rfp: &TypedStrongRef) -> SubmitRfp {
SubmitRfp { rfp_uri: rfp.uri.clone(), rfp_cid: rfp.cid.clone() }
}
#[derive(Default, Serialize, Deserialize)]
pub struct Request {
bidders: Vec<Route>,
chain: RfpChain,
submits: Fan,
}
pub fn request(run: Run<Args>) -> Result<Handle, String> {
let (mut step, on) = Step::<Request>::new(run)?;
step.args.signed_in()?;
match on {
On::Start => {
let bidders = routes(step.args.arg("bidders"), "bidders")?;
let on_behalf_of = match step.args.arg("onBehalfOf") {
Value::Null => None,
value => Some(strong(value, "onBehalfOf")?),
};
step.state.chain = RfpChain::start(&step.args, on_behalf_of, &mut step.out)?;
step.state.bidders = bidders;
}
On::Result(id @ (CallId::UserData | CallId::Vm | CallId::AttestRfp | CallId::Rfp), result) => {
let Request { bidders, chain, submits } = &mut step.state;
let Some(rfp) = chain.step(&step.args, id, result, &mut step.out, &mut vec![])? else { return step.handle() };
if bidders.is_empty() {
step.out.finish(json!({ "vm": chain.vm, "rfp": rfp, "submitted": [] }));
} else {
let input = submit_rfp(&rfp);
step.out.fan_out(submits, bidders.iter().enumerate().map(|(i, bidder)| (CallId::SubmitRfp(i), bidder.call(SUBMIT_RFP, &input))));
}
}
On::Result(id @ CallId::SubmitRfp(_), result) => {
let Request { bidders, chain, submits } = &mut step.state;
let Some(results) = submits.land(&id, result) else { return step.handle() };
let submitted: Vec<Value> = bidders.iter().zip(results).map(|(bidder, result)| match result {
Ok(output) => json!({ "bidder": bidder, "ok": true, "output": output }),
Err(error) => json!({ "bidder": bidder, "ok": false, "error": error }),
}).collect();
step.out.finish(json!({ "vm": chain.vm, "rfp": chain.rfp, "submitted": submitted }));
}
_ => {}
}
step.handle()
}
// ---- bids ---------------------------------------------------------------------------------------
#[derive(Default, Serialize, Deserialize)]
pub struct Bids {
rfp: String,
bidders: Vec<String>,
reads: Fan,
}
/// A free bid's payload is (a strong ref to) a market.bids.free.
pub fn is_free(payload: &Value) -> bool {
let nsid = match payload["$type"].as_str() {
Some(nsid) if nsid != STRONG_REF => Some(nsid.to_string()),
_ => payload["uri"].as_str().and_then(|u| u.parse::<AtUri>().ok()).map(|u| u.collection),
};
nsid.as_deref() == Some(BIDS_FREE)
}
/// A bid URI (at://did/market.bid/rkey, rkey = submitRfp's `name`) is read exactly; a bidder DID's
/// bid is read at the key it gives a bid on this rfp {uri, cid} (bid_uri).
/// Winner as requester-xrpc picks it: the first free bid, else the lowest payload.cost.
pub fn bids(run: Run<Args>) -> Result<Handle, String> {
let (mut step, on) = Step::<Bids>::new(run)?;
step.args.signed_in()?;
match on {
On::Start => {
let rfp_ref = strong(step.args.arg("rfp"), "rfp")?;
let rfp = rfp_ref.uri.clone();
let sources: Vec<String> = list(step.args.arg("bidders"), "bidders")?.into_iter().chain(list(step.args.arg("bids"), "bids")?).collect();
if sources.is_empty() {
return Err("bidders or bids is required".into());
}
if sources.len() > MAX_PEERS {
return Err(format!("at most {MAX_PEERS} bidders and bids"));
}
let (mut reads, mut bidders) = (Vec::new(), Vec::new());
for (i, source) in sources.iter().enumerate() {
let (bidder, call) = if source.starts_with("at://") {
let uri = source.parse::<AtUri>().ok().filter(|u| u.collection == BID).ok_or("bids must be market.bid at:// URIs")?;
(did(&json!(uri.repo), "bids")?, read(&uri))
} else {
let bidder = did(&json!(source), "bidders")?;
let call = read(&bid_uri(&bidder, &rfp_ref));
(bidder, call)
};
reads.push((CallId::Bid(i), call));
bidders.push(bidder);
}
step.out.fan_out(&mut step.state.reads, reads);
step.state.rfp = rfp;
step.state.bidders = bidders;
}
On::Result(id @ CallId::Bid(_), result) => {
let Bids { rfp, bidders, reads } = &mut step.state;
let Some(results) = reads.land(&id, result) else { return step.handle() };
let (mut bids, mut errors) = (Vec::new(), Vec::new());
for (bidder, result) in bidders.iter().zip(results) {
let output = match result {
Ok(output) => output,
Err(error) => {
errors.push(json!({ "bidder": bidder, "error": error }));
continue;
}
};
for record in [&output].into_iter().filter(|r| r["value"]["rfp"]["uri"] == **rfp) {
let payload = &record["value"]["payload"];
bids.push(json!({
"bidder": bidder,
"uri": record["uri"],
"cid": record["cid"],
"free": is_free(payload),
"cost": payload["cost"],
"submitAccept": record["value"]["submitAccept"],
}));
}
}
let cost = |b: &Value| b["cost"].as_f64().unwrap_or(f64::INFINITY);
let winner = bids.iter().find(|b| b["free"] == true).or_else(|| bids.iter().reduce(|best, b| if cost(b) < cost(best) { b } else { best })).cloned();
step.out.finish(json!({ "rfp": rfp, "bids": bids, "winner": winner, "errors": errors }));
}
_ => {}
}
step.handle()
}
// ---- policy: the hook (module.json hooks), kcp's Verdict for a counterparty ------
/// The hook's input (kcp participant.PolicyInput, the requester's perspective): scope mode names the
/// bidder and the rfp (kcp accept.go bidAllowedWithPolicy), full mode adds the bid (runner.go:547).
pub fn evaluate_policy(me: &str, mode: Mode, rfp: &TypedStrongRef, bidder: &str, bid: Option<evaluate_policy::Record>) -> Call {
let input = EvaluatePolicy {
self_did: me.into(),
perspective: Some("requester".into()),
mode: Some(mode.to_string()),
counterparty_did: Some(bidder.into()),
subject_did: Some(bidder.into()),
root_requester_did: Some(me.into()),
rfp: Some(evaluate_policy::Record { uri: rfp.uri.clone(), cid: Some(rfp.cid.clone()), value: None }),
bid,
accept: None,
};
Call::new("", EVALUATE_POLICY).input(json!(input))
}
/// Anything but allow: true is a deny, with its messages; a failed call is one too, and null (a hook
/// nothing serves: the record lost its default) is "no verdict" (fail closed).
pub fn denied(verdict: &Result<Value, String>) -> Option<String> {
match verdict {
Ok(verdict) if verdict["allow"] == true => None,
Ok(verdict) => {
let msgs: Vec<&str> = verdict["violations"].as_array().into_iter().flatten().filter_map(|v| v["msg"].as_str()).collect();
Some(if msgs.is_empty() { "no verdict".into() } else { msgs.join("; ") })
}
Err(error) => Some(error.clone()),
}
}
// ---- accept: attest -> accept -> submitAccept ---------------------------------------------------
#[derive(Clone, Serialize, Deserialize)]
pub struct Receipt {
pub uri: String,
pub cid: Value,
#[serde(rename = "submitEvent")]
pub submit_event: Value,
}
impl Receipt {
pub fn of(record: &Value) -> Self {
Receipt { uri: record["uri"].as_str().unwrap_or_default().into(), cid: record["cid"].clone(), submit_event: record["value"]["submitEvent"].clone() }
}
}
#[derive(Serialize)]
pub struct Accepted {
pub accept: TypedStrongRef,
pub receipt: Option<Receipt>,
#[serde(skip_serializing_if = "Option::is_none")]
pub submitted: Option<Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
}
/// attest accept -> createRecord accept -> submitAccept. Without submitAccept the bidder finds the
/// accept on the firehose (read the receipt later); the accept exists by submitAccept, so a failed
/// one is reported, not raised.
#[derive(Default, Serialize, Deserialize)]
#[serde(default, rename_all = "camelCase")]
pub struct AcceptChain {
unsigned: Value,
submit_accept: Option<Route>,
pub accept: Option<TypedStrongRef>,
}
impl AcceptChain {
pub fn start(args: &Args, rfp: TypedStrongRef, bid: TypedStrongRef, submit_accept: Option<Route>, out: &mut Out) -> Result<Self, String> {
let unsigned = typed(Accept { rfp, bid, payload: None, submit_event: None, created_at: args.created_at()? });
out.call(CallId::AttestAccept, attest(args.me(), Some(&unsigned)));
Ok(Self { unsigned, submit_accept, accept: None })
}
pub fn step(&mut self, args: &Args, id: CallId, result: Result<Value, String>, out: &mut Out, kept: &mut Vec<String>) -> Result<Option<Accepted>, String> {
let accepted = |accept: &TypedStrongRef| Accepted { accept: accept.clone(), receipt: None, submitted: None, error: None };
match id {
CallId::AttestAccept => out.call(CallId::Accept, create_record(args.me(), Accept::lexicon_type(), None, signed(result, &self.unsigned)?)),
CallId::Accept => {
let accept = created(result)?;
kept.push(accept.uri.clone());
self.accept = Some(accept.clone());
let Some(route) = &self.submit_accept else { return Ok(Some(accepted(&accept))) };
out.call(CallId::SubmitAccept, route.call(SUBMIT_ACCEPT, SubmitAccept { accept_uri: accept.uri.clone(), accept_cid: accept.cid.clone() }));
}
CallId::SubmitAccept => {
let accept = accepted(self.accept.as_ref().ok_or("bad state")?);
return Ok(Some(match result {
// kcp bidders answer {ok, id} and mint the receipt later: find it with `receipt`.
Ok(output) if !output["uri"].is_string() => Accepted { submitted: Some(output), ..accept },
Ok(output) => Accepted { receipt: Some(Receipt { uri: output["uri"].as_str().unwrap_or_default().into(), cid: output["cid"].clone(), submit_event: output["submitEvent"].clone() }), ..accept },
Err(error) => Accepted { error: Some(error), ..accept },
}));
}
_ => {}
}
Ok(None)
}
}
#[derive(Default, Serialize, Deserialize)]
#[serde(default, rename_all = "camelCase")]
pub struct AcceptBid {
rfp: Option<TypedStrongRef>,
bid: Option<TypedStrongRef>,
submit_accept: Option<Route>,
chain: AcceptChain,
}
/// The policy hook in full mode first: the requester accepts no bid its policy denies (kcp runner.go:547).
pub fn accept(run: Run<Args>) -> Result<Handle, String> {
let (mut step, on) = Step::<AcceptBid>::new(run)?;
step.args.signed_in()?;
match on {
On::Start => {
let rfp = strong(step.args.arg("rfp"), "rfp")?;
let bid = strong(step.args.arg("bid"), "bid")?;
let submit_accept = optional_route(step.args.arg("submitAccept"), "submitAccept")?;
let bidder = bid.uri.parse::<AtUri>().map(|u| u.repo).unwrap_or_default();
let record = evaluate_policy::Record { uri: bid.uri.clone(), cid: Some(bid.cid.clone()), value: None };
step.out.call(CallId::AcceptPolicy, evaluate_policy(step.args.me(), Mode::Full, &rfp, &bidder, Some(record)));
step.state = AcceptBid { rfp: Some(rfp), bid: Some(bid), submit_accept, chain: AcceptChain::default() };
}
On::Result(CallId::AcceptPolicy, verdict) => {
let AcceptBid { rfp, bid, submit_accept, chain } = &mut step.state;
let (rfp, bid) = (rfp.clone().ok_or("bad state")?, bid.clone().ok_or("bad state")?);
if let Some(why) = denied(&verdict) {
return Err(format!("bid {} denied by policy (full): {why}", bid.uri));
}
*chain = AcceptChain::start(&step.args, rfp, bid, submit_accept.take(), &mut step.out)?;
}
On::Result(id, result) => {
if let Some(accepted) = step.state.chain.step(&step.args, id, result, &mut step.out, &mut vec![])? {
step.out.finish(accepted);
}
}
_ => {}
}
step.handle()
}