rsky-relay: resolver/verify

Author: utkarshgupta137

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = [ "rsky-common", "rsky-crypto","rsky-feedgen", "rsky-firehose", "rsky-identity", "rsky-labeler", "rsky-lexicon", "rsky-pds", "rsky-syntax", "rsky-jetstream-subscriber", "rsky-repo", "cypher/backend", "cypher/frontend", "rsky-satnav"]
+members = ["rsky-common", "rsky-crypto", "rsky-identity", "rsky-lexicon", "rsky-relay", "rsky-satnav"]
 resolver = "2"
 
 [workspace.dependencies]

rsky-common/src/lib.rs

@@ -91,7 +91,7 @@ pub fn get_handle(doc: &DidDocument) -> Option<String> {
 
 pub fn get_verification_material(
     doc: &DidDocument,
-    key_id: &String,
+    key_id: &str,
 ) -> Option<VerificationMaterial> {
     let did = get_did(doc);
     let keys = &doc.verification_method;

rsky-identity/Cargo.toml

@@ -12,7 +12,7 @@ documentation = "https://docs.rs/rsky-identity"
 
 [dependencies]
 anyhow = "1.0.82"
-reqwest = { version = "0.12.3", features = ["json","blocking"] }
+reqwest = { version = "0.12", default-features = false, features = ["gzip", "hickory-dns", "http2", "json", "rustls-tls-webpki-roots-no-provider"] }
 serde_json = { version = "1.0.115",features = ["preserve_order"] }
 urlencoding = "2.1.3"
 thiserror = "1.0.58"

rsky-relay/Cargo.toml

@@ -7,21 +7,43 @@ edition = "2024"
 # external
 bstr = "1"
 bus = "2"
+chrono = { version = "0.4", features = ["serde"] }
+ciborium = "0.2"
+cid = { version = "0.11", features = ["serde-codec"] }
 color-eyre = "0.6"
 derive_more = { version = "2", features = ["full"] }
+futures = { version = "0.3", default-features = false, features = ["std"] }
 hashbrown = "0.15"
 http = "1"
 httparse = "1.10"
+k256 = "0.13"
 magnetic = "2"
+mimalloc = "0.1"
+multibase = "0.9"
 nix = { version = "0.29", features = ["event", "socket"] }
+p256 = "0.13"
+reqwest = { version = "0.12", default-features = false, features = ["gzip", "hickory-dns", "http2", "json", "rustls-tls-webpki-roots-no-provider"] }
+rs-car-sync = "0.4"
 rtrb = "0.3"
 rustls = "0.23"
 serde = { version = "1", features = ["derive"] }
+serde_bytes = "0.11"
+serde_cbor = "0.11"
+serde_ipld_dagcbor = "0.6"
 serde_json = "1"
 signal-hook = { version = "0.3", features = ["extended-siginfo"] }
+sled = { version = "0.34", features = ["compression"] }
 thingbuf = "0.1"
 thiserror = "2"
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] }
 tracing = { version = "0.1", features = ["release_max_level_debug"] }
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
-tungstenite = { version = "0.26", features = ["rustls-tls-native-roots"] }
+tungstenite = { version = "0.26", features = ["rustls-tls-webpki-roots"] }
 url = "2"
+zerocopy = { version = "0.8", features = ["derive"] }
+
+# internal
+rsky-common = { workspace = true }
+rsky-crypto = { workspace = true }
+rsky-identity = { workspace = true }
+rsky-lexicon = { workspace = true }

rsky-relay/src/crawler/connection.rs

@@ -73,7 +73,10 @@ impl Connection {
             let bytes = match msg {
                 Message::Binary(bytes) => bytes,
                 Message::Close(_) => todo!(),
-                _ => unreachable!("{msg}"),
+                _ => {
+                    tracing::debug!("unknown message: {msg}");
+                    continue;
+                },
             };
 
             let mut slot = self.message_tx.try_send_ref()?;

rsky-relay/src/crawler/manager.rs

@@ -1,5 +1,6 @@
 use std::sync::atomic::Ordering;
 use std::thread;
+use std::time::Duration;
 
 use hashbrown::HashMap;
 use http::Uri;
@@ -15,6 +16,7 @@ use crate::crawler::worker::{Worker, WorkerError};
 use crate::types::{Cursor, MessageSender, RequestCrawlReceiver};
 
 const CAPACITY: usize = 1024;
+const SLEEP: Duration = Duration::from_millis(10);
 
 #[derive(Debug, Error)]
 pub enum ManagerError {
@@ -68,7 +70,9 @@ impl Manager {
     }
 
     pub fn run(mut self) -> Result<(), ManagerError> {
-        while self.update()? {}
+        while self.update()? {
+            thread::sleep(SLEEP);
+        }
         self.shutdown()
     }
 

rsky-relay/src/crawler/worker.rs

@@ -54,6 +54,7 @@ impl Worker {
         while self.update() {
             thread::yield_now();
         }
+        tracing::info!("shutting down crawler: {}", self.worker_id.0);
         self.shutdown()
     }
 
@@ -89,7 +90,6 @@ impl Worker {
                 }
             }
             Command::Shutdown => {
-                tracing::debug!("shutting down crawler: {}", self.worker_id.0);
                 return false;
             }
         }

