From 1df7007f3955fc74f8068118936eeb076cbcf046 Mon Sep 17 00:00:00 2001 From: phil Date: Tue, 29 Jul 2025 12:19:56 -0400 Subject: [PATCH] scaffold dropshot + cache lookup (no actual fetch) the response almost matches but dropshot uses "error_code: string" instead of "error: string" in its error responses --- Cargo.lock | 5 +- slingshot/Cargo.toml | 4 + slingshot/src/consumer.rs | 8 +- slingshot/src/error.rs | 20 ++- slingshot/src/lib.rs | 2 + slingshot/src/main.rs | 9 +- slingshot/src/record.rs | 25 ++- slingshot/src/server.rs | 308 +++++++++++++++++++++++++++++++++++ slingshot/static/favicon.ico | 0 slingshot/static/index.html | 53 ++++++ 10 files changed, 421 insertions(+), 13 deletions(-) create mode 100644 slingshot/src/server.rs create mode 100644 slingshot/static/favicon.ico create mode 100644 slingshot/static/index.html diff --git a/Cargo.lock b/Cargo.lock index 2197e61..c1c36b9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4640,15 +4640,18 @@ dependencies = [ name = "slingshot" version = "0.1.0" dependencies = [ - "axum", "clap", "ctrlc", + "dropshot", "env_logger", "foyer", + "http", "jetstream", "log", "metrics", "metrics-exporter-prometheus 0.17.2", + "schemars", + "semver", "serde", "serde_json", "thiserror 2.0.12", diff --git a/slingshot/Cargo.toml b/slingshot/Cargo.toml index 7d76e31..d034716 100644 --- a/slingshot/Cargo.toml +++ b/slingshot/Cargo.toml @@ -6,12 +6,16 @@ edition = "2024" [dependencies] clap = { version = "4.5.41", features = ["derive"] } ctrlc = "3.4.7" +dropshot = "0.16.2" env_logger = "0.11.8" foyer = { version = "0.18.0", features = ["serde"] } +http = "1.3.1" jetstream = { path = "../jetstream", features = ["metrics"] } log = "0.4.27" metrics = "0.24.2" metrics-exporter-prometheus = { version = "0.17.1", features = ["http-listener"] } +schemars = { version = "0.8.22", features = ["raw_value"] } +semver = "1.0.26" serde = { version = "1.0.219", features = ["derive"] } serde_json = { version = "1.0.141", features = ["raw_value"] } thiserror = "2.0.12" diff --git a/slingshot/src/consumer.rs b/slingshot/src/consumer.rs index 84f2cc6..51fb3ec 100644 --- a/slingshot/src/consumer.rs +++ b/slingshot/src/consumer.rs @@ -64,11 +64,15 @@ pub async fn consume( cache.insert(at_uri, CachedRecord::Deleted); } else { let Some(record) = commit.record.take() else { - log::warn!("consumer: commit update/delete missing record, ignoring"); + log::warn!("consumer: commit insert or update missing record, ignoring"); + continue; + }; + let Some(cid) = commit.cid.take() else { + log::warn!("consumer: commit insert or update missing CID, ignoring"); continue; }; - cache.insert(at_uri, CachedRecord::Found(record.into())); + cache.insert(at_uri, CachedRecord::Found((cid, record).into())); } } diff --git a/slingshot/src/error.rs b/slingshot/src/error.rs index b427c42..ba90399 100644 --- a/slingshot/src/error.rs +++ b/slingshot/src/error.rs @@ -12,10 +12,26 @@ pub enum ConsumerError { DelayQueueOutputDropped, } +#[derive(Debug, Error)] +pub enum ServerError { + #[error("failed to configure server logger: {0}")] + ConfigLogError(std::io::Error), + #[error("failed to render json for openapi: {0}")] + OpenApiJsonFail(serde_json::Error), + #[error(transparent)] + FailedToBuildServer(#[from] dropshot::BuildError), + #[error("server exited: {0}")] + ServerExited(String), + #[error("server closed badly: {0}")] + BadClose(String), + #[error("blahhhahhahha")] + OhNo(String), +} + #[derive(Debug, Error)] pub enum MainTaskError { #[error(transparent)] ConsumerTaskError(#[from] ConsumerError), - // #[error(transparent)] - // ServerTaskError(#[from] ServerError), + #[error(transparent)] + ServerTaskError(#[from] ServerError), } diff --git a/slingshot/src/lib.rs b/slingshot/src/lib.rs index 75dd162..240b4a7 100644 --- a/slingshot/src/lib.rs +++ b/slingshot/src/lib.rs @@ -2,7 +2,9 @@ mod consumer; pub mod error; mod firehose_cache; mod record; +mod server; pub use consumer::consume; pub use firehose_cache::firehose_cache; pub use record::CachedRecord; +pub use server::serve; diff --git a/slingshot/src/main.rs b/slingshot/src/main.rs index 4c46187..d7ce143 100644 --- a/slingshot/src/main.rs +++ b/slingshot/src/main.rs @@ -1,7 +1,7 @@ // use foyer::HybridCache; // use foyer::{Engine, DirectFsDeviceOptions, HybridCacheBuilder}; use metrics_exporter_prometheus::PrometheusBuilder; -use slingshot::{consume, error::MainTaskError, firehose_cache}; +use slingshot::{consume, error::MainTaskError, firehose_cache, serve}; use clap::Parser; use tokio_util::sync::CancellationToken; @@ -45,6 +45,13 @@ async fn main() -> Result<(), String> { let mut tasks: tokio::task::JoinSet> = tokio::task::JoinSet::new(); + let server_shutdown = shutdown.clone(); + let server_cache_handle = cache.clone(); + tasks.spawn(async move { + serve(server_cache_handle, server_shutdown).await?; + Ok(()) + }); + let consumer_shutdown = shutdown.clone(); tasks.spawn(async move { consume( diff --git a/slingshot/src/record.rs b/slingshot/src/record.rs index e6d7d7f..9df588b 100644 --- a/slingshot/src/record.rs +++ b/slingshot/src/record.rs @@ -1,19 +1,30 @@ use serde_json::value::RawValue; use serde::{Serialize, Deserialize}; +use jetstream::exports::Cid; #[derive(Debug, Serialize, Deserialize)] -pub struct RawRecord(String); +pub struct RawRecord { + cid: Cid, + record: String, +} -impl From> for RawRecord { - fn from(rv: Box) -> Self { - Self(rv.get().to_string()) +// TODO: should be able to do typed CID +impl From<(Cid, Box)> for RawRecord { + fn from((cid, rv): (Cid, Box)) -> Self { + Self { + cid, + record: rv.get().to_string(), + } } } /// only for use with stored (validated) values, not general strings -impl From for Box { - fn from(RawRecord(s): RawRecord) -> Self { - RawValue::from_string(s).expect("stored string from RawValue to be valid") +impl From<&RawRecord> for (Cid, Box) { + fn from(RawRecord { cid, record }: &RawRecord) -> Self { + ( + cid.clone(), + RawValue::from_string(record.to_string()).expect("stored string from RawValue to be valid"), + ) } } diff --git a/slingshot/src/server.rs b/slingshot/src/server.rs new file mode 100644 index 0000000..b071fee --- /dev/null +++ b/slingshot/src/server.rs @@ -0,0 +1,308 @@ +use serde_json::value::RawValue; +use crate::CachedRecord; +use foyer::HybridCache; +use crate::error::ServerError; +use dropshot::{ + ApiDescription, Body, ConfigDropshot, ConfigLogging, + ConfigLoggingLevel, HttpError, HttpResponse, Query, RequestContext, + ServerBuilder, ServerContext, endpoint, + ClientErrorStatusCode, +}; +use http::{ + Response, StatusCode, + header::{ORIGIN, USER_AGENT}, +}; +use metrics::{counter, histogram}; +use std::sync::Arc; + +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use tokio::time::Instant; +use tokio_util::sync::CancellationToken; + +const INDEX_HTML: &str = include_str!("../static/index.html"); +const FAVICON: &[u8] = include_bytes!("../static/favicon.ico"); + +pub async fn serve( + cache: HybridCache, + shutdown: CancellationToken, +) -> Result<(), ServerError> { + let config_logging = ConfigLogging::StderrTerminal { + level: ConfigLoggingLevel::Info, + }; + + let log = config_logging + .to_logger("example-basic") + .map_err(ServerError::ConfigLogError)?; + + let mut api = ApiDescription::new(); + api.register(index).unwrap(); + api.register(favicon).unwrap(); + api.register(openapi).unwrap(); + api.register(get_record).unwrap(); + + // TODO: put spec in a once cell / lazy lock thing? + let spec = Arc::new( + api.openapi( + "Slingshot", + env!("CARGO_PKG_VERSION") + .parse() + .inspect_err(|e| { + eprintln!("failed to parse cargo package version for openapi: {e:?}") + }) + .unwrap_or(semver::Version::new(0, 0, 1)), + ) + .description("A fast edge cache for getRecord") + .contact_name("part of @microcosm.blue") + .contact_url("https://microcosm.blue") + .json() + .map_err(ServerError::OpenApiJsonFail)?, + ); + + let sub_shutdown = shutdown.clone(); + let ctx = Context { + cache, + spec, + shutdown: sub_shutdown, + }; + + let server = ServerBuilder::new(api, ctx, log) + .config(ConfigDropshot { + bind_address: "0.0.0.0:9996".parse().unwrap(), + ..Default::default() + }) + .start()?; + + tokio::select! { + s = server.wait_for_shutdown() => { + s.map_err(ServerError::ServerExited)?; + log::info!("server shut down normally."); + }, + _ = shutdown.cancelled() => { + log::info!("shutting down: closing server"); + server.close().await.map_err(ServerError::BadClose)?; + }, + } + Ok(()) +} + +#[derive(Debug, Clone)] +struct Context { + pub cache: HybridCache, + pub spec: Arc, + pub shutdown: CancellationToken, +} + +async fn instrument_handler(ctx: &RequestContext, handler: H) -> Result +where + R: HttpResponse, + H: Future>, + T: ServerContext, +{ + let start = Instant::now(); + let result = handler.await; + let latency = start.elapsed(); + let status_code = match &result { + Ok(response) => response.status_code(), + Err(e) => e.status_code.as_status(), + } + .as_str() // just the number (.to_string()'s Display does eg `200 OK`) + .to_string(); + let endpoint = ctx.endpoint.operation_id.clone(); + let headers = ctx.request.headers(); + let origin = headers + .get(ORIGIN) + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_string(); + let ua = headers + .get(USER_AGENT) + .and_then(|v| v.to_str().ok()) + .map(|ua| { + if ua.starts_with("Mozilla/5.0 ") { + "browser" + } else { + ua + } + }) + .unwrap_or("") + .to_string(); + counter!("server_requests_total", + "endpoint" => endpoint.clone(), + "origin" => origin, + "ua" => ua, + "status_code" => status_code, + ) + .increment(1); + histogram!("server_handler_latency", "endpoint" => endpoint).record(latency.as_micros() as f64); + result +} + +use dropshot::{HttpResponseHeaders, HttpResponseOk}; + +pub type OkCorsResponse = Result>, HttpError>; + +/// Helper for constructing Ok responses: return OkCors(T).into() +/// (not happy with this yet) +pub struct OkCors(pub T); + +impl From> for OkCorsResponse +where + T: Serialize + JsonSchema + Send + Sync, +{ + fn from(ok: OkCors) -> OkCorsResponse { + let mut res = HttpResponseHeaders::new_unnamed(HttpResponseOk(ok.0)); + res.headers_mut() + .insert("access-control-allow-origin", "*".parse().unwrap()); + Ok(res) + } +} + +pub fn cors_err(e: HttpError) -> HttpError { + e.with_header("access-control-allow-origin", "*").unwrap() +} + + +// TODO: cors for HttpError + +/// Serve index page as html +#[endpoint { + method = GET, + path = "/", + /* + * not useful to have this in openapi + */ + unpublished = true, +}] +async fn index(ctx: RequestContext) -> Result, HttpError> { + instrument_handler(&ctx, async { + Ok(Response::builder() + .status(StatusCode::OK) + .header(http::header::CONTENT_TYPE, "text/html") + .body(INDEX_HTML.into())?) + }) + .await +} + +/// Serve index page as html +#[endpoint { + method = GET, + path = "/favicon.ico", + /* + * not useful to have this in openapi + */ + unpublished = true, +}] +async fn favicon(ctx: RequestContext) -> Result, HttpError> { + instrument_handler(&ctx, async { + Ok(Response::builder() + .status(StatusCode::OK) + .header(http::header::CONTENT_TYPE, "image/x-icon") + .body(FAVICON.to_vec().into())?) + }) + .await +} + +/// Meta: get the openapi spec for this api +#[endpoint { + method = GET, + path = "/openapi", + /* + * not useful to have this in openapi + */ + unpublished = true, +}] +async fn openapi(ctx: RequestContext) -> OkCorsResponse { + instrument_handler(&ctx, async { + let spec = (*ctx.context().spec).clone(); + OkCors(spec).into() + }) + .await +} + + +#[derive(Debug, Deserialize, JsonSchema)] +struct GetRecordQuery { + /// The DID of the repo + /// + /// NOTE: handles should be accepted here but this is still TODO in slingshot + pub repo: String, + /// The NSID of the record collection + pub collection: String, + /// The Record key + pub rkey: String, + /// Optional: the CID of the version of the record. + /// + /// If not specified, then return the most recent version. + /// + /// If specified and a newer version of the record exists, returns 404 not + /// found. That is: slingshot only retains the most recent version of a + /// record. + #[serde(default)] + pub cid: Option, +} + +#[derive(Debug, Serialize, JsonSchema)] +struct GetRecordResponse { + pub uri: String, + pub cid: String, + pub value: Box, +} + +/// com.atproto.repo.getRecord +/// +/// Get a single record from a repository. Does not require auth. +/// +/// See https://docs.bsky.app/docs/api/com-atproto-repo-get-record for the +/// canonical XRPC documentation that this endpoint aims to be compatible with. +#[endpoint { + method = GET, + path = "/xrpc/com.atproto.repo.getRecord", +}] +async fn get_record( + ctx: RequestContext, + query: Query, +) -> OkCorsResponse { + + let Context { cache, .. } = ctx.context(); + let GetRecordQuery { repo, collection, rkey, cid } = query.into_inner(); + + // TODO: yeah yeah + let at_uri = format!( + "at://{}/{}/{}", + &*repo, &*collection, &*rkey + ); + + instrument_handler(&ctx, async { + let entry = cache + .fetch(at_uri.clone(), || async move { + Err(foyer::Error::Other(Box::new(ServerError::OhNo("booo".to_string())))) + }) + .await + .unwrap(); + + match *entry { + CachedRecord::Found(ref raw) => { + let (found_cid, raw_value) = raw.into(); + let found_cid = found_cid.as_ref().to_string(); + if cid.map(|c| c != found_cid).unwrap_or(false) { + Err(HttpError::for_not_found(None, "CID mismatch".to_string())) + .map_err(cors_err)?; + } + OkCors(GetRecordResponse { + uri: at_uri, + cid: found_cid, + value: raw_value, + }).into() + }, + CachedRecord::Deleted => { + Err(HttpError::for_client_error_with_status( + Some("Gone".to_string()), + ClientErrorStatusCode::GONE, + )).map_err(cors_err) + } + } + }) + .await + +} diff --git a/slingshot/static/favicon.ico b/slingshot/static/favicon.ico new file mode 100644 index 0000000..e69de29 diff --git a/slingshot/static/index.html b/slingshot/static/index.html new file mode 100644 index 0000000..d598eba --- /dev/null +++ b/slingshot/static/index.html @@ -0,0 +1,53 @@ + + + + + Slingshot documentation + + + + + +
+

+ todo: what link goes here?: blah +

+ +
+ + + + + + + + -- 2.51.2