+81 -27

Cargo.lock

reviewed

··· 654 654 655 655 [[package]] 656 656 name = "clap" 657 657 - version = "4.5.35" 657 657 + version = "4.5.40" 658 658 source = "registry+https://github.com/rust-lang/crates.io-index" 659 659 - checksum = "d8aa86934b44c19c50f87cc2790e19f54f7a67aedb64101c2e1a2e5ecfb73944" 659 659 + checksum = "40b6887a1d8685cebccf115538db5c0efe625ccac9696ad45c409d96566e910f" 660 660 dependencies = [ 661 661 "clap_builder", 662 662 "clap_derive", ··· 664 664 665 665 [[package]] 666 666 name = "clap_builder" 667 667 - version = "4.5.35" 667 667 + version = "4.5.40" 668 668 source = "registry+https://github.com/rust-lang/crates.io-index" 669 669 - checksum = "2414dbb2dd0695280da6ea9261e327479e9d37b0630f6b53ba2a11c60c679fd9" 669 669 + checksum = "e0c66c08ce9f0c698cbce5c0279d0bb6ac936d8674174fe48f736533b964f59e" 670 670 dependencies = [ 671 671 "anstream", 672 672 "anstyle", ··· 676 676 677 677 [[package]] 678 678 name = "clap_derive" 679 679 - version = "4.5.32" 679 679 + version = "4.5.40" 680 680 source = "registry+https://github.com/rust-lang/crates.io-index" 681 681 - checksum = "09176aae279615badda0765c0c0b3f6ed53f4709118af73cf4655d85d1530cd7" 681 681 + checksum = "d2c7947ae4cc3d851207c1adb5b5e260ff0cca11446b1d6d1423788e442257ce" 682 682 dependencies = [ 683 683 "heck", 684 684 "proc-macro2", ··· 755 755 "tokio", 756 756 "tokio-util", 757 757 "tower-http", 758 758 - "tungstenite", 758 758 + "tungstenite 0.26.2", 759 759 "zstd", 760 760 ] 761 761 ··· 988 988 989 989 [[package]] 990 990 name = "dropshot" 991 991 - version = "0.16.0" 991 991 + version = "0.16.2" 992 992 source = "registry+https://github.com/rust-lang/crates.io-index" 993 993 - checksum = "a37c505dad56e0c1fa5ed47e29fab1a1ab2d1a9d93e952024bb47168969705f6" 993 993 + checksum = "50e8fed669e35e757646ad10f97c4d26dd22cce3da689b307954f7000d2719d0" 994 994 dependencies = [ 995 995 "async-stream", 996 996 "async-trait", ··· 1038 1038 1039 1039 [[package]] 1040 1040 name = "dropshot_endpoint" 1041 1041 - version = "0.16.0" 1041 1041 + version = "0.16.2" 1042 1042 source = "registry+https://github.com/rust-lang/crates.io-index" 1043 1043 - checksum = "8b1a6db3728f0195e3ad62807649913aaba06d45421e883416e555e51464ef67" 1043 1043 + checksum = "acebb687581abdeaa2c89fa448818a5f803b0e68e5d7e7a1cf585a8f3c5c57ac" 1044 1044 dependencies = [ 1045 1045 "heck", 1046 1046 "proc-macro2", ··· 1872 1872 "serde_json", 1873 1873 "thiserror 2.0.12", 1874 1874 "tokio", 1875 1875 - "tokio-tungstenite", 1875 1875 + "tokio-tungstenite 0.26.2", 1876 1876 "url", 1877 1877 "zstd", 1878 1878 ] ··· 3184 3184 3185 3185 [[package]] 3186 3186 name = "serde_spanned" 3187 3187 - version = "0.6.8" 3187 3187 + version = "0.6.9" 3188 3188 source = "registry+https://github.com/rust-lang/crates.io-index" 3189 3189 - checksum = "87607cb1398ed59d48732e575a4c28a7a8ebf2454b964fe3f224f2afc07909e1" 3189 3189 + checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3" 3190 3190 dependencies = [ 3191 3191 "serde", 3192 3192 ] ··· 3378 3378 ] 3379 3379 3380 3380 [[package]] 3381 3381 + name = "spacedust" 3382 3382 + version = "0.1.0" 3383 3383 + dependencies = [ 3384 3384 + "clap", 3385 3385 + "dropshot", 3386 3386 + "futures", 3387 3387 + "jetstream", 3388 3388 + "links", 3389 3389 + "metrics", 3390 3390 + "schemars", 3391 3391 + "serde", 3392 3392 + "serde_json", 3393 3393 + "tinyjson", 3394 3394 + "tokio", 3395 3395 + "tokio-tungstenite 0.27.0", 3396 3396 + ] 3397 3397 + 3398 3398 + [[package]] 3381 3399 name = "spin" 3382 3400 version = "0.9.8" 3383 3401 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 3412 3430 3413 3431 [[package]] 3414 3432 name = "syn" 3415 3415 - version = "2.0.100" 3433 3433 + version = "2.0.103" 3416 3434 source = "registry+https://github.com/rust-lang/crates.io-index" 3417 3417 - checksum = "b09a44accad81e1ba1cd74a32461ba89dee89095ba17b32f5d03683b1b1fc2a0" 3435 3435 + checksum = "e4307e30089d6fd6aff212f2da3a1f9e32f3223b1f010fb09b7c95f90f3ca1e8" 3418 3436 dependencies = [ 3419 3437 "proc-macro2", 3420 3438 "quote", ··· 3595 3613 3596 3614 [[package]] 3597 3615 name = "tokio" 3598 3598 - version = "1.44.2" 3616 3616 + version = "1.45.1" 3599 3617 source = "registry+https://github.com/rust-lang/crates.io-index" 3600 3600 - checksum = "e6b88822cbe49de4185e3a4cbf8321dd487cf5fe0c5c65695fef6346371e9c48" 3618 3618 + checksum = "75ef51a33ef1da925cea3e4eb122833cb377c61439ca401b770f54902b806779" 3601 3619 dependencies = [ 3602 3620 "backtrace", 3603 3621 "bytes", ··· 3654 3672 "native-tls", 3655 3673 "tokio", 3656 3674 "tokio-native-tls", 3657 3657 - "tungstenite", 3675 3675 + "tungstenite 0.26.2", 3676 3676 + ] 3677 3677 + 3678 3678 + [[package]] 3679 3679 + name = "tokio-tungstenite" 3680 3680 + version = "0.27.0" 3681 3681 + source = "registry+https://github.com/rust-lang/crates.io-index" 3682 3682 + checksum = "489a59b6730eda1b0171fcfda8b121f4bee2b35cba8645ca35c5f7ba3eb736c1" 3683 3683 + dependencies = [ 3684 3684 + "futures-util", 3685 3685 + "log", 3686 3686 + "tokio", 3687 3687 + "tungstenite 0.27.0", 3658 3688 ] 3659 3689 3660 3690 [[package]] ··· 3672 3702 3673 3703 [[package]] 3674 3704 name = "toml" 3675 3675 - version = "0.8.20" 3705 3705 + version = "0.8.23" 3676 3706 source = "registry+https://github.com/rust-lang/crates.io-index" 3677 3677 - checksum = "cd87a5cdd6ffab733b2f74bc4fd7ee5fff6634124999ac278c35fc78c6120148" 3707 3707 + checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362" 3678 3708 dependencies = [ 3679 3709 "serde", 3680 3710 "serde_spanned", ··· 3684 3714 3685 3715 [[package]] 3686 3716 name = "toml_datetime" 3687 3687 - version = "0.6.8" 3717 3717 + version = "0.6.11" 3688 3718 source = "registry+https://github.com/rust-lang/crates.io-index" 3689 3689 - checksum = "0dd7358ecb8fc2f8d014bf86f6f638ce72ba252a2c3a2572f2a795f1d23efb41" 3719 3719 + checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c" 3690 3720 dependencies = [ 3691 3721 "serde", 3692 3722 ] 3693 3723 3694 3724 [[package]] 3695 3725 name = "toml_edit" 3696 3696 - version = "0.22.24" 3726 3726 + version = "0.22.27" 3697 3727 source = "registry+https://github.com/rust-lang/crates.io-index" 3698 3698 - checksum = "17b4795ff5edd201c7cd6dca065ae59972ce77d1b80fa0a84d94950ece7d1474" 3728 3728 + checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a" 3699 3729 dependencies = [ 3700 3730 "indexmap 2.9.0", 3701 3731 "serde", 3702 3732 "serde_spanned", 3703 3733 "toml_datetime", 3734 3734 + "toml_write", 3704 3735 "winnow", 3705 3736 ] 3737 3737 + 3738 3738 + [[package]] 3739 3739 + name = "toml_write" 3740 3740 + version = "0.1.2" 3741 3741 + source = "registry+https://github.com/rust-lang/crates.io-index" 3742 3742 + checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" 3706 3743 3707 3744 [[package]] 3708 3745 name = "tower" ··· 3829 3866 "sha1", 3830 3867 "thiserror 2.0.12", 3831 3868 "url", 3869 3869 + "utf-8", 3870 3870 + ] 3871 3871 + 3872 3872 + [[package]] 3873 3873 + name = "tungstenite" 3874 3874 + version = "0.27.0" 3875 3875 + source = "registry+https://github.com/rust-lang/crates.io-index" 3876 3876 + checksum = "eadc29d668c91fcc564941132e17b28a7ceb2f3ebf0b9dae3e03fd7a6748eb0d" 3877 3877 + dependencies = [ 3878 3878 + "bytes", 3879 3879 + "data-encoding", 3880 3880 + "http", 3881 3881 + "httparse", 3882 3882 + "log", 3883 3883 + "rand 0.9.1", 3884 3884 + "sha1", 3885 3885 + "thiserror 2.0.12", 3832 3886 "utf-8", 3833 3887 ] 3834 3888 ··· 4346 4400 4347 4401 [[package]] 4348 4402 name = "winnow" 4349 4349 - version = "0.7.6" 4403 4403 + version = "0.7.11" 4350 4404 source = "registry+https://github.com/rust-lang/crates.io-index" 4351 4351 - checksum = "63d3fcd9bba44b03821e7d699eeee959f3126dcc4aa8e4ae18ec617c2a5cea10" 4405 4405 + checksum = "74c7b26e3480b707944fc872477815d29a8e429d2f93a1ce000f5fa84a15cbcd" 4352 4406 dependencies = [ 4353 4407 "memchr", 4354 4408 ]

+1

Cargo.toml

reviewed

··· 6 6 "jetstream", 7 7 "ufos", 8 8 "ufos/fuzz", 9 9 + "spacedust", 9 10 ]

+18

spacedust/Cargo.toml

reviewed

··· 1 1 + [package] 2 2 + name = "spacedust" 3 3 + version = "0.1.0" 4 4 + edition = "2024" 5 5 + 6 6 + [dependencies] 7 7 + clap = { version = "4.5.40", features = ["derive"] } 8 8 + dropshot = "0.16.2" 9 9 + futures = "0.3.31" 10 10 + jetstream = { path = "../jetstream", features = ["metrics"] } 11 11 + links = { path = "../links" } 12 12 + metrics = "0.24.2" 13 13 + schemars = "0.8.22" 14 14 + serde = { version = "1.0.219", features = ["derive"] } 15 15 + serde_json = "1.0.140" 16 16 + tinyjson = "2.5.1" 17 17 + tokio = { version = "1.45.1", features = ["full"] } 18 18 + tokio-tungstenite = "0.27.0"

+82

spacedust/src/consumer.rs

reviewed

··· 1 1 + use crate::LinkEvent; 2 2 + use jetstream::{ 3 3 + DefaultJetstreamEndpoints, JetstreamCompression, JetstreamConfig, JetstreamConnector, 4 4 + events::{CommitOp, Cursor, EventKind}, 5 5 + }; 6 6 + use links::collect_links; 7 7 + use std::error::Error; 8 8 + use tokio::sync::broadcast; 9 9 + 10 10 + const MAX_LINKS_PER_EVENT: usize = 100; 11 11 + 12 12 + pub async fn consume( 13 13 + b: broadcast::Sender<LinkEvent>, 14 14 + jetstream_endpoint: &str, 15 15 + cursor: Option<Cursor>, 16 16 + no_zstd: bool, 17 17 + ) -> Result<(), Box<dyn Error>> { 18 18 + let endpoint = DefaultJetstreamEndpoints::endpoint_or_shortcut(jetstream_endpoint); 19 19 + if endpoint == jetstream_endpoint { 20 20 + std::println!("connecting to jetstream at {endpoint}"); 21 21 + } else { 22 22 + std::println!("connecting to jetstream at {jetstream_endpoint} => {endpoint}"); 23 23 + } 24 24 + let config: JetstreamConfig = JetstreamConfig { 25 25 + endpoint, 26 26 + compression: if no_zstd { 27 27 + JetstreamCompression::None 28 28 + } else { 29 29 + JetstreamCompression::Zstd 30 30 + }, 31 31 + replay_on_reconnect: true, 32 32 + channel_size: 1024, // buffer up to ~1s of jetstream events 33 33 + ..Default::default() 34 34 + }; 35 35 + let mut receiver = JetstreamConnector::new(config)? 36 36 + .connect_cursor(cursor) 37 37 + .await?; 38 38 + 39 39 + while let Some(event) = receiver.recv().await { 40 40 + if event.kind != EventKind::Commit { 41 41 + continue; 42 42 + } 43 43 + let Some(commit) = event.commit else { 44 44 + eprintln!("jetstream commit event missing commit data, ignoring"); 45 45 + continue; 46 46 + }; 47 47 + 48 48 + // TODO: keep a buffer and remove quick deletes to debounce notifs 49 49 + // for now we just drop all deletes eek 50 50 + if commit.operation == CommitOp::Delete { 51 51 + continue; 52 52 + } 53 53 + let Some(record) = commit.record else { 54 54 + eprintln!("jetstream commit update/delete missing record, ignoring"); 55 55 + continue; 56 56 + }; 57 57 + 58 58 + let jv = record.get().parse()?; 59 59 + 60 60 + // todo: indicate if the link limit was reached (-> links omitted) 61 61 + for (i, link) in collect_links(&jv).into_iter().enumerate() { 62 62 + if i >= MAX_LINKS_PER_EVENT { 63 63 + eprintln!("jetstream event has too many links, ignoring the rest"); 64 64 + break; 65 65 + } 66 66 + let link_ev = LinkEvent { 67 67 + collection: commit.collection.to_string(), 68 68 + path: link.path, 69 69 + origin: format!( 70 70 + "at://{}/{}/{}", 71 71 + &*event.did, 72 72 + &*commit.collection, 73 73 + &*commit.rkey, 74 74 + ), 75 75 + target: link.target.into_string(), 76 76 + }; 77 77 + let _ = b.send(link_ev); // only errors if no subscribers are connected, which is just fine. 78 78 + } 79 79 + } 80 80 + 81 81 + Err("jetstream consumer ended".into()) 82 82 + }

+31

spacedust/src/lib.rs

reviewed

··· 1 1 + pub mod consumer; 2 2 + pub mod server; 3 3 + 4 4 + use serde::Serialize; 5 5 + 6 6 + #[derive(Debug, Clone)] 7 7 + pub struct LinkEvent { 8 8 + collection: String, 9 9 + path: String, 10 10 + origin: String, 11 11 + target: String, 12 12 + } 13 13 + 14 14 + #[derive(Debug, Serialize)] 15 15 + struct ClientEvent { 16 16 + source: String, 17 17 + origin: String, 18 18 + target: String, 19 19 + // TODO: include the record too? would save clients a level of hydration 20 20 + } 21 21 + 22 22 + impl From<LinkEvent> for ClientEvent { 23 23 + fn from(link: LinkEvent) -> Self { 24 24 + let undotted = link.path.get(1..).unwrap_or(""); 25 25 + Self { 26 26 + source: format!("{}:{undotted}", link.collection), 27 27 + origin: link.origin, 28 28 + target: link.target, 29 29 + } 30 30 + } 31 31 + }

+54

spacedust/src/main.rs

reviewed

··· 1 1 + use spacedust::consumer; 2 2 + use spacedust::server; 3 3 + 4 4 + use clap::Parser; 5 5 + use tokio::sync::broadcast; 6 6 + 7 7 + /// Aggregate links in the at-mosphere 8 8 + #[derive(Parser, Debug, Clone)] 9 9 + #[command(version, about, long_about = None)] 10 10 + struct Args { 11 11 + /// Jetstream server to connect to (exclusive with --fixture). Provide either a wss:// URL, or a shorhand value: 12 12 + /// 'us-east-1', 'us-east-2', 'us-west-1', or 'us-west-2' 13 13 + #[arg(long)] 14 14 + jetstream: String, 15 15 + /// don't request zstd-compressed jetstream events 16 16 + /// 17 17 + /// reduces CPU at the expense of more ingress bandwidth 18 18 + #[arg(long, action)] 19 19 + jetstream_no_zstd: bool, 20 20 + } 21 21 + 22 22 + #[tokio::main] 23 23 + async fn main() -> Result<(), String> { 24 24 + let args = Args::parse(); 25 25 + 26 26 + // tokio broadcast keeps a single main output queue for all subscribers. 27 27 + // each subscriber clones off a copy of an individual value for each recv. 28 28 + // since there's no large per-client buffer, we can make this one kind of 29 29 + // big and accommodate more slow/bursty clients. 30 30 + // 31 31 + // in fact, we *could* even keep lagging clients alive, inserting lag- 32 32 + // indicating messages to their output.... but for now we'll drop them to 33 33 + // avoid accumulating zombies. 34 34 + // 35 35 + // events on the channel are individual links as they are discovered. a link 36 36 + // contains a source and a target. the target is an at-uri, so it's up to 37 37 + // ~1KB max; source is a collection + link path, which can be more but in 38 38 + // practice the whole link rarely approaches 1KB total. 39 39 + // 40 40 + // TODO: determine if a pathological case could blow this up (eg 1MB link 41 41 + // paths + slow subscriber -> 16GiB queue) 42 42 + let (b, _) = broadcast::channel(16_384); 43 43 + 44 44 + let consuming = consumer::consume(b.clone(), &args.jetstream, None, args.jetstream_no_zstd); 45 45 + 46 46 + let serving = server::serve(b); 47 47 + 48 48 + tokio::select! { 49 49 + e = serving => eprintln!("serving failed: {e:?}"), 50 50 + e = consuming => eprintln!("consuming failed: {e:?}"), 51 51 + }; 52 52 + 53 53 + Ok(()) 54 54 + }

+83

spacedust/src/server.rs

reviewed

··· 1 1 + use crate::{ClientEvent, LinkEvent}; 2 2 + use dropshot::{ 3 3 + ApiDescription, ConfigDropshot, ConfigLogging, ConfigLoggingLevel, Query, RequestContext, 4 4 + ServerBuilder, WebsocketConnection, channel, 5 5 + }; 6 6 + use futures::SinkExt; 7 7 + use schemars::JsonSchema; 8 8 + use serde::Deserialize; 9 9 + use tokio::sync::broadcast; 10 10 + use tokio_tungstenite::tungstenite::Message; 11 11 + use tokio_tungstenite::tungstenite::protocol::Role; 12 12 + 13 13 + pub async fn serve(b: broadcast::Sender<LinkEvent>) -> Result<(), String> { 14 14 + let config_logging = ConfigLogging::StderrTerminal { 15 15 + level: ConfigLoggingLevel::Info, 16 16 + }; 17 17 + 18 18 + let log = config_logging 19 19 + .to_logger("example-basic") 20 20 + .map_err(|error| format!("failed to create logger: {}", error))?; 21 21 + 22 22 + let mut api = ApiDescription::new(); 23 23 + api.register(subscribe).unwrap(); 24 24 + 25 25 + let server = ServerBuilder::new(api, b, log) 26 26 + .config(ConfigDropshot { 27 27 + bind_address: "0.0.0.0:9998".parse().unwrap(), 28 28 + ..Default::default() 29 29 + }) 30 30 + .start() 31 31 + .map_err(|error| format!("failed to create server: {}", error))?; 32 32 + 33 33 + server.await 34 34 + } 35 35 + 36 36 + #[derive(Deserialize, JsonSchema)] 37 37 + struct QueryParams { 38 38 + _hello: Option<String>, 39 39 + } 40 40 + 41 41 + #[channel { 42 42 + protocol = WEBSOCKETS, 43 43 + path = "/subscribe", 44 44 + }] 45 45 + async fn subscribe( 46 46 + ctx: RequestContext<broadcast::Sender<LinkEvent>>, 47 47 + _qp: Query<QueryParams>, 48 48 + upgraded: WebsocketConnection, 49 49 + ) -> dropshot::WebsocketChannelResult { 50 50 + let mut ws = tokio_tungstenite::WebSocketStream::from_raw_socket( 51 51 + upgraded.into_inner(), 52 52 + Role::Server, 53 53 + None, 54 54 + ) 55 55 + .await; 56 56 + let mut sub = ctx.context().subscribe(); 57 57 + 58 58 + // TODO: pingpong 59 59 + // TODO: filtering subscription 60 60 + 61 61 + loop { 62 62 + match sub.recv().await { 63 63 + Ok(link) => { 64 64 + let json = serde_json::to_string::<ClientEvent>(&link.into())?; 65 65 + if let Err(e) = ws.send(Message::Text(json.into())).await { 66 66 + eprintln!("client: failed to send event: {e:?}"); 67 67 + ws.close(None).await?; // TODO: do we need this one?? 68 68 + break; 69 69 + } 70 70 + } 71 71 + Err(broadcast::error::RecvError::Closed) => { 72 72 + ws.close(None).await?; // TODO: send reason 73 73 + break; 74 74 + } 75 75 + Err(broadcast::error::RecvError::Lagged(_n_missed)) => { 76 76 + eprintln!("client lagged, closing"); 77 77 + ws.close(None).await?; // TODO: send reason 78 78 + break; 79 79 + } 80 80 + } 81 81 + } 82 82 + Ok(()) 83 83 + }