scaffold dropshot + cache lookup (no actual fetch) · microcosm.blue/microcosm-rs@1df7007

+4 -1

Cargo.lock

··· 4640 4640 name = "slingshot" 4641 4641 version = "0.1.0" 4642 4642 dependencies = [ 4643 - "axum", 4644 4643 "clap", 4645 4644 "ctrlc", 4645 + "dropshot", 4646 4646 "env_logger", 4647 4647 "foyer", 4648 + "http", 4648 4649 "jetstream", 4649 4650 "log", 4650 4651 "metrics", 4651 4652 "metrics-exporter-prometheus 0.17.2", 4653 + "schemars", 4654 + "semver", 4652 4655 "serde", 4653 4656 "serde_json", 4654 4657 "thiserror 2.0.12",

+4

slingshot/Cargo.toml

··· 6 6 [dependencies] 7 7 clap = { version = "4.5.41", features = ["derive"] } 8 8 ctrlc = "3.4.7" 9 + dropshot = "0.16.2" 9 10 env_logger = "0.11.8" 10 11 foyer = { version = "0.18.0", features = ["serde"] } 12 + http = "1.3.1" 11 13 jetstream = { path = "../jetstream", features = ["metrics"] } 12 14 log = "0.4.27" 13 15 metrics = "0.24.2" 14 16 metrics-exporter-prometheus = { version = "0.17.1", features = ["http-listener"] } 17 + schemars = { version = "0.8.22", features = ["raw_value"] } 18 + semver = "1.0.26" 15 19 serde = { version = "1.0.219", features = ["derive"] } 16 20 serde_json = { version = "1.0.141", features = ["raw_value"] } 17 21 thiserror = "2.0.12"

+6 -2

slingshot/src/consumer.rs

··· 64 64 cache.insert(at_uri, CachedRecord::Deleted); 65 65 } else { 66 66 let Some(record) = commit.record.take() else { 67 - log::warn!("consumer: commit update/delete missing record, ignoring"); 67 + log::warn!("consumer: commit insert or update missing record, ignoring"); 68 + continue; 69 + }; 70 + let Some(cid) = commit.cid.take() else { 71 + log::warn!("consumer: commit insert or update missing CID, ignoring"); 68 72 continue; 69 73 }; 70 74 71 - cache.insert(at_uri, CachedRecord::Found(record.into())); 75 + cache.insert(at_uri, CachedRecord::Found((cid, record).into())); 72 76 } 73 77 } 74 78

+18 -2

slingshot/src/error.rs

··· 13 13 } 14 14 15 15 #[derive(Debug, Error)] 16 + pub enum ServerError { 17 + #[error("failed to configure server logger: {0}")] 18 + ConfigLogError(std::io::Error), 19 + #[error("failed to render json for openapi: {0}")] 20 + OpenApiJsonFail(serde_json::Error), 21 + #[error(transparent)] 22 + FailedToBuildServer(#[from] dropshot::BuildError), 23 + #[error("server exited: {0}")] 24 + ServerExited(String), 25 + #[error("server closed badly: {0}")] 26 + BadClose(String), 27 + #[error("blahhhahhahha")] 28 + OhNo(String), 29 + } 30 + 31 + #[derive(Debug, Error)] 16 32 pub enum MainTaskError { 17 33 #[error(transparent)] 18 34 ConsumerTaskError(#[from] ConsumerError), 19 - // #[error(transparent)] 20 - // ServerTaskError(#[from] ServerError), 35 + #[error(transparent)] 36 + ServerTaskError(#[from] ServerError), 21 37 }

+2

slingshot/src/lib.rs

··· 2 2 pub mod error; 3 3 mod firehose_cache; 4 4 mod record; 5 + mod server; 5 6 6 7 pub use consumer::consume; 7 8 pub use firehose_cache::firehose_cache; 8 9 pub use record::CachedRecord; 10 + pub use server::serve;

+8 -1

slingshot/src/main.rs

