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

@@ -530,6 +530,54 @@ dependencies = [
  "cfg-if",
 ]
 
+[[package]]
+name = "crossbeam"
+version = "0.8.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2801af0d36612ae591caa9568261fddce32ce6e08a7275ea334a06a4ad021a2c"
+dependencies = [
+ "cfg-if",
+ "crossbeam-channel",
+ "crossbeam-deque",
+ "crossbeam-epoch",
+ "crossbeam-queue",
+ "crossbeam-utils",
+]
+
+[[package]]
+name = "crossbeam-channel"
+version = "0.5.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a33c2bf77f2df06183c3aa30d1e96c0695a313d4f9c453cc3762a6db39f99200"
+dependencies = [
+ "cfg-if",
+ "crossbeam-utils",
+]
+
+[[package]]
+name = "crossbeam-deque"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ce6fd6f855243022dcecf8702fef0c297d4338e226845fe067f6341ad9fa0cef"
+dependencies = [
+ "cfg-if",
+ "crossbeam-epoch",
+ "crossbeam-utils",
+]
+
+[[package]]
+name = "crossbeam-epoch"
+version = "0.9.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ae211234986c545741a7dc064309f67ee1e5ad243d0e48335adc0484d960bcc7"
+dependencies = [
+ "autocfg",
+ "cfg-if",
+ "crossbeam-utils",
+ "memoffset",
+ "scopeguard",
+]
+
 [[package]]
 name = "crossbeam-queue"
 version = "0.3.8"
@@ -1399,6 +1447,15 @@ version = "2.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2dffe52ecf27772e601905b7522cb4ef790d2cc203488bbd0e2fe85fcb74566d"
 
+[[package]]
+name = "memoffset"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a634b1c61a95585bd15607c6ab0c4e5b226e695ff2800ba0cdccddf208c406c"
+dependencies = [
+ "autocfg",
+]
+
 [[package]]
 name = "mime"
 version = "0.3.17"
