Wire pubsub through iroh-gossip

Cargo.lock

@@ -1599,6 +1599,8 @@ name = "geth-node"
 version = "0.1.0"
 dependencies = [
  "base64",
+ "blake3",
+ "bytes",
  "futures",
  "geth-auth",
  "geth-cas",

README.md

@@ -160,9 +160,9 @@ The bootstrap implementation provides:
   CLI input and output are JSON views, while the store keeps durable Automerge
   save bytes. `geth document sync <node-id> <name> [--bearer-secret <secret>]`
   pulls authorized remote Automerge state.
-- local daemon-lifetime pubsub snapshots: `geth pubsub pub/sub`; `geth pubsub
+- lossy pubsub wakeups: `geth pubsub pub/sub`; local messages are retained in
-  pub <topic> <message> --node <node-id>` publishes to an authorized peer;
+  a daemon-lifetime ring buffer, while authorized remote publish/subscribe joins
-  `geth pubsub sub <topic> --node <node-id>` reads an authorized peer snapshot
+  deterministic native `iroh-gossip` topics after geth control authorization
 - SSH certificate flow metadata:
   - `geth ssh cert request --public-key <path> --principal <name> [--subject <principal>]`
   - `geth ssh cert requests [--subject <principal>]`
@@ -204,8 +204,9 @@ serving peer as a provider visible with `geth cas providers <hash>`.
 libraries `iroh-blobs 0.97.0`, `iroh-docs 0.95.0`, and `iroh-gossip 0.95.0`
 against the same daemon-owned endpoint generation. KV stores are mirrored into
 native `iroh-docs` namespaces and peers receive read-only document tickets only
-after geth authorization succeeds. Pubsub still uses its documented bootstrap
+after geth authorization succeeds. Pubsub joins native `iroh-gossip` topics
-equivalent until the native gossip migration lands.
+only after the geth control path has authenticated the peer-card endpoint
+binding and checked the topic capability.
 Remote resource commands that accept `--bearer-secret` can also authorize with a
 resource-scoped bearer proof generated from the private bearer token returned at
 creation time. The persisted auth log stores a public bearer id and token
@@ -245,14 +246,15 @@ manual `geth kv sync <node-id> <name>` and background ticks require `kv.read`
 on the remote `resource:kv:<name>`. Authorized sync imports from the remote
 Iroh Documents namespace where available, keeps SQLite as the durable local
 index, and imports only remote entries that are not older than the local value.
-Remote pubsub publish uses the protected Iroh control path too. The remote peer
+Remote pubsub publish uses the protected Iroh control path as an authorization
-requires `pubsub.publish` on `resource:pubsub:<topic>` before recording the
+preflight. The remote peer requires `pubsub.publish` on
-message in its local daemon-lifetime ring buffer. Pubsub remains lossy and is
+`resource:pubsub:<topic>` before recording the message and broadcasting it on a
-not durable storage; facts that must survive restart or reconcile offline
+deterministic native `iroh-gossip` topic. Pubsub remains lossy and is not
-belong in CAS, KV, document, or DB resources. Remote pubsub subscribe uses the
+durable storage; facts that must survive restart or reconcile offline belong in
-same protected path and requires `pubsub.subscribe` on
+CAS, KV, document, or DB resources. Remote pubsub subscribe uses the same
-`resource:pubsub:<topic>` before returning the peer's current daemon-lifetime
+protected path, requires `pubsub.subscribe` on `resource:pubsub:<topic>`, joins
-snapshot for that topic.
+the gossip topic, and returns the peer's current daemon-lifetime snapshot for
+that topic.
 Remote pipe connect uses the same protected Iroh control path and requires
 `pipe.connect` on `resource:pipe:<name>`. The current prototype records a remote
 connection attempt and whether a listener exists. `geth pipe send <name>
@@ -313,8 +315,8 @@ Everything meaningful is modeled as a resource. Planned resource kinds are:
 - `pipe`: dumbpipe-like byte streams over Iroh; the bootstrap has a local
   daemon registry only
 - `document`: Automerge documents over Iroh streams
-- `pubsub`: lossy notifications, not authoritative storage; the bootstrap
+- `pubsub`: lossy notifications over native `iroh-gossip` after geth
-  keeps only an in-memory daemon-lifetime ring buffer
+  authorization, with only an in-memory daemon-lifetime ring buffer
 - `cas`: content-addressed blob storage and distribution
 - `ssh-proxy`: authorized SSH proxy/admin access over Iroh
 

crates/geth-node/Cargo.toml

@@ -7,6 +7,8 @@ license.workspace = true
 
 [dependencies]
 base64.workspace = true
+blake3.workspace = true
+bytes.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true

crates/geth-node/src/lib.rs

@@ -175,6 +175,12 @@ struct NodeRuntime {
 #[derive(Debug, Default)]
 struct PubsubRuntime {
     messages: VecDeque<PubsubMessage>,
+    gossip_topics: BTreeMap<String, PubsubGossipTopicRuntime>,
+}
+
+#[derive(Debug, Clone)]
+struct PubsubGossipTopicRuntime {
+    sender: iroh_gossip::api::GossipSender,
 }
 
 #[derive(Debug, Default)]
@@ -785,6 +791,15 @@ pub async fn handle_request_async(
                         .unwrap_or(entry.store.as_str());
                     mirror_kv_store_to_iroh_docs(node, name).await?;
                 }
+                ControlResponse::PubsubPublished { message } => {
+                    broadcast_pubsub_gossip(
+                        node,
+                        message.topic.as_str(),
+                        message.message.as_str(),
+                        Vec::new(),
+                    )
+                    .await?;
+                }
                 _ => {}
             }
             Ok(response)
@@ -2169,6 +2184,8 @@ async fn pubsub_publish_to_peer(
 ) -> Result<ControlResponse, NodeError> {
     geth_pubsub::validate_topic(&topic)?;
     geth_pubsub::validate_message(&message)?;
+    let bootstrap_peer = peer_gossip_endpoint_id(node, peer_node)?;
+    ensure_pubsub_gossip_topic(node, topic.as_str(), vec![bootstrap_peer]).await?;
     let response = request_peer_control(node, peer_node, "pubsub-publish", |peer_card, nonce| {
         PeerControlRequest::PubsubPublish {
             peer_card,
@@ -2224,6 +2241,8 @@ async fn pubsub_subscribe_from_peer(
     bearer_secret: Option<String>,
 ) -> Result<ControlResponse, NodeError> {
     geth_pubsub::validate_topic(&topic)?;
+    let bootstrap_peer = peer_gossip_endpoint_id(node, peer_node)?;
+    ensure_pubsub_gossip_topic(node, topic.as_str(), vec![bootstrap_peer]).await?;
     let response = request_peer_control(node, peer_node, "pubsub-subscribe", |peer_card, nonce| {
         PeerControlRequest::PubsubSubscribe {
             peer_card,
@@ -5429,8 +5448,20 @@ async fn handle_iroh_control_connection(
                 &nonce,
                 bearer_proof.as_ref(),
             )?;
+            drop(store);
             let published = if explanation.allowed {
-                Some(record_pubsub_message(&node, topic, message)?)
+                let bootstrap_peer = remote_endpoint_id
+                    .parse::<iroh::EndpointId>()
+                    .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+                let published = record_pubsub_message(&node, topic.clone(), message.clone())?;
+                broadcast_pubsub_gossip(
+                    &node,
+                    topic.as_str(),
+                    message.as_str(),
+                    vec![bootstrap_peer],
+                )
+                .await?;
+                Some(published)
             } else {
                 None
             };
@@ -5477,7 +5508,12 @@ async fn handle_iroh_control_connection(
                 &nonce,
                 bearer_proof.as_ref(),
             )?;
+            drop(store);
             let messages = if explanation.allowed {
+                let bootstrap_peer = remote_endpoint_id
+                    .parse::<iroh::EndpointId>()
+                    .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+                ensure_pubsub_gossip_topic(&node, topic.as_str(), vec![bootstrap_peer]).await?;
                 pubsub_messages_for_topic(&node, &topic)?
             } else {
                 Vec::new()
@@ -6384,6 +6420,28 @@ fn ensure_peer_card_matches_endpoint(card: &PeerCard, endpoint_id: &str) -> Resu
     }
 }
 
+fn peer_gossip_endpoint_id(
+    node: &LocalNode,
+    peer_node: &str,
+) -> Result<iroh::EndpointId, NodeError> {
+    let store = Store::open(&node.paths.metadata_db())?;
+    let peer_node = resolve_peer_node_for_control(&store, peer_node);
+    let stored = store
+        .get_peer_card(&peer_node)?
+        .ok_or_else(|| NodeError::PeerNotFound(peer_node.clone()))?;
+    let peer_card: PeerCard = serde_json::from_str(&stored.card_json)?;
+    peer_card.validate_candidate()?;
+    let candidate = peer_card
+        .endpoints
+        .first()
+        .ok_or(geth_discovery::DiscoveryError::MissingEndpoint)?;
+    ensure_peer_card_matches_endpoint(&peer_card, &candidate.endpoint_id)?;
+    candidate
+        .endpoint_id
+        .parse::<iroh::EndpointId>()
+        .map_err(|error| NodeError::IrohPeer(error.to_string()))
+}
+
 fn restricted_admin_shell_output(node: &LocalNode, command: &str) -> Result<String, NodeError> {
     match command {
         "help" => Ok("commands: help, status, node-id".to_owned()),
@@ -8078,6 +8136,13 @@ fn native_backend_statuses() -> Vec<NativeBackendStatus> {
                         library.alpn
                     ),
                 ),
+                "pubsub" => (
+                    "wired",
+                    format!(
+                        "none; pubsub topics broadcast over native ALPN {} after geth control authorization preflight",
+                        library.alpn
+                    ),
+                ),
                 _ => (
                     "ready-to-wire",
                     format!(
@@ -8190,11 +8255,20 @@ fn record_pubsub_message(
     node: &LocalNode,
     topic: String,
     message: String,
+) -> Result<PubsubMessage, NodeError> {
+    record_pubsub_message_at(node, topic, message, UnixMillis(geth_store::now_ms()))
+}
+
+fn record_pubsub_message_at(
+    node: &LocalNode,
+    topic: String,
+    message: String,
+    published_at: UnixMillis,
 ) -> Result<PubsubMessage, NodeError> {
     let message = PubsubMessage {
         topic: topic.into(),
         message,
-        published_at: UnixMillis(geth_store::now_ms()),
+        published_at,
     };
     let mut runtime = node
         .runtime
@@ -8225,6 +8299,158 @@ fn pubsub_messages_for_topic(
         .collect())
 }
 
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+struct PubsubGossipPayload {
+    version: u8,
+    source_node_id: String,
+    source_agent_id: String,
+    topic: String,
+    message: String,
+    published_at_ms: i64,
+}
+
+fn pubsub_gossip_topic_id(topic: &str) -> iroh_gossip::TopicId {
+    blake3::hash(format!("geth.pubsub.v1\0{topic}").as_bytes()).into()
+}
+
+async fn ensure_pubsub_gossip_topic(
+    node: &LocalNode,
+    topic: &str,
+    bootstrap: Vec<iroh::EndpointId>,
+) -> Result<Option<iroh_gossip::api::GossipSender>, NodeError> {
+    let existing = {
+        node.runtime
+            .pubsub
+            .lock()
+            .map_err(|_| NodeError::RuntimeLockPoisoned)?
+            .gossip_topics
+            .get(topic)
+            .cloned()
+    };
+    if let Some(existing) = existing {
+        if !bootstrap.is_empty() {
+            existing
+                .sender
+                .join_peers(bootstrap)
+                .await
+                .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+        }
+        return Ok(Some(existing.sender));
+    }
+
+    let Some(gossip) = iroh_gossip(node)? else {
+        return Ok(None);
+    };
+    let topic_id = pubsub_gossip_topic_id(topic);
+    let subscribed = gossip
+        .subscribe(topic_id, bootstrap)
+        .await
+        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+    let (sender, mut receiver) = subscribed.split();
+    let topic_name = topic.to_owned();
+    let receiver_node = node.clone();
+    tokio::spawn(async move {
+        while let Some(event) = receiver.next().await {
+            match event {
+                Ok(iroh_gossip::api::Event::Received(message)) => {
+                    if message.content.len() > 64 * 1024 {
+                        tracing::warn!(
+                            topic = %topic_name,
+                            bytes = message.content.len(),
+                            "dropping oversized pubsub gossip message"
+                        );
+                        continue;
+                    }
+                    match serde_json::from_slice::<PubsubGossipPayload>(&message.content) {
+                        Ok(payload) => {
+                            if let Err(error) =
+                                record_pubsub_gossip_payload(&receiver_node, &topic_name, payload)
+                            {
+                                tracing::warn!(%error, topic = %topic_name, "dropping invalid pubsub gossip message");
+                            }
+                        }
+                        Err(error) => {
+                            tracing::warn!(%error, topic = %topic_name, "dropping undecodable pubsub gossip message");
+                        }
+                    }
+                }
+                Ok(iroh_gossip::api::Event::Lagged) => {
+                    tracing::warn!(topic = %topic_name, "pubsub gossip receiver lagged");
+                }
+                Ok(iroh_gossip::api::Event::NeighborUp(endpoint_id)) => {
+                    tracing::debug!(%endpoint_id, topic = %topic_name, "pubsub gossip neighbor up");
+                }
+                Ok(iroh_gossip::api::Event::NeighborDown(endpoint_id)) => {
+                    tracing::debug!(%endpoint_id, topic = %topic_name, "pubsub gossip neighbor down");
+                }
+                Err(error) => {
+                    tracing::warn!(%error, topic = %topic_name, "pubsub gossip receiver closed");
+                    break;
+                }
+            }
+        }
+    });
+
+    node.runtime
+        .pubsub
+        .lock()
+        .map_err(|_| NodeError::RuntimeLockPoisoned)?
+        .gossip_topics
+        .insert(
+            topic.to_owned(),
+            PubsubGossipTopicRuntime {
+                sender: sender.clone(),
+            },
+        );
+    Ok(Some(sender))
+}
+
+fn record_pubsub_gossip_payload(
+    node: &LocalNode,
+    expected_topic: &str,
+    payload: PubsubGossipPayload,
+) -> Result<(), NodeError> {
+    if payload.version != 1
+        || payload.topic != expected_topic
+        || payload.source_node_id == node.node_id
+    {
+        return Ok(());
+    }
+    geth_pubsub::validate_topic(&payload.topic)?;
+    geth_pubsub::validate_message(&payload.message)?;
+    record_pubsub_message_at(
+        node,
+        payload.topic,
+        payload.message,
+        UnixMillis(payload.published_at_ms),
+    )?;
+    Ok(())
+}
+
+async fn broadcast_pubsub_gossip(
+    node: &LocalNode,
+    topic: &str,
+    message: &str,
+    bootstrap: Vec<iroh::EndpointId>,
+) -> Result<(), NodeError> {
+    let Some(sender) = ensure_pubsub_gossip_topic(node, topic, bootstrap).await? else {
+        return Ok(());
+    };
+    let payload = PubsubGossipPayload {
+        version: 1,
+        source_node_id: node.node_id.clone(),
+        source_agent_id: node.agent_id.clone(),
+        topic: topic.to_owned(),
+        message: message.to_owned(),
+        published_at_ms: geth_store::now_ms(),
+    };
+    let bytes = serde_json::to_vec(&payload)?;
+    sender
+        .broadcast(bytes::Bytes::from(bytes))
+        .await
+        .map_err(|error| NodeError::IrohPeer(error.to_string()))
+}
+
 fn record_pipe_listener(node: &LocalNode, name: String) -> Result<PipeListener, NodeError> {
     let listener = PipeListener {
         id: format!("pipe:{name}").into(),
@@ -11874,6 +12100,23 @@ mod tests {
             }
             other => panic!("unexpected allowed pubsub publish response: {other:?}"),
         }
+        let mut gossiped_to_left = false;
+        for _ in 0..20 {
+            let local_messages = pubsub_messages_for_topic(&left, "presence/test")
+                .expect("left pubsub local messages");
+            if local_messages
+                .iter()
+                .any(|message| message.message == "hello")
+            {
+                gossiped_to_left = true;
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(50)).await;
+        }
+        assert!(
+            gossiped_to_left,
+            "authorized remote pubsub publish should also arrive through iroh-gossip"
+        );
         let remote_messages = handle_request(
             &right,
             ControlRequest::PubsubSub {

crates/geth/tests/bootstrap.rs

@@ -163,7 +163,7 @@ fn geth_status_against_running_daemon() {
         )
     );
     assert!(stdout.contains(
-        "native backend pubsub: shared-iroh-endpoint-ready target iroh-gossip 0.95.0 (ready-to-wire)"
+        "native backend pubsub: shared-iroh-endpoint-ready target iroh-gossip 0.95.0 (wired)"
     ));
 }
 

docs/adr/0013-pipe-and-pubsub.md

@@ -12,8 +12,12 @@ status, and ephemeral notifications.
 
 Pubsub is not authoritative storage. Important facts must live in durable
 resources such as kv, document, db, CAS manifests, or future auth logs.
+The prototype uses the geth control ALPN as an authorization preflight before a
+daemon joins or broadcasts on a deterministic native `iroh-gossip` topic.
 
 ## Consequences
 
 The modules can support ad hoc workflow and presence without confusing gossip
 with durable state.
+Gossip delivery is best-effort. Consumers must tolerate missed, duplicated, and
+out-of-order notifications and use durable resources for reconciliation.

docs/architecture.md

@@ -38,16 +38,15 @@ geth control-ALPN authorization preflight before transferring payload bytes over
 `iroh-blobs`. KV now starts `iroh-docs` with `iroh-gossip` and the same native
 blob store, mirrors named KV stores into read-shared Iroh Documents namespaces,
 and sends read-only docs tickets only after geth control authorization succeeds.
-`geth status` reports CAS and KV as wired native backends. Pubsub still uses its
+Pubsub joins deterministic native `iroh-gossip` topics after the geth control
-explicit documented bootstrap equivalent until it is migrated to native gossip.
+path authenticates the peer-card endpoint binding and authorizes the topic
+capability. `geth status` reports CAS, KV, and pubsub as wired native backends.
 
 Module ALPNs are registered through `geth-iroh`'s protocol router scaffold. The
 router owns the default protocol descriptors, rejects duplicate ALPN
 registrations, and returns explicit unknown-ALPN errors. The current daemon
 accept loop dispatches geth control, pipe, SSH-proxy, and native CAS blob
-streams directly, plus native docs and gossip streams used by KV. The next
+streams directly, plus native docs and gossip streams used by KV and pubsub.
-backend migration should attach application pubsub behavior to iroh-gossip
-instead of creating parallel endpoints.
 
 The target product should use Iroh relay support for practical internet
 connectivity and mDNS/LAN discovery for local networks. These are connectivity
@@ -228,19 +227,18 @@ loop runs the same sync for local documents and known peers using
 per-peer/per-document cursors. Received Automerge documents are merged before
 being stored.
 
-`geth-pubsub` currently supports local publish/subscribe snapshots through the
+`geth-pubsub` supports local publish/subscribe snapshots through the daemon
-daemon control protocol. Messages live in a bounded in-memory ring buffer and
+control protocol. Messages live in a bounded in-memory ring buffer and are lost
-are lost when the daemon stops. This is deliberate: pubsub is a lossy wakeup and
+when the daemon stops. This is deliberate: pubsub is a lossy wakeup and presence
-presence channel, not authoritative storage. Durable facts must be written to
+channel, not authoritative storage. Durable facts must be written to CAS, KV,
-CAS, KV, document, or DB resources before pubsub is used as a wakeup. `geth
+document, or DB resources before pubsub is used as a wakeup. `geth pubsub pub
-pubsub pub <topic> <message> --node <node-id>` can publish to an imported peer
+<topic> <message> --node <node-id>` first uses the protected Iroh control ALPN
-over the protected Iroh control ALPN. The remote daemon validates endpoint/card
+for authorization. The remote daemon validates endpoint/card binding and
-binding and requires
+requires `pubsub.publish` on `resource:pubsub:<topic>` before recording the
-`pubsub.publish` on `resource:pubsub:<topic>` before recording the message in
+message and broadcasting it through a deterministic native `iroh-gossip` topic.
-its local ring buffer. `geth pubsub sub <topic> --node <node-id>` can read an
+`geth pubsub sub <topic> --node <node-id>` uses the same protected path, joins
-authorized peer's current snapshot for that topic over the same protected path
+the gossip topic when the caller has `pubsub.subscribe`, and returns the peer's
-when the caller has `pubsub.subscribe` on `resource:pubsub:<topic>`. Iroh-gossip
+current daemon-lifetime snapshot. Private topics remain future work.
-replication and private topics are future work.
 
 `geth-pipe` currently supports `pipe listen/connect/send/recv` against a
 daemon-lifetime runtime. `geth pipe connect <name> --node <node-id>` sends an

docs/roadmap.md

@@ -505,7 +505,7 @@ authorization and durable-state boundaries clear.
     capabilities, after geth control authorization succeeds.
   - `[x]` Tests cover imported KV docs state after authorized remote sync.
 
-- `[~]` Iroh-gossip pubsub integration.
+- `[x]` Iroh-gossip pubsub integration.
   Acceptance criteria:
   - `[x]` `geth pubsub pub/sub` works against the local daemon.
   - `[x]` Local pubsub messages are kept in a bounded daemon-lifetime ring
@@ -521,7 +521,9 @@ authorization and durable-state boundaries clear.
   - `[x]` Remote pubsub subscribe requires `pubsub.subscribe` on
     `resource:pubsub:<topic>`.
   - `[x]` Tests cover denied and allowed remote pubsub subscribe.
-  - `[ ]` Replace bootstrap remote publish with iroh-gossip topics.
+  - `[x]` Replace bootstrap remote publish with iroh-gossip topics.
+  - `[x]` Authorized remote publish/subscribe joins deterministic
+    `iroh-gossip` topics after geth control authorization.
   - `[x]` Docs and tests keep durable state in CAS/KV/document/db instead.
 
 ## Phase 4: Pipes, SSH Proxy, And SSH Distribution