··· 1 1 // use foyer::HybridCache; 2 2 // use foyer::{Engine, DirectFsDeviceOptions, HybridCacheBuilder}; 3 3 use metrics_exporter_prometheus::PrometheusBuilder; 4 - use slingshot::{consume, error::MainTaskError, firehose_cache}; 4 + use slingshot::{consume, error::MainTaskError, firehose_cache, serve}; 5 5 6 6 use clap::Parser; 7 7 use tokio_util::sync::CancellationToken; ··· 44 44 log::info!("firehose cache ready."); 45 45 46 46 let mut tasks: tokio::task::JoinSet<Result<(), MainTaskError>> = tokio::task::JoinSet::new(); 47 + 48 + let server_shutdown = shutdown.clone(); 49 + let server_cache_handle = cache.clone(); 50 + tasks.spawn(async move { 51 + serve(server_cache_handle, server_shutdown).await?; 52 + Ok(()) 53 + }); 47 54 48 55 let consumer_shutdown = shutdown.clone(); 49 56 tasks.spawn(async move {

+18 -7

slingshot/src/record.rs

··· 1 1 use serde_json::value::RawValue; 2 2 use serde::{Serialize, Deserialize}; 3 + use jetstream::exports::Cid; 3 4 4 5 #[derive(Debug, Serialize, Deserialize)] 5 - pub struct RawRecord(String); 6 + pub struct RawRecord { 7 + cid: Cid, 8 + record: String, 9 + } 6 10 7 - impl From<Box<RawValue>> for RawRecord { 8 - fn from(rv: Box<RawValue>) -> Self { 9 - Self(rv.get().to_string()) 11 + // TODO: should be able to do typed CID 12 + impl From<(Cid, Box<RawValue>)> for RawRecord { 13 + fn from((cid, rv): (Cid, Box<RawValue>)) -> Self { 14 + Self { 15 + cid, 16 + record: rv.get().to_string(), 17 + } 10 18 } 11 19 } 12 20 13 21 /// only for use with stored (validated) values, not general strings 14 - impl From<RawRecord> for Box<RawValue> { 15 - fn from(RawRecord(s): RawRecord) -> Self { 16 - RawValue::from_string(s).expect("stored string from RawValue to be valid") 22 + impl From<&RawRecord> for (Cid, Box<RawValue>) { 23 + fn from(RawRecord { cid, record }: &RawRecord) -> Self { 24 + ( 25 + cid.clone(), 26 + RawValue::from_string(record.to_string()).expect("stored string from RawValue to be valid"), 27 + ) 17 28 } 18 29 } 19 30

+308

slingshot/src/server.rs