rsky-relay/src/lib.rs

@@ -2,6 +2,7 @@ mod crawler;
 mod publisher;
 mod server;
 mod types;
+mod validator;
 
 use std::sync::atomic::AtomicBool;
 
@@ -11,3 +12,4 @@ pub use crawler::Manager as CrawlerManager;
 pub use publisher::Manager as PublisherManager;
 pub use server::Server;
 pub use types::{MessageRecycle, RequestCrawl};
+pub use validator::Manager as ValidatorManager;

rsky-relay/src/main.rs

@@ -4,21 +4,32 @@ use std::thread;
 
 use color_eyre::Result;
 
-use rsky_relay::{CrawlerManager, MessageRecycle, PublisherManager, SHUTDOWN, Server};
+use mimalloc::MiMalloc;
+use rustls::crypto::aws_lc_rs::default_provider;
 use signal_hook::consts::{SIGINT, TERM_SIGNALS};
 use signal_hook::flag;
 use signal_hook::iterator::SignalsInfo;
 use signal_hook::iterator::exfiltrator::WithOrigin;
 
+use rsky_relay::{
+    CrawlerManager, MessageRecycle, PublisherManager, SHUTDOWN, Server, ValidatorManager,
+};
+
 const CAPACITY1: usize = 1 << 20;
 const CAPACITY2: usize = 1 << 10;
 const WORKERS: usize = 4;
 
