use self::locations::{INFLATE_LOCATION, SEND_TIMEOUT_SECS};
pub mod conf;
mod locations;
mod maps;
#[cfg(test)]
mod tests;
pub use locations::WORKER_SHUTDOWN_SECS;
pub(crate) use locations::{
PackRedirect, firehose_location, git_loopback, inflate_location, limit_zones, oauth_callback,
pack_location, push_location, repo_locations, write_locations,
};
pub(crate) use maps::{
base_fits_row, firehose_maps, git_maps, limit_maps, moved_fits_row, trusted_rate,
};
use std::path::Path;
use anyhow::Context as _;
use crate::config::{ServeLimits, ValidatedConfig};
use crate::manifest::GenerationManifest;
use conf::{Block, NONE};
/// `no-cache` forces revalidation, and shared files keep the same etag across generations
pub const REVALIDATE: &str = "no-cache";
/// call it after any `types` block, because nginx reads them in order.
pub fn cached(location: Block, content_type: &str, cache_control: &str) -> Block {
location
.directive("default_type", [content_type])
.directive("add_header", ["Cache-Control", cache_control, "always"])
.directive("add_header", ["Access-Control-Allow-Origin", "*", "always"])
// nginx omits headers whose value is empty, only 429 sets `$ks_retry_after`
.directive("add_header", ["Retry-After", "$ks_retry_after", "always"])
}
pub(crate) fn error_body(error: &str, message: &str) -> String {
serde_json::json!({ "error": error, "message": message }).to_string()
}
fn error_locations(name: &str, status: u16, generation_path: &str) -> [Block; 2] {
let status_eq = format!("={status}");
let dispatch = Block::exact(format!("/_ks_error/{name}")).directive("internal", NONE);
let body = Block::exact(format!("/_ks_error_body/{name}.json")).directive("internal", NONE);
[
cached(dispatch, "application/json", "no-store")
.directive(
"error_page",
[
status.to_string(),
status_eq,
format!("/_ks_error_body/{name}.json"),
],
)
.directive("return", [status]),
cached(body, "application/json", "no-store").directive(
"alias",
[format!("{generation_path}/www/_errors/{name}.json")],
),
]
}
/// a page's og card, else its repo's, else the knot's
fn og_card_locations(generation_path: &str, limits: &ServeLimits) -> [Block; 2] {
let www = format!("{generation_path}/www");
let dir = crate::og_card::WWW_DIR;
let location = |head: Block, fallbacks: &[&str]| {
let location = cached(
head.block(Block::new("types", NONE)),
"image/png",
REVALIDATE,
)
.directive("root", [&www])
.directive(
"try_files",
std::iter::once("$uri").chain(fallbacks.iter().copied()),
);
read_limited(location, limits)
};
[
location(
Block::new("location", ["~", &format!("^/{dir}/(?<ks_og_repo>[^/]+)/")]),
&[&format!("/{dir}/$ks_og_repo.png"), "/og.png"],
),
location(Block::new("location", [format!("/{dir}/")]), &["/og.png"]),
]
}
fn file_location(
path: &str,
content_type: &str,
cache_control: &str,
file: String,
limits: &ServeLimits,
) -> Block {
let location =
cached(Block::exact(path), content_type, cache_control).directive("alias", [file]);
read_limited(location, limits)
}
/// per location, because a server-wide zone would count packs a client holds for minutes.
pub(crate) fn read_limited(mut location: Block, limits: &ServeLimits) -> Block {
if limits.reads.is_some() {
let mut limit = vec!["zone=ks_reads".to_string()];
if limits.reads_burst > 0 {
limit.push(format!("burst={}", limits.reads_burst));
}
location = location.directive("limit_req", limit);
}
if let Some(connections) = limits.read_connections {
location = location.directive("limit_conn", ["ks_reads_conn", &connections.to_string()]);
}
location
}
fn not_found(location: Block, message: &str, limits: &ServeLimits) -> Block {
let location = cached(location, "application/json", "no-store").directive(
"return",
["404".to_string(), error_body("InvalidRequest", message)],
);
read_limited(location, limits)
}
const ERRORS: [(&str, u16, &str, &str); 22] = [
("invalid_request", 400, "InvalidRequest", "invalid request"),
(
"atp_repo_not_found",
400,
"RepoNotFound",
"couldn't find repo",
),
(
"atp_record_not_found",
400,
"RecordNotFound",
"couldn't locate record",
),
(
"rate_limited",
429,
"RateLimitExceeded",
"too many write requests, retry shortly",
),
(
"fetches_saturated",
429,
"RateLimitExceeded",
"this knot is answering as many fetches as it can, retry shortly",
),
(
"request_too_large",
413,
"RequestTooLarge",
"a write request is at most 65536 bytes",
),
(
"push_too_large",
413,
"RequestTooLarge",
"a push is at most 8 GiB",
),
(
"attachment_too_large",
413,
"RequestTooLarge",
"an attachment is over this knot's upload limit",
),
(
"blob_not_found",
404,
"BlobNotFound",
"No such blob for this repo.",
),
(
"write_unavailable",
503,
"UpstreamUnavailable",
"the write handler is unavailable, retry shortly",
),
("not_found", 404, "NotFound", "no such record on this knot"),
(
"unsupported_encoding",
415,
"UnsupportedEncoding",
"a request body is gzip or plain",
),
(
"gzip_needs_cgi",
415,
"UnsupportedEncoding",
"gzipped fetches need this knot's fcgiwrap; send the request uncompressed",
),
(
"atp_not_implemented",
501,
"MethodNotImplemented",
"Method Not Implemented",
),
(
"git_repo_not_found",
404,
"RepoNotFound",
"repository not found on this knot",
),
(
"git_ref_not_found",
404,
"RefNotFound",
"git reference not found",
),
(
"git_path_not_found",
404,
"PathNotFound",
"path not found in repository tree",
),
(
"git_file_not_found",
404,
"FileNotFound",
"file not found at specified path",
),
(
"git_entry_not_found",
404,
"EntryNotFound",
"entry not found at specified path",
),
(
"git_branch_not_found",
404,
"BranchNotFound",
"branch not found",
),
(
"git_no_merge_base",
404,
"NoMergeBase",
"two refs share no common ancestor",
),
(
"git_not_projected",
404,
"NotFound",
"this static knot precomputes a bounded set of reads, and this isn't one of them",
),
];
/// a ceiling far above the biggest map makes `nginx -t` slow.
fn map_hash_max_size(maps_dir: &Path) -> anyhow::Result<usize> {
let mut biggest: usize = 0;
let mut carried = std::collections::HashMap::new();
for entry in std::fs::read_dir(maps_dir)? {
let path = entry?.path();
let text =
std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
let mut keys = 0;
for line in text.lines() {
match line.trim() {
line if line.starts_with("map ") => keys = 0,
"}" => biggest = biggest.max(keys),
line if let Some(fragment) = line.strip_prefix("include ") => {
keys += crate::git_xrpc::fragment_entries(
Path::new(fragment.trim_end_matches(';')),
&mut carried,
)?;
}
_ => keys += 1,
}
}
}
Ok(biggest.next_power_of_two().max(2048))
}
/// the build writes to `generation_dir`, but the config embeds `served_dir`.
pub fn emit_at(
config: &ValidatedConfig,
manifest: &GenerationManifest,
generation_dir: &Path,
served_dir: &Path,
) -> anyhow::Result<()> {
let served_dir = if served_dir.exists() {
std::fs::canonicalize(served_dir)?
} else {
let parent = served_dir.parent().context("generation parent")?;
std::fs::canonicalize(parent)?.join(served_dir.file_name().context("generation name")?)
};
let nginx_dir = generation_dir.join("nginx");
let maps_dir = nginx_dir.join("maps");
let errors_dir = generation_dir.join("www/_errors");
std::fs::create_dir_all(&maps_dir)
.with_context(|| format!("create maps dir: {}", maps_dir.display()))?;
if !maps_dir.join(crate::git_xrpc::MAPS_FILE).exists() {
crate::git_xrpc::emit(generation_dir, &[])?;
}
std::fs::create_dir_all(&errors_dir)
.with_context(|| format!("create errors dir: {}", errors_dir.display()))?;
let runtime_dir = config
.output_root
.join("runtime")
.join(served_dir.file_name().context("generation runtime name")?);
for name in [
"client_body_temp",
"proxy_temp",
"fastcgi_temp",
"uwsgi_temp",
"scgi_temp",
] {
std::fs::create_dir_all(runtime_dir.join(name))?;
}
let runtime_path = std::fs::canonicalize(&runtime_dir)?.display().to_string();
for (name, _, error, message) in ERRORS {
let path = errors_dir.join(format!("{name}.json"));
let body =
serde_json::to_vec_pretty(&serde_json::json!({ "error": error, "message": message }))?;
crate::cas::atomic_write(&path, &body)
.with_context(|| format!("write error body {}", path.display()))?;
}
let gen_path_str = served_dir.display().to_string();
let gen_path = gen_path_str.as_str();
let limits = &config.serve.limits;
let mut server = Block::new("server", NONE)
.directive("listen", [&config.activate.listen])
.directive("server_name", [&config.hostname])
.comment("JSON error dispatch locations");
for proxy in &config.activate.real_ip_from {
server = server.directive("set_real_ip_from", [proxy]);
}
if !config.activate.real_ip_from.is_empty() {
server = server
.directive("real_ip_header", ["X-Forwarded-For"])
.directive("real_ip_recursive", ["on"]);
}
for (name, status, _, _) in ERRORS {
server = server.blocks(error_locations(name, status, gen_path));
}
server = server
.comment("identity endpoints")
.block(read_limited(
cached(
Block::exact("/").block(Block::new("types", NONE)),
"text/html; charset=utf-8",
REVALIDATE,
)
.directive("root", [format!("{gen_path}/www")])
.directive(
"try_files",
[
format!("/{}", crate::index_page::INDEX_FILE),
"=404".to_string(),
],
),
limits,
))
.block(file_location(
"/favicon.svg",
"image/svg+xml",
REVALIDATE,
format!("{gen_path}/www/{}", crate::index_page::FAVICON_SVG_FILE),
limits,
))
.block(file_location(
"/favicon.ico",
"image/png",
REVALIDATE,
format!("{gen_path}/www/{}", crate::index_page::FAVICON_PNG_FILE),
limits,
))
.block(file_location(
"/apple-touch-icon.png",
"image/png",
REVALIDATE,
format!(
"{gen_path}/www/{}",
crate::index_page::APPLE_TOUCH_ICON_FILE
),
limits,
))
.block(file_location(
"/og.png",
"image/png",
REVALIDATE,
format!("{gen_path}/www/{}", crate::index_page::OG_IMAGE_FILE),
limits,
))
.blocks(og_card_locations(gen_path, limits))
.block(file_location(
"/.well-known/did.json",
"application/did+json",
"public, max-age=300",
format!("{gen_path}/www/.well-known/did.json"),
limits,
))
.block(file_location(
"/xrpc/org.tangled.knot.describeKnot",
"application/json",
REVALIDATE,
format!("{gen_path}/www/_describe.json"),
limits,
))
.block(file_location(
"/xrpc/_health",
"application/json",
REVALIDATE,
format!("{gen_path}/www/_health.json"),
limits,
));
if generation_dir
.join("www/oauth/client-metadata.json")
.exists()
{
server = server.block(file_location(
"/oauth/client-metadata.json",
"application/json",
"public, max-age=300",
format!("{gen_path}/www/oauth/client-metadata.json"),
limits,
));
}
server = server.block(oauth_callback(config)?);
// the reply is picked from the POST body, which nginx only reads for a proxy.
// nginx's static file handler returns 405 on POST, the loopback proxies to itself as a GET
server = server.comment("smart git");
let inflate = match config.writes.fastcgi_socket {
Some(_) => INFLATE_LOCATION,
None => "/_ks_error/gzip_needs_cgi",
};
for repo in &manifest.repos {
server = server.blocks(repo_locations(
&repo.repo_did,
gen_path,
config.writes.enabled,
inflate,
limits,
));
}
if config.writes.fastcgi_socket.is_some() {
server = server.block(inflate_location(config)?);
}
server = server
.block(git_loopback(gen_path, limits))
.block(pack_location(gen_path, limits))
.comment("com.atproto surface for repo DIDs (static files, keyed by query)")
.blocks(crate::atproto::locations(gen_path, limits))
.comment("org.tangled.git reads (precomputed, keyed by query)")
.blocks(crate::git_xrpc::locations(gen_path, limits))
.blocks(crate::blobs::locations(gen_path, limits));
if let Some(socket) = &config.firehose_socket {
server = server
.comment("subscribeRepos (the firehose server)")
.block(firehose_location(socket));
}
if config.writes.enabled {
server = server
.comment("writes (fcgiwrap)")
.blocks(write_locations(config)?)
.block(push_location(config)?);
}
server = server
.block(not_found(
Block::new("location", ["/xrpc/"]),
"unknown endpoint",
limits,
))
.block(not_found(
Block::new("location", ["/"]),
"not found",
limits,
));
let runtime = runtime_path.as_str();
let mut http = Block::new("http", NONE)
.directive(
"access_log",
[
format!("{runtime}/access.log"),
"combined".into(),
"buffer=64k".into(),
"flush=5s".into(),
],
)
// no `open_file_cache`, which serves a deleted file for up to a minute.
.directive("sendfile", ["on"])
.directive("tcp_nopush", ["on"])
.directive("send_timeout", [format!("{SEND_TIMEOUT_SECS}s")]);
for name in [
"client_body_temp",
"proxy_temp",
"fastcgi_temp",
"uwsgi_temp",
"scgi_temp",
] {
http = http.directive(&format!("{name}_path"), [format!("{runtime}/{name}")]);
}
if config.firehose_socket.is_some() {
http = http.directive("limit_conn_zone", ["$ks_client", "zone=ks_firehose:1m"]);
}
let conf = Block::file()
.directive("error_log", [format!("{runtime}/error.log")])
.directive("worker_processes", ["auto"])
// pcre_jit is on because interpreted matching of a 2 MiB body costs a second of cpu.
.directive("pcre_jit", ["on"])
.directive(
"worker_shutdown_timeout",
[format!("{WORKER_SHUTDOWN_SECS}s")],
)
.block(Block::new("events", NONE).directive("worker_connections", ["1024"]))
.block(
limit_zones(http, limits)
// the bigger tables have to be set before the maps include. a bucket holds every key
// that hashes to it, and git keys run to 400 bytes, so a few long ones need 4k.
// nginx allocates only what the keys take.
.directive("map_hash_bucket_size", ["4096"])
.directive(
"map_hash_max_size",
[map_hash_max_size(&maps_dir)?.to_string()],
)
.directive("variables_hash_max_size", ["4096"])
.directive("variables_hash_bucket_size", ["256"])
.directive("include", [format!("{gen_path}/nginx/maps/*.conf")])
.directive(
"limit_req_zone",
["$binary_remote_addr", "zone=ks_writes:10m", "rate=2r/s"],
)
.directive(
"limit_conn_zone",
["$binary_remote_addr", "zone=ks_writes_conn:10m"],
)
.directive("limit_req_status", ["429"])
.directive("limit_conn_status", ["429"])
.blocks(limit_maps()?)
.blocks(
limits
.pack_rate
.iter()
.map(|rate| trusted_rate(config, rate))
.collect::<anyhow::Result<Vec<_>>>()?
.into_iter()
.flatten(),
)
.blocks(git_maps(manifest)?.blocks())
.blocks(
config
.firehose_socket
.iter()
.map(|_| firehose_maps())
.collect::<anyhow::Result<Vec<_>>>()?
.into_iter()
.flatten(),
)
.block(server),
);
let conf = conf.to_string();
let conf_path = nginx_dir.join("knot-static.conf");
crate::cas::atomic_write(&conf_path, conf.as_bytes())
.with_context(|| format!("write {}", conf_path.display()))?;
emit_check(&conf, &nginx_dir, gen_path)
}
/// what `nginx -t` checks when the git read maps are the ones it already accepted: the same conf
/// with those maps' heads and none of their fragments, which are most of the test's time
fn emit_check(conf: &str, nginx_dir: &Path, gen_path: &str) -> anyhow::Result<()> {
let git_maps =
std::fs::read_to_string(nginx_dir.join("maps").join(crate::git_xrpc::MAPS_FILE))?;
let heads: String = git_maps
.lines()
.filter(|line| !line.trim_start().starts_with("include "))
.map(|line| format!("{line}\n"))
.collect();
let check_dir = nginx_dir.join(CHECK_DIR);
std::fs::create_dir_all(&check_dir)?;
crate::cas::atomic_write(
&check_dir.join(crate::git_xrpc::MAPS_FILE),
heads.as_bytes(),
)?;
let mut maps: Vec<String> = std::fs::read_dir(nginx_dir.join("maps"))?
.map(|entry| Ok(entry?.file_name().to_string_lossy().into_owned()))
.collect::<std::io::Result<_>>()?;
maps.retain(|name| name != crate::git_xrpc::MAPS_FILE);
maps.sort();
let glob = format!("include {gen_path}/nginx/maps/*.conf;");
let mut check = String::with_capacity(conf.len());
let mut found = false;
for line in conf.lines() {
if line.trim() == glob {
found = true;
let indent = &line[..line.len() - line.trim_start().len()];
for name in &maps {
check += &format!("{indent}include {gen_path}/nginx/maps/{name};\n");
}
check += &format!(
"{indent}include {gen_path}/nginx/{CHECK_DIR}/{};\n",
crate::git_xrpc::MAPS_FILE
);
} else {
check += line;
check += "\n";
}
}
anyhow::ensure!(
found,
"the conf has no maps include to check without the git read maps"
);
crate::cas::atomic_write(&nginx_dir.join(CHECK_FILE), check.as_bytes())
.context("write the conf nginx -t checks")?;
Ok(())
}
pub(crate) const CHECK_FILE: &str = "knot-static.check.conf";
const CHECK_DIR: &str = "check";