More groundwork for gossip

Working UDP -> message handler and cache middleware -> message
handler. Now I just need to figure out the format of the messages
passed, and do testing with other peers.

Cargo.lock

@@ -1923,6 +1923,7 @@ dependencies = [
  "lazy_static 1.4.0",
  "maud",
  "orgize",
+ "rand",
  "rss",
  "serde",
  "serde_dhall",

Cargo.toml

@@ -28,3 +28,4 @@ rss = "2.0.6"
 time = { version = "0.3.28", features = ["parsing", "formatting", "macros"] }
 async-trait = "0.1.73"
 crossbeam = "0.8.2"
+rand = "0.8.5"

src/main.rs

@@ -5,6 +5,7 @@ mod cache;
 mod posts;
 
 use axum::extract::Path;
+use axum::http::request;
 use axum::response::IntoResponse;
 use axum::{
     body::Full,
@@ -20,9 +21,12 @@ use crossbeam::channel::{unbounded, Receiver, Sender};
 use file_format::{FileFormat, Kind};
 use hyper::body::Bytes;
 use maud::{html, Markup, PreEscaped, Render, DOCTYPE};
+use rand::seq::SliceRandom;
 use rss::ChannelBuilder;
 use serde::Deserialize;
 use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
+use std::collections::HashMap;
+use std::fmt::Display;
 use std::net::UdpSocket;
 use std::str::FromStr;
 use std::{env, io, thread};
@@ -31,17 +35,57 @@ use tower_http::services::ServeDir;
 
 use crate::cache::{init_cache, Cache};
 
+enum PeerStatus {
+    Healthy,
+    Suspect,
+    Down
+}
+
+enum GossipMessages {
+    Ready,
+    Cache,
+    Ack,
+}
+impl GossipMessages {
+    fn from_str(request: String) -> Option<Self> {
+        match request.as_str() {
+            "rdy" => Some(Self::Ready),
+            "cache" => Some(Self::Cache),
+            "ack" => Some(Self::Ack),
+            _ => None,
+        }
+    }
+}
+
+impl Display for GossipMessages {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            GossipMessages::Ready => write!(f, "rdy"),
+            GossipMessages::Cache => write!(f, "cache"),
+            GossipMessages::Ack => write!(f, "ack"),
+        }
+    }
+}
+
+struct Peer {
+    counter: i64,
+    health: PeerStatus
+}
+
 #[derive(Parser)]
 struct Cli {
     #[arg(short, long, default_value_t=("gs.db").to_string())]
     database_path: String,
     #[arg(short, long, default_value_t=("0.0.0.0:3000").to_string())]
     bind: String,
+    #[arg(short, long)]
+    peers: Option<String>
 }
 
 #[derive(Clone, Debug)]
 struct AppState {
     cache: Cache,
+    to_gossip: Sender<GossipMessages>,
 }
 
 #[derive(Deserialize)]
