From 6c85a341b86b79106fe1f41c94b8fa3dc16abc12 Mon Sep 17 00:00:00 2001 From: Eric Wendland Date: Sun, 17 May 2026 18:23:51 +0200 Subject: [PATCH] Add local in-memory pubsub --- AGENTS.md | 5 ++- Cargo.lock | 3 ++ README.md | 6 ++- crates/geth-cli/src/lib.rs | 22 +++++++++-- crates/geth-control/Cargo.toml | 1 + crates/geth-control/src/lib.rs | 16 ++++++++ crates/geth-node/Cargo.toml | 1 + crates/geth-node/src/lib.rs | 61 +++++++++++++++++++++++++++++ crates/geth-pubsub/Cargo.toml | 1 + crates/geth-pubsub/src/lib.rs | 49 +++++++++++++++++++++++- crates/geth/tests/bootstrap.rs | 70 ++++++++++++++++++++++++++++++++++ docs/architecture.md | 10 ++++- docs/roadmap.md | 13 +++++-- 13 files changed, 245 insertions(+), 13 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 0362252..cfc9621 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -125,11 +125,14 @@ Roadmap items should be actionable and checkable: replication and prefix-capability enforcement are still roadmap work. - Document resources can be registered locally with empty JSON state and local-only status. Automerge editing/state and sync are still roadmap work. +- Pubsub supports local daemon-lifetime publish/subscribe snapshots through a + bounded in-memory ring buffer. Iroh-gossip replication, private topics, and + pubsub capability enforcement are still roadmap work. - Resource secret epoch metadata can be created, rotated, and listed locally. Bearer access metadata can be created/listed/revoked as resource-scoped auth ops and must not allow trust graph mutation capabilities. Payload encryption, key envelopes, and bearer challenge-response are still roadmap work. - Signed peer-card LAN discovery payloads, peer auth over Iroh, cr-sqlite, - iroh-docs, iroh-gossip, iroh-blobs, Automerge sync, real auth enforcement, + iroh-docs, iroh-blobs, Automerge sync, real auth enforcement, OpenSSH KRL generation, and Keyhive/BeeKEM-style authorization are future roadmap items unless implemented later. diff --git a/Cargo.lock b/Cargo.lock index b74b42f..ad899a7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1120,6 +1120,7 @@ dependencies = [ "geth-document", "geth-keychain", "geth-kv", + "geth-pubsub", "geth-resource", "geth-secrets", "geth-ssh-identity", @@ -1221,6 +1222,7 @@ dependencies = [ "geth-iroh", "geth-keychain", "geth-kv", + "geth-pubsub", "geth-resource", "geth-secrets", "geth-ssh-identity", @@ -1246,6 +1248,7 @@ version = "0.1.0" dependencies = [ "geth-types", "serde", + "thiserror 2.0.18", ] [[package]] diff --git a/README.md b/README.md index c04438d..eb90ea9 100644 --- a/README.md +++ b/README.md @@ -91,6 +91,7 @@ The bootstrap implementation provides: `geth db status ` - local SQLite-backed KV commands: `geth kv create/set/get` - local document resource registration: `geth document create/status` +- local daemon-lifetime pubsub snapshots: `geth pubsub pub/sub` - SSH certificate flow metadata: - `geth ssh cert request --public-key --principal ` - `geth ssh cert requests` @@ -101,7 +102,7 @@ The bootstrap implementation provides: - `geth ssh revocation list` - `geth ssh revocation export --out ` -Other command groups exist as explicit stubs: `pipe`, `pubsub`, and `ssh`. +Other command groups exist as explicit stubs: `pipe` and `ssh`. ## Resource Modules @@ -111,7 +112,8 @@ Everything meaningful is modeled as a resource. Planned resource kinds are: - `kv`: Iroh Documents backed key-value stores - `pipe`: dumbpipe-like byte streams over Iroh - `document`: Automerge documents over Iroh streams -- `pubsub`: lossy notifications, not authoritative storage +- `pubsub`: lossy notifications, not authoritative storage; the bootstrap + keeps only an in-memory daemon-lifetime ring buffer - `cas`: content-addressed blob storage and distribution - `ssh-proxy`: authorized SSH proxy/admin access over Iroh diff --git a/crates/geth-cli/src/lib.rs b/crates/geth-cli/src/lib.rs index 1f5b981..42b2099 100644 --- a/crates/geth-cli/src/lib.rs +++ b/crates/geth-cli/src/lib.rs @@ -454,9 +454,9 @@ fn request_for_command(command: Command) -> Result { KvCommand::Set { name, key, value } => ControlRequest::KvSet { name, key, value }, KvCommand::Get { name, key } => ControlRequest::KvGet { name, key }, }, - Command::Pubsub { command } => ControlRequest::ModuleStub { - module: "pubsub".to_owned(), - command: format!("{command:?}"), + Command::Pubsub { command } => match command { + PubsubCommand::Pub { topic, message } => ControlRequest::PubsubPub { topic, message }, + PubsubCommand::Sub { topic } => ControlRequest::PubsubSub { topic }, }, Command::Pipe { command } => ControlRequest::ModuleStub { module: "pipe".to_owned(), @@ -893,6 +893,22 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> { println!("sync_status: {}", document.sync_status); println!("state_bytes: {}", document.state_bytes); } + ControlResponse::PubsubPublished { message } => { + println!("published: {}", message.topic); + println!("published_at_ms: {}", message.published_at.0); + } + ControlResponse::PubsubMessages { + topic, + messages, + note, + } => { + println!("topic: {topic}"); + println!("messages: {}", messages.len()); + for message in messages { + println!("{}\t{}", message.published_at.0, message.message); + } + println!("note: {note}"); + } ControlResponse::NotImplemented { module, command } => { println!("{module} {command}: not implemented yet"); } diff --git a/crates/geth-control/Cargo.toml b/crates/geth-control/Cargo.toml index 93458ec..45d163a 100644 --- a/crates/geth-control/Cargo.toml +++ b/crates/geth-control/Cargo.toml @@ -14,6 +14,7 @@ geth-db = { path = "../geth-db" } geth-document = { path = "../geth-document" } geth-keychain = { path = "../geth-keychain" } geth-kv = { path = "../geth-kv" } +geth-pubsub = { path = "../geth-pubsub" } geth-resource = { path = "../geth-resource" } geth-secrets = { path = "../geth-secrets" } geth-ssh-identity = { path = "../geth-ssh-identity" } diff --git a/crates/geth-control/src/lib.rs b/crates/geth-control/src/lib.rs index 7bb12a4..b2050c7 100644 --- a/crates/geth-control/src/lib.rs +++ b/crates/geth-control/src/lib.rs @@ -3,6 +3,7 @@ use geth_db::DbResource; use geth_document::DocumentResource; use geth_keychain::KeychainOp; use geth_kv::{KvEntry, KvResource}; +use geth_pubsub::PubsubMessage; use geth_resource::ResourceDescriptor; use geth_secrets::{BearerAccess, ResourceMasterSecret}; use geth_ssh_identity::{ @@ -136,6 +137,13 @@ pub enum ControlRequest { DocumentStatus { name: String, }, + PubsubPub { + topic: String, + message: String, + }, + PubsubSub { + topic: String, + }, ModuleStub { module: String, command: String, @@ -252,6 +260,14 @@ pub enum ControlResponse { DocumentStatus { document: DocumentResource, }, + PubsubPublished { + message: PubsubMessage, + }, + PubsubMessages { + topic: String, + messages: Vec, + note: String, + }, NotImplemented { module: String, command: String, diff --git a/crates/geth-node/Cargo.toml b/crates/geth-node/Cargo.toml index c9ccca0..c18aa53 100644 --- a/crates/geth-node/Cargo.toml +++ b/crates/geth-node/Cargo.toml @@ -20,6 +20,7 @@ geth-document = { path = "../geth-document" } geth-iroh = { path = "../geth-iroh" } geth-keychain = { path = "../geth-keychain" } geth-kv = { path = "../geth-kv" } +geth-pubsub = { path = "../geth-pubsub" } geth-resource = { path = "../geth-resource" } geth-secrets = { path = "../geth-secrets" } geth-ssh-identity = { path = "../geth-ssh-identity" } diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index aa1a127..a869900 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -13,6 +13,7 @@ use geth_document::DocumentResource; use geth_iroh::{EndpointStatus, GethIrohConfig, GethIrohEndpoint, GethRelayMode}; use geth_keychain::{KeychainOp, KeychainOpKind}; use geth_kv::{KvEntry, KvResource}; +use geth_pubsub::PubsubMessage; use geth_resource::ResourceDescriptor; use geth_secrets::{BearerAccess, ResourceMasterSecret}; use geth_ssh_identity::{ @@ -29,7 +30,9 @@ use geth_types::{ AuthOpId, Capability, KeyId, NodeId, PrincipalId, ResourceId, ResourceKind, ResourceName, SshCertId, SshCertRequestId, UnixMillis, }; +use std::collections::VecDeque; use std::path::Path; +use std::sync::{Arc, Mutex}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; @@ -73,6 +76,10 @@ pub enum NodeError { ResourceNotFound(String), #[error("secrets error: {0}")] Secrets(#[from] geth_secrets::SecretsError), + #[error("pubsub error: {0}")] + Pubsub(#[from] geth_pubsub::PubsubError), + #[error("runtime state lock poisoned")] + RuntimeLockPoisoned, #[error("invalid ssh certificate kind: {0}")] InvalidSshCertKind(String), #[error("invalid ssh certificate request status: {0}")] @@ -93,8 +100,21 @@ pub struct LocalNode { pub agent_id: String, pub node_id: String, pub iroh_status: EndpointStatus, + runtime: Arc, } +#[derive(Debug)] +struct NodeRuntime { + pubsub: Mutex, +} + +#[derive(Debug, Default)] +struct PubsubRuntime { + messages: VecDeque, +} + +const PUBSUB_RING_LIMIT: usize = 256; + pub fn init_node(paths: &GethPaths) -> Result { paths.ensure_base_dirs()?; if !paths.config_file().exists() { @@ -117,6 +137,9 @@ pub fn init_node(paths: &GethPaths) -> Result { agent_id, node_id, iroh_status: EndpointStatus::scaffolded(), + runtime: Arc::new(NodeRuntime { + pubsub: Mutex::new(PubsubRuntime::default()), + }), }) } @@ -784,6 +807,44 @@ pub fn handle_request( document: document_resource_from_stored(&stored), }) } + ControlRequest::PubsubPub { topic, message } => { + geth_pubsub::validate_topic(&topic)?; + geth_pubsub::validate_message(&message)?; + let message = PubsubMessage { + topic: topic.clone().into(), + message, + published_at: UnixMillis(geth_store::now_ms()), + }; + let mut runtime = node + .runtime + .pubsub + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)?; + runtime.messages.push_back(message.clone()); + while runtime.messages.len() > PUBSUB_RING_LIMIT { + runtime.messages.pop_front(); + } + Ok(ControlResponse::PubsubPublished { message }) + } + ControlRequest::PubsubSub { topic } => { + geth_pubsub::validate_topic(&topic)?; + let runtime = node + .runtime + .pubsub + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)?; + let messages = runtime + .messages + .iter() + .filter(|message| message.topic.as_str() == topic) + .cloned() + .collect(); + Ok(ControlResponse::PubsubMessages { + topic, + messages, + note: geth_pubsub::pubsub_storage_warning().to_owned(), + }) + } ControlRequest::ModuleStub { module, command } => { Ok(ControlResponse::NotImplemented { module, command }) } diff --git a/crates/geth-pubsub/Cargo.toml b/crates/geth-pubsub/Cargo.toml index 0f110cf..9481607 100644 --- a/crates/geth-pubsub/Cargo.toml +++ b/crates/geth-pubsub/Cargo.toml @@ -7,4 +7,5 @@ license.workspace = true [dependencies] serde.workspace = true +thiserror.workspace = true geth-types = { path = "../geth-types" } diff --git a/crates/geth-pubsub/src/lib.rs b/crates/geth-pubsub/src/lib.rs index fe14b55..3eb68d3 100644 --- a/crates/geth-pubsub/src/lib.rs +++ b/crates/geth-pubsub/src/lib.rs @@ -1,4 +1,4 @@ -use geth_types::{ResourceId, TopicId}; +use geth_types::{ResourceId, TopicId, UnixMillis}; use serde::{Deserialize, Serialize}; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -8,7 +8,54 @@ pub struct PubsubTopic { pub name: String, } +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PubsubMessage { + pub topic: TopicId, + pub message: String, + pub published_at: UnixMillis, +} + +#[derive(Debug, thiserror::Error)] +pub enum PubsubError { + #[error("invalid pubsub topic: {0}")] + InvalidTopic(String), + #[error("pubsub message must not be empty")] + EmptyMessage, +} + +pub fn validate_topic(topic: &str) -> Result<(), PubsubError> { + if topic.is_empty() + || topic + .bytes() + .any(|byte| byte == 0 || byte == b'\n' || byte == b'\r') + { + return Err(PubsubError::InvalidTopic(topic.to_owned())); + } + Ok(()) +} + +pub fn validate_message(message: &str) -> Result<(), PubsubError> { + if message.is_empty() { + return Err(PubsubError::EmptyMessage); + } + Ok(()) +} + #[must_use] pub fn pubsub_storage_warning() -> &'static str { "pubsub is lossy notification transport, not authoritative storage" } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pubsub_validation_rejects_invalid_topics_and_empty_messages() { + assert!(validate_topic("presence/laptop").is_ok()); + assert!(validate_topic("").is_err()); + assert!(validate_topic("presence\nlaptop").is_err()); + assert!(validate_message("online").is_ok()); + assert!(validate_message("").is_err()); + } +} diff --git a/crates/geth/tests/bootstrap.rs b/crates/geth/tests/bootstrap.rs index 6b74f56..c034b86 100644 --- a/crates/geth/tests/bootstrap.rs +++ b/crates/geth/tests/bootstrap.rs @@ -734,6 +734,76 @@ fn bearer_access_create_list_revoke_uses_resource_scoped_auth_ops() { ); } +#[test] +fn pubsub_pub_sub_uses_lossy_in_memory_runtime() { + let home = tempfile::tempdir().expect("tempdir"); + let paths = geth_config::GethPaths::from_home(home.path()); + let node = geth_node::init_node(&paths).expect("init node"); + + let response = geth_node::handle_request( + &node, + geth_control::ControlRequest::PubsubPub { + topic: "presence/laptop".to_owned(), + message: "online".to_owned(), + }, + ) + .expect("publish"); + match response { + geth_control::ControlResponse::PubsubPublished { message } => { + assert_eq!(message.topic.to_string(), "presence/laptop"); + assert_eq!(message.message, "online"); + } + other => panic!("unexpected response: {other:?}"), + } + + let response = geth_node::handle_request( + &node, + geth_control::ControlRequest::PubsubSub { + topic: "presence/laptop".to_owned(), + }, + ) + .expect("subscribe snapshot"); + match response { + geth_control::ControlResponse::PubsubMessages { + topic, + messages, + note, + } => { + assert_eq!(topic, "presence/laptop"); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].message, "online"); + assert!(note.contains("not authoritative storage")); + } + other => panic!("unexpected response: {other:?}"), + } + + let reopened = geth_node::open_node(&paths).expect("reopen node"); + let response = geth_node::handle_request( + &reopened, + geth_control::ControlRequest::PubsubSub { + topic: "presence/laptop".to_owned(), + }, + ) + .expect("subscribe reopened snapshot"); + match response { + geth_control::ControlResponse::PubsubMessages { messages, .. } => { + assert!(messages.is_empty()); + } + other => panic!("unexpected response: {other:?}"), + } + + assert!( + geth_node::handle_request( + &node, + geth_control::ControlRequest::PubsubPub { + topic: "presence/laptop".to_owned(), + message: String::new(), + }, + ) + .is_err() + ); +} + #[test] fn ssh_cert_request_approval_and_revocation_export_use_local_state() { let home = tempfile::tempdir().expect("tempdir"); diff --git a/docs/architecture.md b/docs/architecture.md index 97fe3a8..67a0c00 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -105,8 +105,14 @@ enforcement, and replication are future work. state placeholder and local-only sync status. Automerge state, editing, and sync are future work. -`geth-pubsub`, `geth-pipe`, and `geth-ssh-proxy` currently define types, -command shape, and roadmap stubs. +`geth-pubsub` currently supports local publish/subscribe snapshots through the +daemon control protocol. Messages live in a bounded in-memory ring buffer and +are lost when the daemon stops. This is deliberate: pubsub is a lossy wakeup and +presence channel, not authoritative storage. Iroh-gossip replication, private +topics, and capability enforcement are future work. + +`geth-pipe` and `geth-ssh-proxy` currently define types, command shape, and +roadmap stubs. `geth-ssh-identity` defines SSH trust namespaces plus certificate request, approval, certificate import, and revocation-list data models. The bootstrap diff --git a/docs/roadmap.md b/docs/roadmap.md index ae69de6..ed3d640 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -228,11 +228,16 @@ authorization and durable-state boundaries clear. - `[ ]` Prefix-scoped capabilities can allow or deny writes. - `[ ]` KV metadata is replicated through Iroh Documents. -- `[ ]` Iroh-gossip pubsub integration. +- `[~]` Iroh-gossip pubsub integration. Acceptance criteria: - - `geth pubsub pub/sub` works for local test nodes. - - Pubsub messages are lossy notifications, not durable facts. - - Docs and tests keep durable state in CAS/KV/document/db instead. + - `[x]` `geth pubsub pub/sub` works against the local daemon. + - `[x]` Local pubsub messages are kept in a bounded daemon-lifetime ring + buffer. + - `[x]` Pubsub messages are documented and tested as lossy notifications, not + durable facts. + - `[ ]` `geth pubsub pub/sub` works across local test nodes over Iroh. + - `[ ]` Pubsub publish/subscribe access is capability checked. + - `[ ]` Docs and tests keep durable state in CAS/KV/document/db instead. ## Phase 4: Pipes, SSH Proxy, And SSH Distribution