~ knotless

chadtmiller.com/knotless

Git hosted in your atproto account, on any stock PDS, fast enough for agents.

git clone knotless://chadtmiller.com/knotless

needs the knotless git helper. set it up

Push: blob parts no larger than the PDS says it takes

upload reads blobUploadLimit from describeServer, cached as identities are,
and starts at the smaller of it and KNOTLESS_CHUNK_SIZE. A PDS under 4 MB
no longer costs a refused round per push, each re-sending the whole pack.
Halving on a refusal stays, for a limit that is not what is in front of
the PDS.

examples/chunks measures what the part size costs over a shaped link;
harness/pds.mjs takes --blob-limit for it. 4 MB stays the default: larger
parts were no faster, and one 64 MiB part was slower at 20 Mbit/s.

package-lock.json matches package.json's pinned playwright again.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Chad Miller and Claude Opus 5.5 committed on Oct 21, 2026 parent 58a9176 commit 8e2fff1
Showing 8 changed files with 300 additions and 13 deletions
README.md+2−1
@@ -328,7 +328,8 @@ its page with), `KNOTLESS_SERVICE`
(skip DID resolution), `KNOTLESS_PLC_URL`, `KNOTLESS_HANDLE_RESOLVER`,
`KNOTLESS_APPVIEW`, `KNOTLESS_MODERATOR` (the appview's),
`KNOTLESS_ADMIN_PASSWORD`, `KNOTLESS_DID_WEB_SCHEME` (`https`),
`KNOTLESS_INLINE_MAX` (192 KiB), `KNOTLESS_CHUNK_SIZE` (4 MB),
`KNOTLESS_INLINE_MAX` (192 KiB), `KNOTLESS_CHUNK_SIZE` (4 MB, or the PDS's
blob limit if smaller),
`KNOTLESS_COMPACT` (`background`, `inline`, `off`),
`KNOTLESS_COMPACT_THRESHOLD` (32), `KNOTLESS_COMPACT_GRACE_SECS` (600),
`KNOTLESS_CLAIM_GRACE_SECS` (86400),
crates/knotless/examples/chunks.rsnew+206−0
@@ -0,0 +1,206 @@
//! What the part size costs a large push and its clone, over a link with a
//! chosen round-trip time and bandwidth.
//!
//! node harness/pds.mjs --accounts alice --blob-limit 100000000 > target/chunks-net.json
//! cargo build --release
//! KNOTLESS_TEST_NET=target/chunks-net.json cargo run --release --example chunks -- \
//! --rtt 50 --mbit 0,100,20 --parts 4,16,64 --pack 64 -n 3
//!
//! The PDS needs a blob limit above the largest part: the stock 5 MB
//! refuses anything larger, and push would halve its way down.
#[path = "../tests/common/mod.rs"]
mod common;
use common::*;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
/// One direction of the link, shared by every connection, as an uplink is:
/// bytes leave one after another at `bits` per second.
struct Link {
free_at: Mutex<Instant>,
bits: AtomicU64,
}
impl Link {
fn new() -> Arc<Link> {
Arc::new(Link {
free_at: Mutex::new(Instant::now()),
bits: AtomicU64::new(u64::MAX),
})
}
/// When `n` bytes read now have finished crossing.
fn send(&self, n: usize) -> Instant {
let bits = self.bits.load(Ordering::Relaxed);
let mut free_at = self.free_at.lock().unwrap();
let start = (*free_at).max(Instant::now());
*free_at = start + Duration::from_secs_f64(n as f64 * 8.0 / bits as f64);
*free_at
}
}
/// Forwards to `upstream` over a link `rtt` long each way round, at the
/// bandwidth `up` and `down` are set to.
async fn shaped_proxy(upstream: String, rtt: Duration, up: Arc<Link>, down: Arc<Link>) -> String {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
loop {
let (client, _) = listener.accept().await.unwrap();
let server = TcpStream::connect(&upstream).await.unwrap();
client.set_nodelay(true).unwrap();
server.set_nodelay(true).unwrap();
let (cr, cw) = client.into_split();
let (sr, sw) = server.into_split();
tokio::spawn(pipe(cr, sw, rtt / 2, up.clone()));
tokio::spawn(pipe(sr, cw, rtt / 2, down.clone()));
}
});
format!("http://{addr}")
}
async fn pipe(
mut from: tokio::net::tcp::OwnedReadHalf,
mut to: tokio::net::tcp::OwnedWriteHalf,
delay: Duration,
link: Arc<Link>,
) {
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<(Instant, Vec<u8>)>();
tokio::spawn(async move {
while let Some((due, bytes)) = rx.recv().await {
tokio::time::sleep_until(due.into()).await;
if to.write_all(&bytes).await.is_err() {
return;
}
}
let _ = to.shutdown().await;
});
let mut buf = vec![0u8; 64 * 1024];
loop {
match from.read(&mut buf).await {
Ok(0) | Err(_) => return,
Ok(n) => {
let _ = tx.send((link.send(n) + delay, buf[..n].to_vec()));
}
}
}
}
fn arg<'a>(args: &'a [String], name: &str, default: &'a str) -> &'a str {
args.iter()
.position(|a| a == name)
.and_then(|i| args.get(i + 1))
.map_or(default, String::as_str)
}
fn list(s: &str) -> Vec<u64> {
s.split(',').map(|v| v.parse().unwrap()).collect()
}
fn secs(f: impl FnOnce()) -> f64 {
let start = Instant::now();
f();
start.elapsed().as_secs_f64()
}
fn median(mut v: Vec<f64>) -> f64 {
v.sort_by(f64::total_cmp);
v[v.len() / 2]
}
/// Bytes nothing compresses, from a xorshift generator.
fn noise(len: usize, mut x: u64) -> Vec<u8> {
let mut out = Vec::with_capacity(len + 8);
while out.len() < len {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
out.extend_from_slice(&x.to_le_bytes());
}
out.truncate(len);
out
}
fn main() {
let args: Vec<String> = std::env::args().collect();
let rtt: u64 = arg(&args, "--rtt", "50").parse().unwrap();
let mbits = list(arg(&args, "--mbit", "100,20"));
let parts = list(arg(&args, "--parts", "4,16,64"));
let pack_mib: usize = arg(&args, "--pack", "64").parse().unwrap();
let n: usize = arg(&args, "-n", "3").parse().unwrap();
let release = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target/release");
assert!(
release.join("git-remote-knotless").exists(),
"run `cargo build --release` first"
);
// SAFETY: set before any thread that reads the environment starts.
unsafe { std::env::set_var("KNOTLESS_BIN_DIR", &release) };
let rt = tokio::runtime::Runtime::new().unwrap();
let (up, down) = (Link::new(), Link::new());
let upstream = net().pds.trim_start_matches("http://").to_string();
let proxy = rt.block_on(shaped_proxy(
upstream,
Duration::from_millis(rtt),
up.clone(),
down.clone(),
));
let mut s = Sandbox::new();
s.set("KNOTLESS_SERVICE", &proxy);
s.set("KNOTLESS_COMPACT", "off");
s.set("KNOTLESS_TIMEOUT_SECS", "600");
let work = s.work("work", 1);
std::fs::write(
work.join("big.bin"),
noise(pack_mib << 20, 0x9e3779b97f4a7c15),
)
.unwrap();
s.git(&work, &["add", "big.bin"]);
s.git(&work, &["commit", "-q", "-m", "big"]);
// A login and the identity, cached before anything is timed.
s.git(&work, &["push", "-q", &s.url("warm"), "main"]);
println!("{pack_mib} MiB of noise, {rtt} ms RTT, median of {n}\n");
println!(" link part parts push s clone s");
let mut run = 0;
for &mbit in &mbits {
// 0 is no cap: the cost of the work itself, on this machine.
let bits = if mbit == 0 {
u64::MAX
} else {
mbit * 1_000_000
};
for link in [&up, &down] {
link.bits.store(bits, Ordering::Relaxed);
}
for &part in &parts {
s.set("KNOTLESS_CHUNK_SIZE", &(part << 20).to_string());
let (mut push, mut clone) = (Vec::new(), Vec::new());
for _ in 0..n {
run += 1;
let url = s.url(&format!("chunks{run}"));
push.push(secs(|| {
s.git(&work, &["push", "-q", &url, "main"]);
}));
let dir = s.path(&format!("clone{run}"));
clone.push(secs(|| {
s.git(s.dir.path(), &["clone", "-q", &url, dir.to_str().unwrap()]);
}));
std::fs::remove_dir_all(&dir).unwrap();
}
println!(
"{mbit:4} Mbit/s {part:4} MiB {:6} {:6.1} {:6.1}",
(pack_mib as u64).div_ceil(part),
median(push),
median(clone)
);
}
}
}
crates/knotless/src/env.rs+2−1
@@ -39,7 +39,8 @@ pub struct Env {
pub browser: Option<String>,
/// Packs up to this many bytes ride inside the pack record itself.
pub inline_max: usize,
/// Blob part size for larger packs; halved when a PDS refuses it.
/// Blob part size for larger packs, at most the PDS's advertised limit;
/// halved when a PDS refuses it.
pub chunk_size: usize,
pub compact_threshold: usize,
pub compact_mode: CompactMode,
crates/knotless/src/identity.rs+1−1
@@ -23,7 +23,7 @@ struct Cached {
at: u64,
}
fn now_secs() -> u64 {
pub(crate) fn now_secs() -> u64 {
crate::keys::now_micros() / 1_000_000
}
crates/knotless/src/push.rs+48−3
@@ -8,11 +8,13 @@ use crate::records::{
self, BlobRef, Bytes, ClaimRecord, PACK_MIME, PackRecord, RefRecord, RepoRecord, nsid,
};
use crate::remote::{self, Remote, Snapshot, Stored};
use crate::xrpc::{Auth, XrpcError};
use crate::xrpc::{Auth, Xrpc, XrpcError};
use crate::{identity, store};
use anyhow::{Result, bail};
use futures::{StreamExt, TryStreamExt};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::BTreeSet;
use std::collections::{BTreeMap, BTreeSet};
/// One refspec of a push. `src: None` deletes `dst`.
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -323,7 +325,14 @@ pub async fn place(
/// Uploads `pack` as blobs of at most the chunk size, four at a time,
/// halving the size while the PDS calls a part too large.
pub async fn upload(ctx: &Ctx, remote: &Remote, auth: &Auth, pack: &[u8]) -> Result<Vec<BlobRef>> {
let mut chunk = ctx.env.chunk_size.max(1);
// Never above what the PDS says it takes; still halved below that when
// it refuses, since what it says is not always what is in front of it.
let limit = blob_limit(ctx, &remote.xrpc).await;
let mut chunk = ctx
.env
.chunk_size
.min(limit.map_or(usize::MAX, |l| l as usize))
.max(1);
loop {
// Owned parts, so the stream's future is Send for any caller.
let parts: Vec<Vec<u8>> = pack.chunks(chunk).map(<[u8]>::to_vec).collect();
@@ -343,6 +352,42 @@ pub async fn upload(ctx: &Ctx, remote: &Remote, auth: &Auth, pack: &[u8]) -> Res
}
}
#[derive(Serialize, Deserialize)]
struct Server {
blob_limit: Option<u64>,
at: u64,
}
/// The largest blob `xrpc`'s PDS says it takes, from `describeServer`:
/// `None` if it does not say, or cannot be asked. Cached on disk, as
/// identities are.
async fn blob_limit(ctx: &Ctx, xrpc: &Xrpc) -> Option<u64> {
let path = ctx.env.cache_dir.join("servers.json");
let mut cache: BTreeMap<String, Server> = store::read(&path).unwrap_or_default();
if let Some(hit) = cache.get(&xrpc.base)
&& identity::now_secs().saturating_sub(hit.at) < ctx.env.identity_ttl.as_secs()
{
return hit.blob_limit;
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct Out {
blob_upload_limit: Option<u64>,
}
let described = xrpc.query::<Out>("com.atproto.server.describeServer", &[], None);
let out = described.await.ok()?;
cache.insert(
xrpc.base.clone(),
Server {
blob_limit: out.blob_upload_limit,
at: identity::now_secs(),
},
);
// A cache that cannot be written only costs the next push.
let _ = store::write(&path, &cache, false);
out.blob_upload_limit
}
fn too_large(err: &anyhow::Error) -> bool {
XrpcError::of(err).is_some_and(|e| {
e.status == 413 || e.error == "BlobTooLarge" || e.error == "PayloadTooLargeError"
crates/knotless/tests/git_flows.rs+38−5
@@ -102,14 +102,13 @@ fn tags_and_empty_repositories() {
assert_eq!(s.git(&clone, &["describe"]), "v1");
}
#[test]
fn large_packs_go_over_blobs() {
let mut s = Sandbox::new();
/// Pushes 7 MB nothing compresses and clones it back, through blobs of at
/// most `KNOTLESS_CHUNK_SIZE` (6 MB, above the reference PDS's 5 MiB
/// limit): the number of parts the pack went up in.
fn push_large(s: &mut Sandbox) -> usize {
s.set("KNOTLESS_INLINE_MAX", "1024");
// Larger than the reference PDS's 5 MB blob limit: the push halves it.
s.set("KNOTLESS_CHUNK_SIZE", "6000000");
let work = s.work("work", 1);
// Incompressible, so the pack stays larger than one blob.
let mut x: u64 = 0x9e3779b97f4a7c15;
let noise: Vec<u8> = (0..7_000_000)
.map(|_| {
@@ -130,6 +129,40 @@ fn large_packs_go_over_blobs() {
&["clone", "-q", &url, clone.to_str().unwrap()],
);
assert_eq!(std::fs::read(clone.join("blob.bin")).unwrap(), noise);
let packs: serde_json::Value = reqwest::blocking::Client::new()
.get(format!("{}/xrpc/com.atproto.repo.listRecords", net().pds))
.query(&[
("repo", s.account.did.as_str()),
("collection", "dev.knotless.pack"),
])
.send()
.and_then(|r| r.json())
.unwrap();
let [pack] = packs["records"].as_array().unwrap().as_slice() else {
panic!("{packs}")
};
pack["value"]["parts"].as_array().unwrap().len()
}
#[test]
fn large_packs_go_over_blobs_the_pds_takes() {
// Its describeServer says 5 MiB, so two parts, with nothing refused:
// halving 6 MB on a refusal would have made three.
assert_eq!(push_large(&mut Sandbox::new()), 2);
}
#[test]
fn a_refused_part_is_halved() {
// A limit the PDS was taken to say, above what it takes.
let mut s = Sandbox::new();
let cache = s.path("cache");
std::fs::create_dir_all(&cache).unwrap();
let now = std::time::UNIX_EPOCH.elapsed().unwrap().as_secs();
let servers = serde_json::json!({
net().pds.trim_end_matches('/'): {"blob_limit": 100_000_000, "at": now},
});
std::fs::write(cache.join("servers.json"), servers.to_string()).unwrap();
assert_eq!(push_large(&mut s), 3);
}
#[test]
harness/package-lock.json+1−1
@@ -9,7 +9,7 @@
"@atproto/dev-env": "0.6.26"
},
"devDependencies": {
"@playwright/test": "^1.58.2"
"@playwright/test": "1.58.2"
}
},
"node_modules/@atproto-labs/did-resolver": {
harness/pds.mjs+2−1
@@ -2,7 +2,7 @@
// directory on free ports, creates test accounts, prints one JSON line
// describing them, and shuts down when stdin closes.
//
// node harness/pds.mjs [--accounts alice,bob,carol] [--rate-limits]
// node harness/pds.mjs [--accounts alice,bob,carol] [--rate-limits] [--blob-limit <bytes>]
//
// KNOTLESS_DEV_ENV names a built dev-env package to use instead of the npm
// one, such as an atproto checkout's packages/dev-env on the spaces alpha.
@@ -58,6 +58,7 @@ const net = await TestNetworkNoAppView.create({
pds: {
hostname: 'localhost',
rateLimitsEnabled: process.argv.includes('--rate-limits'),
...(arg('--blob-limit') && { blobUploadLimit: Number(arg('--blob-limit')) }),
},
})