@@ -90,8 +134,14 @@ async fn main() -> Result<(), sqlx::Error> {
     sqlx::migrate!("./migrations").run(&pool).await?;
 
     env::set_var("DATABASE_PATH", &args.database_path);
+
+    // Create channels for sending messages and receiving results
+    let (s_gossip, r_gossip) = unbounded::<GossipMessages>();
+    let (s_main, r_main) = unbounded::<GossipMessages>();
+
     let state = AppState {
         cache: init_cache().await,
+        to_gossip: s_main.clone()
     };
 
     let app = Router::new()
@@ -105,51 +155,101 @@ async fn main() -> Result<(), sqlx::Error> {
         .layer(middleware::from_fn_with_state(state.clone(), cached_page))
         .with_state(state);
 
-    // Create channels for sending messages and receiving results
-    let (to_gossip, from_gossip) = unbounded::<String>();
-    let (to_main, from_main) = unbounded();
+    println!("Running webserver on {}", args.bind);
+    let webserver = axum::Server::bind(&args.bind.parse().unwrap())
+        .serve(app.into_make_service());
 
+    if args.peers.is_some_and(|f| f != "".to_owned()) {
+        tokio::spawn(webserver);
+        println!("starting gossip worker");
         // Spawn a worker thread
-    let gossip_server = thread::spawn(move || {
-        let _ = gossiper(from_main, to_gossip);
+        let gossip_server = thread::scope(|scope| {
+            let _ = gossiper(s_main.clone(), r_main, s_gossip);
         });
 
-    println!("Running webserver on {}", args.bind);
-    let webserver = axum::Server::bind(&args.bind.parse().unwrap()).serve(app.into_make_service());
-    tokio::spawn(webserver);
-
+        let _ = s_main.send(GossipMessages::Ready);
         loop {
-        for msg in from_gossip.try_iter() {
-            println!("{}", msg);
+            for msg in r_gossip.try_iter() {
+                println!("gossip thrd: {}", msg);
             }
         }
-
-    // Optionally, wait for the worker thread to finish
-    gossip_server.join().unwrap();
+    } else {
+        let _ = webserver.await;
+    }
     
     Ok(())
 }
 
-fn gossiper(rx: Receiver<String>, tx: Sender<String>) -> io::Result<()> {
+fn gossiper(own_sender: Sender<GossipMessages>, rx: Receiver<GossipMessages>, tx: Sender<GossipMessages>) -> io::Result<()> {
+    let mut peers: HashMap<String, Peer> = HashMap::new();
     let mut buf = [0; 1024];
     let socket = UdpSocket::bind("0.0.0.0:1337")?;
-    let _ = tx.send("rdy".to_owned());
+    let r_socket = socket.try_clone().unwrap();
+    let _ = tx.send(GossipMessages::Ready);
 
-    // Handle messages that are passed to us from the main thread.
-    for message in rx.try_iter() {
-        println!("Received: {}", message);
-    }
-    // Read from the UDP socket.
-    loop {
+    // Listen on our UDP socket and pass messages back up.
+    thread::spawn(move || loop {
         let (size, source) = socket.recv_from(&mut buf).expect("Failed to receive data");
-        let request = String::from_utf8_lossy(&buf[..size]);
-        let _ = tx.send(request.to_string()).unwrap();
+        let mut request = String::from_utf8_lossy(&buf[..size]).into_owned();
+        trim_newline(&mut request);
 
-        let response = "ack!";
+        let msg = GossipMessages::from_str(request).unwrap();
+        let _ = own_sender.send(msg).unwrap();
+        let response = "ack";
         socket
             .send_to(response.as_bytes(), source)
             .expect("Failed to send response");
+    });
+    
+    // Handle messages that are passed to us from the main thread or the UDP socket.
+    loop {
+        let message = rx.recv().unwrap();
+        println!("got: {}", &message);
+        match message {
+            GossipMessages::Cache => {
+                let selected_peers: Vec<String> = match select_peers(&peers) {
+                    Some(p) => { println!("found peers"); p },
+                    None => { println!("no peers, not gossiping"); vec![] },
+                };
+                for peer in selected_peers {
+                    let msg = GossipMessages::Cache.to_string().into_bytes();
+                    let result = match r_socket.send_to(&msg, &peer) {
+                        Ok(_) => "",
+                        Err(_) => "",
+                    };
+                    let p = peers.get_mut(&peer).unwrap();
+                    p.counter = p.counter + 1;
                 }
+            },
+            GossipMessages::Ack => {
+                println!("healthy ack");
+            }
+            _ => {}
+        }
+    }
+}
+
+fn trim_newline(s: &mut String) {
+    if s.ends_with('\n') {
+        s.pop();
+        if s.ends_with('\r') {
+            s.pop();
+        }
+    }
+}
+
+/// select a third of the peers we know about to gossip to.
+fn select_peers(peers: &HashMap<String, Peer>) -> Option<Vec<String>> {
+    if peers.len() == 0 {
+        return None;
+    }
+    let healthy: Vec<String> = peers.into_iter()
+        .filter(|(&ref k, &ref peer)| matches!(peer.health, PeerStatus::Healthy))
+        .map(|(&ref ip, &ref peer)| ip.to_owned()).collect();
+    let len = healthy.len() as usize / (3 as usize);
+    let rng = &mut rand::thread_rng();
+    // Select a random number of peers to gossip to.
+    Some(healthy.choose_multiple(rng, len).map(|f| f.to_owned()).collect())
 }
 
 async fn raw_blog_post(Path(post): Path<String>) -> Result<impl IntoResponse, StatusCode> {
@@ -320,6 +420,11 @@ async fn cached_page<T>(
 
         let content = String::from_utf8(res).unwrap();
         state.cache.set(path, contenttype.to_owned(), content).await;
+        match state.to_gossip.send(GossipMessages::Cache) {
+            Ok(_) => { },
+            Err(_) => { },
+        };
+
         return Response::builder()
             .header("content-type", contenttype)
             .header("cache", "miss")