··· 1 + use serde_json::value::RawValue; 2 + use crate::CachedRecord; 3 + use foyer::HybridCache; 4 + use crate::error::ServerError; 5 + use dropshot::{ 6 + ApiDescription, Body, ConfigDropshot, ConfigLogging, 7 + ConfigLoggingLevel, HttpError, HttpResponse, Query, RequestContext, 8 + ServerBuilder, ServerContext, endpoint, 9 + ClientErrorStatusCode, 10 + }; 11 + use http::{ 12 + Response, StatusCode, 13 + header::{ORIGIN, USER_AGENT}, 14 + }; 15 + use metrics::{counter, histogram}; 16 + use std::sync::Arc; 17 + 18 + use schemars::JsonSchema; 19 + use serde::{Deserialize, Serialize}; 20 + use tokio::time::Instant; 21 + use tokio_util::sync::CancellationToken; 22 + 23 + const INDEX_HTML: &str = include_str!("../static/index.html"); 24 + const FAVICON: &[u8] = include_bytes!("../static/favicon.ico"); 25 + 26 + pub async fn serve( 27 + cache: HybridCache<String, CachedRecord>, 28 + shutdown: CancellationToken, 29 + ) -> Result<(), ServerError> { 30 + let config_logging = ConfigLogging::StderrTerminal { 31 + level: ConfigLoggingLevel::Info, 32 + }; 33 + 34 + let log = config_logging 35 + .to_logger("example-basic") 36 + .map_err(ServerError::ConfigLogError)?; 37 + 38 + let mut api = ApiDescription::new(); 39 + api.register(index).unwrap(); 40 + api.register(favicon).unwrap(); 41 + api.register(openapi).unwrap(); 42 + api.register(get_record).unwrap(); 43 + 44 + // TODO: put spec in a once cell / lazy lock thing? 45 + let spec = Arc::new( 46 + api.openapi( 47 + "Slingshot", 48 + env!("CARGO_PKG_VERSION") 49 + .parse() 50 + .inspect_err(|e| { 51 + eprintln!("failed to parse cargo package version for openapi: {e:?}") 52 + }) 53 + .unwrap_or(semver::Version::new(0, 0, 1)), 54 + ) 55 + .description("A fast edge cache for getRecord") 56 + .contact_name("part of @microcosm.blue") 57 + .contact_url("https://microcosm.blue") 58 + .json() 59 + .map_err(ServerError::OpenApiJsonFail)?, 60 + ); 61 + 62 + let sub_shutdown = shutdown.clone(); 63 + let ctx = Context { 64 + cache, 65 + spec, 66 + shutdown: sub_shutdown, 67 + }; 68 + 69 + let server = ServerBuilder::new(api, ctx, log) 70 + .config(ConfigDropshot { 71 + bind_address: "0.0.0.0:9996".parse().unwrap(), 72 + ..Default::default() 73 + }) 74 + .start()?; 75 + 76 + tokio::select! { 77 + s = server.wait_for_shutdown() => { 78 + s.map_err(ServerError::ServerExited)?; 79 + log::info!("server shut down normally."); 80 + }, 81 + _ = shutdown.cancelled() => { 82 + log::info!("shutting down: closing server"); 83 + server.close().await.map_err(ServerError::BadClose)?; 84 + }, 85 + } 86 + Ok(()) 87 + } 88 + 89 + #[derive(Debug, Clone)] 90 + struct Context { 91 + pub cache: HybridCache<String, CachedRecord>, 92 + pub spec: Arc<serde_json::Value>, 93 + pub shutdown: CancellationToken, 94 + } 95 + 96 + async fn instrument_handler<T, H, R>(ctx: &RequestContext<T>, handler: H) -> Result<R, HttpError> 97 + where 98 + R: HttpResponse, 99 + H: Future<Output = Result<R, HttpError>>, 100 + T: ServerContext, 101 + { 102 + let start = Instant::now(); 103 + let result = handler.await; 104 + let latency = start.elapsed(); 105 + let status_code = match &result { 106 + Ok(response) => response.status_code(), 107 + Err(e) => e.status_code.as_status(), 108 + } 109 + .as_str() // just the number (.to_string()'s Display does eg `200 OK`) 110 + .to_string(); 111 + let endpoint = ctx.endpoint.operation_id.clone(); 112 + let headers = ctx.request.headers(); 113 + let origin = headers 114 + .get(ORIGIN) 115 + .and_then(|v| v.to_str().ok()) 116 + .unwrap_or("") 117 + .to_string(); 118 + let ua = headers 119 + .get(USER_AGENT) 120 + .and_then(|v| v.to_str().ok()) 121 + .map(|ua| { 122 + if ua.starts_with("Mozilla/5.0 ") { 123 + "browser" 124 + } else { 125 + ua 126 + } 127 + }) 128 + .unwrap_or("") 129 + .to_string(); 130 + counter!("server_requests_total", 131 + "endpoint" => endpoint.clone(), 132 + "origin" => origin, 133 + "ua" => ua, 134 + "status_code" => status_code, 135 + ) 136 + .increment(1); 137 + histogram!("server_handler_latency", "endpoint" => endpoint).record(latency.as_micros() as f64); 138 + result 139 + } 140 + 141 + use dropshot::{HttpResponseHeaders, HttpResponseOk}; 142 + 143 + pub type OkCorsResponse<T> = Result<HttpResponseHeaders<HttpResponseOk<T>>, HttpError>; 144 + 145 + /// Helper for constructing Ok responses: return OkCors(T).into() 146 + /// (not happy with this yet) 147 + pub struct OkCors<T: Serialize + JsonSchema + Send + Sync>(pub T); 148 + 149 + impl<T> From<OkCors<T>> for OkCorsResponse<T> 150 + where 151 + T: Serialize + JsonSchema + Send + Sync, 152 + { 153 + fn from(ok: OkCors<T>) -> OkCorsResponse<T> { 154 + let mut res = HttpResponseHeaders::new_unnamed(HttpResponseOk(ok.0)); 155 + res.headers_mut() 156 + .insert("access-control-allow-origin", "*".parse().unwrap()); 157 + Ok(res) 158 + } 159 + } 160 + 161 + pub fn cors_err(e: HttpError) -> HttpError { 162 + e.with_header("access-control-allow-origin", "*").unwrap() 163 + } 164 + 165 + 166 + // TODO: cors for HttpError 167 + 168 + /// Serve index page as html 169 + #[endpoint { 170 + method = GET, 171 + path = "/", 172 + /* 173 + * not useful to have this in openapi 174 + */ 175 + unpublished = true, 176 + }] 177 + async fn index(ctx: RequestContext<Context>) -> Result<Response<Body>, HttpError> { 178 + instrument_handler(&ctx, async { 179 + Ok(Response::builder() 180 + .status(StatusCode::OK) 181 + .header(http::header::CONTENT_TYPE, "text/html") 182 + .body(INDEX_HTML.into())?) 183 + }) 184 + .await 185 + } 186 + 187 + /// Serve index page as html 188 + #[endpoint { 189 + method = GET, 190 + path = "/favicon.ico", 191 + /* 192 + * not useful to have this in openapi 193 + */ 194 + unpublished = true, 195 + }] 196 + async fn favicon(ctx: RequestContext<Context>) -> Result<Response<Body>, HttpError> { 197 + instrument_handler(&ctx, async { 198 + Ok(Response::builder() 199 + .status(StatusCode::OK) 200 + .header(http::header::CONTENT_TYPE, "image/x-icon") 201 + .body(FAVICON.to_vec().into())?) 202 + }) 203 + .await 204 + } 205 + 206 + /// Meta: get the openapi spec for this api 207 + #[endpoint { 208 + method = GET, 209 + path = "/openapi", 210 + /* 211 + * not useful to have this in openapi 212 + */ 213 + unpublished = true, 214 + }] 215 + async fn openapi(ctx: RequestContext<Context>) -> OkCorsResponse<serde_json::Value> { 216 + instrument_handler(&ctx, async { 217 + let spec = (*ctx.context().spec).clone(); 218 + OkCors(spec).into() 219 + }) 220 + .await 221 + } 222 + 223 + 224 + #[derive(Debug, Deserialize, JsonSchema)] 225 + struct GetRecordQuery { 226 + /// The DID of the repo 227 + /// 228 + /// NOTE: handles should be accepted here but this is still TODO in slingshot 229 + pub repo: String, 230 + /// The NSID of the record collection 231 + pub collection: String, 232 + /// The Record key 233 + pub rkey: String, 234 + /// Optional: the CID of the version of the record. 235 + /// 236 + /// If not specified, then return the most recent version. 237 + /// 238 + /// If specified and a newer version of the record exists, returns 404 not 239 + /// found. That is: slingshot only retains the most recent version of a 240 + /// record. 241 + #[serde(default)] 242 + pub cid: Option<String>, 243 + } 244 + 245 + #[derive(Debug, Serialize, JsonSchema)] 246 + struct GetRecordResponse { 247 + pub uri: String, 248 + pub cid: String, 249 + pub value: Box<RawValue>, 250 + } 251 + 252 + /// com.atproto.repo.getRecord 253 + /// 254 + /// Get a single record from a repository. Does not require auth. 255 + /// 256 + /// See https://docs.bsky.app/docs/api/com-atproto-repo-get-record for the 257 + /// canonical XRPC documentation that this endpoint aims to be compatible with. 258 + #[endpoint { 259 + method = GET, 260 + path = "/xrpc/com.atproto.repo.getRecord", 261 + }] 262 + async fn get_record( 263 + ctx: RequestContext<Context>, 264 + query: Query<GetRecordQuery>, 265 + ) -> OkCorsResponse<GetRecordResponse> { 266 + 267 + let Context { cache, .. } = ctx.context(); 268 + let GetRecordQuery { repo, collection, rkey, cid } = query.into_inner(); 269 + 270 + // TODO: yeah yeah 271 + let at_uri = format!( 272 + "at://{}/{}/{}", 273 + &*repo, &*collection, &*rkey 274 + ); 275 + 276 + instrument_handler(&ctx, async { 277 + let entry = cache 278 + .fetch(at_uri.clone(), || async move { 279 + Err(foyer::Error::Other(Box::new(ServerError::OhNo("booo".to_string())))) 280 + }) 281 + .await 282 + .unwrap(); 283 + 284 + match *entry { 285 + CachedRecord::Found(ref raw) => { 286 + let (found_cid, raw_value) = raw.into(); 287 + let found_cid = found_cid.as_ref().to_string(); 288 + if cid.map(|c| c != found_cid).unwrap_or(false) { 289 + Err(HttpError::for_not_found(None, "CID mismatch".to_string())) 290 + .map_err(cors_err)?; 291 + } 292 + OkCors(GetRecordResponse { 293 + uri: at_uri, 294 + cid: found_cid, 295 + value: raw_value, 296 + }).into() 297 + }, 298 + CachedRecord::Deleted => { 299 + Err(HttpError::for_client_error_with_status( 300 + Some("Gone".to_string()), 301 + ClientErrorStatusCode::GONE, 302 + )).map_err(cors_err) 303 + } 304 + } 305 + }) 306 + .await 307 + 308 + }

slingshot/static/favicon.ico

This is a binary file and will not be displayed.

+53

slingshot/static/index.html

··· 1 + <!doctype html> 2 + <html lang="en"> 3 + <head> 4 + <meta charset="utf-8" /> 5 + <title>Slingshot documentation</title> 6 + <meta name="viewport" content="width=device-width, initial-scale=1" /> 7 + <meta name="description" content="API Documentation for Slingshot, a fast edge cache for getRecord" /> 8 + <style> 9 + .custom-header { 10 + height: 42px; 11 + background-color: #221828; 12 + box-shadow: inset 0 -1px 0 var(--scalar-border-color); 13 + color: var(--scalar-color-1); 14 + font-size: var(--scalar-font-size-3); 15 + font-family: 'Iowan Old Style', 'Palatino Linotype', 'URW Palladio L', P052, serif; 16 + padding: 0 18px; 17 + justify-content: space-between; 18 + } 19 + .custom-header, 20 + .custom-header nav { 21 + display: flex; 22 + align-items: center; 23 + gap: 18px; 24 + } 25 + .custom-header a:hover { 26 + color: var(--scalar-color-2); 27 + } 28 + </style> 29 + </head> 30 + <body> 31 + <header class="custom-header scalar-app"> 32 + <p> 33 + <a href="https://microcosm.blue">todo: what link goes here?</a>: blah 34 + </p> 35 + <nav> 36 + <b>a <a href="https://microcosm.blue">microcosm</a> project</b> 37 + <a href="https://bsky.app/profile/microcosm.blue">@microcosm.blue</a> 38 + <a href="https://github.com/at-microcosm">github</a> 39 + </nav> 40 + </header> 41 + 42 + <script id="api-reference" type="application/json" data-url="/openapi"></script> 43 + 44 + <script> 45 + var configuration = { 46 + theme: 'purple', 47 + } 48 + document.getElementById('api-reference').dataset.configuration = JSON.stringify(configuration) 49 + </script> 50 + 51 + <script src="https://cdn.jsdelivr.net/npm/@scalar/api-reference"></script> 52 + </body> 53 + </html>

Configure Feed

Configure Feed