@@ -1857,6 +1914,7 @@ dependencies = [
  "axum",
  "clap 4.3.21",
  "comrak",
+ "crossbeam",
  "file-format",
  "frontmatter",
  "futures",
@@ -1865,6 +1923,7 @@ dependencies = [
  "lazy_static 1.4.0",
  "maud",
  "orgize",
+ "rand",
  "rss",
  "serde",
  "serde_dhall",

Cargo.toml

@@ -27,3 +27,5 @@ file-format = "0.18.0"
 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"

flake.lock

@@ -3,11 +3,11 @@
     "advisory-db": {
       "flake": false,
       "locked": {
-        "lastModified": 1689698236,
+        "lastModified": 1696104803,
-        "narHash": "sha256-Qz9JxGKeA3jwuj1CdK9ejMJ7VsJRdiZniF8lx4mft9s=",
+        "narHash": "sha256-xdPl4PHzWaisHy4hQexpHJlx/3xAkCZrOFtEv3ygXco=",
         "owner": "rustsec",
         "repo": "advisory-db",
-        "rev": "4aa517564d1d06f0e79784c8ad973a59d68aa9c8",
+        "rev": "46754ce9372c13a7e9a05cf7d812fadfdca9cf57",
         "type": "github"
       },
       "original": {
@@ -26,11 +26,11 @@
         "rust-overlay": "rust-overlay"
       },
       "locked": {
-        "lastModified": 1688772518,
+        "lastModified": 1696124486,
-        "narHash": "sha256-ol7gZxwvgLnxNSZwFTDJJ49xVY5teaSvF7lzlo3YQfM=",
+        "narHash": "sha256-a/xfX9CZnNQoW38w7+fpxBJ8aFQVQQg+Va3OcHBBCZs=",
         "owner": "ipetkov",
         "repo": "crane",
-        "rev": "8b08e96c9af8c6e3a2b69af5a7fa168750fcf88e",
+        "rev": "af842d7319f8c3d9e5215a79306824897fcb667c",
         "type": "github"
       },
       "original": {
@@ -47,11 +47,11 @@
         "rust-analyzer-src": []
       },
       "locked": {
-        "lastModified": 1689834114,
+        "lastModified": 1696141234,
-        "narHash": "sha256-btRpL43gvbP+5+gHjaaeZ9Uv6x8LkjaO1kyvEN5rQTE=",
+        "narHash": "sha256-0dZpggYjjmWEk+rGixiBHOHuQfLzEzNfrtjSig04s6Q=",
         "owner": "nix-community",
         "repo": "fenix",
-        "rev": "d55d856bcc50ae831f9355465f6a651034f6c8d4",
+        "rev": "9ccae1754eec0341b640d5705302ac0923d22875",
         "type": "github"
       },
       "original": {
@@ -81,11 +81,11 @@
         "systems": "systems"
       },
       "locked": {
-        "lastModified": 1687709756,
+        "lastModified": 1694529238,
-        "narHash": "sha256-Y5wKlQSkgEK2weWdOu4J3riRd+kV/VCgHsqLNTTWQ/0=",
+        "narHash": "sha256-zsNZZGTGnMOf9YpHKJqMSsa0dXbfmxeoJ7xHlrt+xmY=",
         "owner": "numtide",
         "repo": "flake-utils",
-        "rev": "dbabf0ca0c0c4bce6ea5eaf65af5cb694d2082c7",
+        "rev": "ff7b65b44d01cf9ba6a71320833626af21126384",
         "type": "github"
       },
       "original": {
@@ -99,11 +99,11 @@
         "systems": "systems_2"
       },
       "locked": {
-        "lastModified": 1689068808,
+        "lastModified": 1694529238,
-        "narHash": "sha256-6ixXo3wt24N/melDWjq70UuHQLxGV8jZvooRanIHXw0=",
+        "narHash": "sha256-zsNZZGTGnMOf9YpHKJqMSsa0dXbfmxeoJ7xHlrt+xmY=",
         "owner": "numtide",
         "repo": "flake-utils",
-        "rev": "919d646de7be200f3bf08cb76ae1f09402b6f9b4",
+        "rev": "ff7b65b44d01cf9ba6a71320833626af21126384",
         "type": "github"
       },
       "original": {
@@ -114,11 +114,11 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1689752456,
+        "lastModified": 1696009558,
-        "narHash": "sha256-VOChdECcEI8ixz8QY+YC4JaNEFwQd1V8bA0G4B28Ki0=",
+        "narHash": "sha256-/1nNL8lCF0gn38XaFyu2ufpWcBFwCDZyYUxdZkM6GxU=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "7f256d7da238cb627ef189d56ed590739f42f13b",
+        "rev": "c182df2e68bd97deb32c7e4765adfbbbcaf75b60",
         "type": "github"
       },
       "original": {
@@ -149,11 +149,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1688351637,
+        "lastModified": 1695003086,
-        "narHash": "sha256-CLTufJ29VxNOIZ8UTg0lepsn3X03AmopmaLTTeHDCL4=",
+        "narHash": "sha256-d1/ZKuBRpxifmUf7FaedCqhy0lyVbqj44Oc2s+P5bdA=",
         "owner": "oxalica",
         "repo": "rust-overlay",
-        "rev": "f9b92316727af9e6c7fee4a761242f7f46880329",
+        "rev": "b87a14abea512d956f0b89d0d8a1e9b41f3e20ff",
         "type": "github"
       },
       "original": {

src/cache/memory.rs

@@ -3,7 +3,7 @@ use std::collections::HashMap;
 use async_trait::async_trait;
 use tokio::sync::Mutex;
 
-use super::{CacheMechanism, CachedItem, should_use, Tier};
+use super::{should_use, CacheMechanism, CachedItem, Tier};
 
 #[derive(Clone, Debug)]
 pub struct Memory {}
@@ -13,7 +13,7 @@ lazy_static! {
 }
 
 pub fn new() -> Memory {
-    return Memory{};
+    return Memory {};
 }
 
 #[async_trait]
@@ -26,9 +26,11 @@ impl CacheMechanism for Memory {
                     let mut r = c.clone();
                     r.tier = Some(Tier::Memory);
                     Some(r)
-                } else { None }
+                } else {
-            },
+                    None
-            None => None
+                }
+            }
+            None => None,
         }
     }
 

src/cache/mod.rs

@@ -1,9 +1,12 @@
 mod memory;
 mod sqlite;
 
-use std::{time::{SystemTime, UNIX_EPOCH}, fmt};
 use async_trait::async_trait;
 use sqlx::FromRow;