-pub fn main() -> Result<()> {
+#[global_allocator]
+static GLOBAL: MiMalloc = MiMalloc;
+
+#[tokio::main]
+pub async fn main() -> Result<()> {
     tracing_subscriber::fmt()
         .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
         .init();
 
+    default_provider().install_default().unwrap();
+
     let terminate_now = Arc::new(AtomicBool::new(false));
     flag::register_conditional_shutdown(SIGINT, 1, Arc::clone(&terminate_now))?;
     flag::register(SIGINT, Arc::clone(&terminate_now))?;
@@ -27,10 +38,12 @@ pub fn main() -> Result<()> {
         thingbuf::mpsc::blocking::with_recycle(CAPACITY1, MessageRecycle);
     let (request_crawl_tx, request_crawl_rx) = rtrb::RingBuffer::new(CAPACITY2);
     let (subscribe_repos_tx, subscribe_repos_rx) = rtrb::RingBuffer::new(CAPACITY2);
+    let mut validator = ValidatorManager::new(message_rx).await?;
     let crawler = CrawlerManager::new(WORKERS, message_tx, request_crawl_rx)?;
-    let publisher = PublisherManager::new(WORKERS, message_rx, subscribe_repos_rx)?;
+    let publisher = PublisherManager::new(WORKERS, &mut validator, subscribe_repos_rx)?;
     let server = Server::new(request_crawl_tx, subscribe_repos_tx)?;
     thread::scope(move |s| {
+        tokio::spawn(validator.run());
         s.spawn(move || crawler.run());
         s.spawn(move || publisher.run());
         s.spawn(move || server.run());

rsky-relay/src/publisher/connection.rs

@@ -69,7 +69,7 @@ impl Connection {
 
     pub fn send(&mut self, input: &[u8]) -> Result<(), ConnectionError> {
         match self.client.send(Message::binary(input.to_vec())) {
-            Ok(()) => (),
+            Ok(()) => {},
             Err(tungstenite::Error::WriteBufferFull(msg)) => {
                 self.queue.push_back((Instant::now(), msg));
                 if self.queue.len() > MAX_LEN {
@@ -91,7 +91,7 @@ impl Connection {
     pub fn poll(&mut self) -> Result<(), ConnectionError> {
         while let Some((instant, msg)) = self.queue.pop_front() {
             match self.client.send(msg) {
-                Ok(()) => (),
+                Ok(()) => {},
                 Err(tungstenite::Error::WriteBufferFull(msg)) => {
                     self.queue.push_back((instant, msg));
                     break;

rsky-relay/src/publisher/manager.rs

@@ -1,20 +1,20 @@
-use std::sync::Arc;
 use std::sync::atomic::Ordering;
 use std::thread;
+use std::time::Duration;
 
-use bus::Bus;
 use magnetic::Consumer;
 use magnetic::buffer::dynamic::DynamicBufferP2;
 use thiserror::Error;
 
-use crate::SHUTDOWN;
 use crate::publisher::types::{
     Command, CommandSender, Config, LocalId, Status, StatusReceiver, WorkerId,
 };
 use crate::publisher::worker::{Worker, WorkerError};
-use crate::types::{MessageReceiver, SubscribeReposReceiver};
+use crate::types::SubscribeReposReceiver;
+use crate::{SHUTDOWN, ValidatorManager};
 
 const CAPACITY: usize = 1024;
+const SLEEP: Duration = Duration::from_millis(10);
 
 #[derive(Debug, Error)]
 pub enum ManagerError {
@@ -36,22 +36,20 @@ struct WorkerHandle {
 pub struct Manager {
     workers: Box<[WorkerHandle]>,
     next_id: WorkerId,
-    message_rx: MessageReceiver,
-    bus: Bus<Arc<Vec<u8>>>,
     status_rx: StatusReceiver,
     subscribe_repos_rx: SubscribeReposReceiver,
 }
 
 impl Manager {
     pub fn new(
-        n_workers: usize, message_rx: MessageReceiver, subscribe_repos_rx: SubscribeReposReceiver,
+        n_workers: usize, validator: &mut ValidatorManager,
+        subscribe_repos_rx: SubscribeReposReceiver,
     ) -> Result<Self, ManagerError> {
-        let mut bus = Bus::new(CAPACITY);
         let (status_tx, status_rx) =
             magnetic::mpsc::mpsc_queue(DynamicBufferP2::new(CAPACITY).unwrap());
         let workers = (0..n_workers)
             .map(|worker_id| {
-                let message_rx = bus.add_rx();
+                let message_rx = validator.subscribe();
                 let status_tx = status_tx.clone();
                 let (command_tx, command_rx) = rtrb::RingBuffer::new(CAPACITY);
                 let thread_handle = thread::spawn(move || {
@@ -63,15 +61,15 @@ impl Manager {
         Ok(Self {
             workers: workers.into_boxed_slice(),
             next_id: WorkerId(0),
-            message_rx,
-            bus,
             status_rx,
             subscribe_repos_rx,
         })
     }
 
     pub fn run(mut self) -> Result<(), ManagerError> {
-        while self.update()? {}
+        while self.update()? {
+            thread::sleep(SLEEP);
+        }
         self.shutdown()
     }
 
@@ -103,17 +101,6 @@ impl Manager {
             }
         }
 
-        for _ in 0..32 {
-            match self.message_rx.try_recv_ref() {
-                Ok(msg) => {
-                    self.bus.try_broadcast(Arc::new(msg.data.clone().into())).unwrap();
-                }
-                Err(thingbuf::mpsc::errors::TryRecvError::Empty) => break,
-                Err(thingbuf::mpsc::errors::TryRecvError::Closed) => return Ok(false),
-                Err(_) => unreachable!(),
-            }
-        }
-
         if let Ok(subscribe_repos) = self.subscribe_repos_rx.pop() {
             let config = Config {
                 stream: subscribe_repos.stream,

rsky-relay/src/publisher/worker.rs

@@ -53,6 +53,7 @@ impl Worker {
         while self.update() {
             thread::yield_now();
         }
+        tracing::info!("shutting down publisher: {}", self.worker_id.0);
         self.shutdown()
     }
 
@@ -84,7 +85,6 @@ impl Worker {
                 }
             }
             Command::Shutdown => {
-                tracing::debug!("shutting down publisher: {}", self.worker_id.0);
                 return false;
             }
         }

rsky-relay/src/types.rs

@@ -1,10 +1,15 @@
 use std::net::TcpStream;
+use std::sync::LazyLock;
 
 use http::Uri;
 use rtrb::{Consumer, Producer};
+use sled::Db;
 use thingbuf::{Recycle, mpsc};
 use tungstenite::stream::MaybeTlsStream;
 
+pub static DB: LazyLock<Db> =
+    LazyLock::new(|| sled::Config::new().path("db").use_compression(true).open().unwrap());
+
 pub type RequestCrawlSender = Producer<RequestCrawl>;
 pub type RequestCrawlReceiver = Consumer<RequestCrawl>;
 pub type SubscribeReposSender = Producer<SubscribeRepos>;