+use std::{
+    fmt,
+    time::{SystemTime, UNIX_EPOCH},
+};
 
 use self::{memory::Memory, sqlite::Sqlite};
 
@@ -14,7 +17,10 @@ pub struct Cache {
 }
 
 pub async fn init_cache() -> Cache {
-    Cache{ memory: memory::new(), sqlite: sqlite::new().await }
+    Cache {
+        memory: memory::new(),
+        sqlite: sqlite::new().await,
+    }
 }
 
 /// Tier enums take an i64, which is the amount of time in seconds
@@ -24,7 +30,7 @@ pub enum Tier {
     Memory,
     Sqlite,
     External,
-    None
+    None,
 }
 
 impl fmt::Display for Tier {
@@ -44,7 +50,7 @@ pub struct CachedItem {
     pub content: String,
     cached: i64,
     #[sqlx(default)]
-    tier: Option<Tier>
+    tier: Option<Tier>,
 }
 
 impl CachedItem {
@@ -68,7 +74,8 @@ impl Cache {
         }
         if self.sqlite.is_some() {
             let sq = self.sqlite.clone().unwrap();
-            let s = sq.get(key).await; if s.is_some() {
+            let s = sq.get(key).await;
+            if s.is_some() {
                 let current_time = SystemTime::now()
                     .duration_since(UNIX_EPOCH)
                     .expect("SystemTime before UNIX EPOCH!")
@@ -78,11 +85,11 @@ impl Cache {
                 let mut refresh_memory = s.clone().unwrap();
                 refresh_memory.cached = current_time;
                 let _ = self.memory.set(key.clone(), refresh_memory).await;
-                return s
+                return s;
             }
         }
 
-        return None
+        return None;
     }
 
     pub async fn set(&self, key: String, content_type: String, content: String) -> bool {
@@ -92,7 +99,12 @@ impl Cache {
             .as_secs()
             .try_into()
             .unwrap();
-         let cached_item = CachedItem{ content_type, content, cached: current_time, tier: None };
+        let cached_item = CachedItem {
+            content_type,
+            content,
+            cached: current_time,
+            tier: None,
+        };
         self.memory.set(key.clone(), cached_item.clone()).await;
         if self.sqlite.is_some() {
             let sq = self.sqlite.clone().unwrap();
@@ -106,8 +118,8 @@ impl Cache {
 fn should_use(item: &CachedItem, tier: Tier) -> bool {
     // TODO: Make configurable.
     let cache_time = match tier {
-        Tier::Memory => 2*60,
+        Tier::Memory => 2 * 60,
-        Tier::Sqlite => 10*60,
+        Tier::Sqlite => 10 * 60,
         Tier::External => 0,
         Tier::None => 0,
     };
@@ -115,7 +127,9 @@ fn should_use(item: &CachedItem, tier: Tier) -> bool {
     let current_time: i64 = SystemTime::now()
         .duration_since(UNIX_EPOCH)
         .expect("SystemTime before UNIX EPOCH!")
-        .as_secs().try_into().unwrap();
+        .as_secs()
+        .try_into()
+        .unwrap();
 
     current_time <= (item.cached + cache_time) && item.content != ""
 }

src/cache/sqlite.rs

@@ -1,23 +1,27 @@
 use std::{env, str::FromStr};
 
 use async_trait::async_trait;
-use sqlx::{Pool, sqlite::{SqlitePoolOptions, SqliteJournalMode, SqliteConnectOptions}};
+use sqlx::{
+    sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions},
+    Pool,
+};
 
-use super::{CacheMechanism, CachedItem, should_use, Tier};
+use super::{should_use, CacheMechanism, CachedItem, Tier};
 
 #[derive(Clone, Debug)]
 pub struct Sqlite {
-    pool: Pool<sqlx::Sqlite>
+    pool: Pool<sqlx::Sqlite>,
 }
 
 pub async fn new() -> Option<Sqlite> {
     let path = env::var("DATABASE_PATH").unwrap_or("gs.db".to_owned());
-    let opts = SqliteConnectOptions::from_str(&path).unwrap()
+    let opts = SqliteConnectOptions::from_str(&path)
+        .unwrap()
         .journal_mode(SqliteJournalMode::Wal)
         .create_if_missing(true);
 
     let pool = SqlitePoolOptions::new().connect_with(opts).await.unwrap();
-    return Some(Sqlite{ pool });
+    return Some(Sqlite { pool });
 }
 
 #[async_trait]
@@ -25,7 +29,8 @@ impl CacheMechanism for Sqlite {
     async fn get(&self, key: &String) -> Option<CachedItem> {
         let res = sqlx::query_as::<_, CachedItem>("SELECT * FROM cached WHERE route = $1")
             .bind(&key)
-            .fetch_one(&self.pool).await;
+            .fetch_one(&self.pool)
+            .await;
         if res.is_ok() {
             let c = res.unwrap();
             if should_use(&c, Tier::Sqlite) {

src/main.rs

@@ -1,10 +1,11 @@
 #[macro_use]
 extern crate lazy_static;
 
-mod posts;
 mod cache;
+mod posts;
 
 use axum::extract::Path;
+use axum::http::request;
 use axum::response::IntoResponse;
 use axum::{
     body::Full,
@@ -16,19 +17,60 @@ use axum::{
     Router,
 };
 use clap::Parser;
+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::env;
+use std::collections::HashMap;
+use std::fmt::Display;
+use std::net::UdpSocket;
 use std::str::FromStr;
+use std::{env, io, thread};
 use time::{self, format_description, format_description::well_known::Rfc2822};
 use tower_http::services::ServeDir;
 
-use crate::cache::{Cache, init_cache};
+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 {
@@ -36,11 +78,14 @@ struct Cli {
     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)]
@@ -89,7 +134,15 @@ async fn main() -> Result<(), sqlx::Error> {
     sqlx::migrate!("./migrations").run(&pool).await?;
 
     env::set_var("DATABASE_PATH", &args.database_path);
-    let state = AppState { cache: init_cache().await };
+
+    // 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()
         .route("/", get(homepage))
@@ -103,14 +156,102 @@ async fn main() -> Result<(), sqlx::Error> {
         .with_state(state);
 
     println!("Running webserver on {}", args.bind);
-    axum::Server::bind(&args.bind.parse().unwrap())
+    let webserver = axum::Server::bind(&args.bind.parse().unwrap())
-        .serve(app.into_make_service())
+        .serve(app.into_make_service());
-        .await
+
-        .unwrap();
+    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::scope(|scope| {
+            let _ = gossiper(s_main.clone(), r_main, s_gossip);
+        });
+
+        let _ = s_main.send(GossipMessages::Ready);
+        loop {
+            for msg in r_gossip.try_iter() {
+                println!("gossip thrd: {}", msg);
+            }
+        }
+    } else {
+        let _ = webserver.await;
+    }
     
     Ok(())
 }
 
+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 r_socket = socket.try_clone().unwrap();
+    let _ = tx.send(GossipMessages::Ready);
+
+    // 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 mut request = String::from_utf8_lossy(&buf[..size]).into_owned();
+        trim_newline(&mut request);
+
+        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> {
     let post = posts::blog_post(post);
     if post.is_err() {
@@ -279,12 +420,17 @@ 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")
             .status(StatusCode::OK)
             .body(Full::from(bytes))
-            .unwrap()
+            .unwrap();
     } else {
         let i = item.unwrap();
         return Response::builder()

src/posts.rs

@@ -1,6 +1,9 @@
-use std::{fs::{self, File}, io::Read};
+use std::{
+    fs::{self, File},
+    io::Read,
+};
 
-use maud::{Render, Markup, html};
+use maud::{html, Markup, Render};
 use orgize::Org;
 
 pub struct PostMetadata {
@@ -13,7 +16,7 @@ pub struct PostContent {
     pub title: String,
     pub date: String,
     pub content: String,
-    pub html: String
+    pub html: String,
 }
 
 impl Render for PostMetadata {
@@ -110,11 +113,11 @@ pub fn blog_post(post: String) -> Result<PostContent, bool> {
                 parsed.write_html(&mut writer).unwrap();
                 html = String::from_utf8(writer).unwrap();
             }
-            return Ok(PostContent{
+            return Ok(PostContent {
                 title,
                 date,
                 content,
-                html
+                html,
             });
         }
     }