Subscribe to peer pubsub snapshots

AGENTS.md

@@ -155,8 +155,10 @@ Roadmap items should be actionable and checkable:
   and true CRDT sync are still roadmap work.
 - Pubsub supports local daemon-lifetime publish/subscribe snapshots through a
   bounded in-memory ring buffer. Remote publish over the protected Iroh control
-  ALPN requires `pubsub.publish` on `resource:pubsub:<topic>`. Iroh-gossip
+  ALPN requires `pubsub.publish` on `resource:pubsub:<topic>`. Remote subscribe
-  replication, private topics, and remote subscribe are still roadmap work.
+  over the same path requires `pubsub.subscribe` and returns the peer's current
+  daemon-lifetime snapshot. Iroh-gossip replication and private topics are still
+  roadmap work.
 - Pipe listen/connect supports a daemon-lifetime registry. Remote
   `geth pipe connect <name> --node <node-id>` uses the protected Iroh control
   ALPN and requires `pipe.connect` on `resource:pipe:<name>`. Iroh byte streams,

README.md

@@ -109,7 +109,8 @@ The bootstrap implementation provides:
 - local JSON document commands: `geth document create/status/set/get`; `geth
   document sync <node-id> <name>` pulls authorized remote JSON state
 - local daemon-lifetime pubsub snapshots: `geth pubsub pub/sub`; `geth pubsub
-  pub <topic> <message> --node <node-id>` publishes to an authorized peer
+  pub <topic> <message> --node <node-id>` publishes to an authorized peer;
+  `geth pubsub sub <topic> --node <node-id>` reads an authorized peer snapshot
 - SSH certificate flow metadata:
   - `geth ssh cert request --public-key <path> --principal <name>`
   - `geth ssh cert requests`
@@ -157,7 +158,9 @@ older than the local value.
 Remote pubsub publish uses the protected Iroh control path too. The remote peer
 requires `pubsub.publish` on `resource:pubsub:<topic>` before recording the
 message in its local daemon-lifetime ring buffer. Pubsub remains lossy and is
-not durable storage.
+not durable storage. Remote pubsub subscribe uses the same protected path and
+requires `pubsub.subscribe` on `resource:pubsub:<topic>` before returning 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; byte streaming and forwarding

crates/geth-cli/src/lib.rs

@@ -320,6 +320,8 @@ pub enum PubsubCommand {
     },
     Sub {
         topic: String,
+        #[arg(long)]
+        node: Option<String>,
     },
 }
 
@@ -653,7 +655,7 @@ fn request_for_command(command: Command) -> Result<ControlRequest> {
                 message,
                 node,
             },
-            PubsubCommand::Sub { topic } => ControlRequest::PubsubSub { topic },
+            PubsubCommand::Sub { topic, node } => ControlRequest::PubsubSub { topic, node },
         },
         Command::Pipe { command } => match command {
             PipeCommand::Listen { name } => ControlRequest::PipeListen { name },
@@ -1477,6 +1479,32 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> {
             }
             println!("note: {note}");
         }
+        ControlResponse::PubsubRemoteMessages {
+            peer_node_id,
+            peer_agent_id,
+            endpoint_id,
+            topic,
+            messages,
+            allowed,
+            reason,
+            note,
+        } => {
+            if allowed {
+                println!("topic: {topic}");
+                println!("peer: {peer_node_id}");
+                println!("messages: {}", messages.len());
+                for message in messages {
+                    println!("{}\t{}", message.published_at.0, message.message);
+                }
+            } else {
+                println!("pubsub subscribe denied by {peer_node_id}");
+            }
+            println!("agent: {peer_agent_id}");
+            println!("endpoint: {endpoint_id}");
+            println!("allowed: {allowed}");
+            println!("reason: {reason}");
+            println!("note: {note}");
+        }
         ControlResponse::PipeListening { listener } => {
             println!("listening pipe: {}", listener.name);
             println!("id: {}", listener.id);

crates/geth-control/src/lib.rs

@@ -229,6 +229,7 @@ pub enum ControlRequest {
     },
     PubsubSub {
         topic: String,
+        node: Option<String>,
     },
     PipeListen {
         name: String,
@@ -498,6 +499,16 @@ pub enum ControlResponse {
         messages: Vec<PubsubMessage>,
         note: String,
     },
+    PubsubRemoteMessages {
+        peer_node_id: String,
+        peer_agent_id: String,
+        endpoint_id: String,
+        topic: String,
+        messages: Vec<PubsubMessage>,
+        allowed: bool,
+        reason: String,
+        note: String,
+    },
     PipeListening {
         listener: PipeListener,
     },
@@ -602,6 +613,11 @@ pub enum PeerControlRequest {
         message: String,
         nonce: String,
     },
+    PubsubSubscribe {
+        peer_card: PeerCard,
+        topic: String,
+        nonce: String,
+    },
     PipeConnect {
         peer_card: PeerCard,
         target: String,
@@ -723,6 +739,19 @@ pub enum PeerControlResponse {
         nonce: String,
         note: String,
     },
+    PubsubSubscribed {
+        node_id: String,
+        agent_id: String,
+        endpoint_id: String,
+        remote_endpoint_id: String,
+        topic: String,
+        messages: Vec<PubsubMessage>,
+        allowed: bool,
+        reason: String,
+        evaluated_ops: usize,
+        nonce: String,
+        note: String,
+    },
     PipeConnected {
         node_id: String,
         agent_id: String,
@@ -987,6 +1016,15 @@ mod tests {
             request
         );
 
+        let request = ControlRequest::PubsubSub {
+            topic: "presence/test".to_owned(),
+            node: Some("node:peer".to_owned()),
+        };
+        assert_eq!(
+            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
+            request
+        );
+
         let response = ControlResponse::PubsubRemotePublished {
             peer_node_id: "node:peer".to_owned(),
             peer_agent_id: "agent:peer".to_owned(),
@@ -1005,6 +1043,25 @@ mod tests {
             response
         );
 
+        let response = ControlResponse::PubsubRemoteMessages {
+            peer_node_id: "node:peer".to_owned(),
+            peer_agent_id: "agent:peer".to_owned(),
+            endpoint_id: "endpoint:peer".to_owned(),
+            topic: "presence/test".to_owned(),
+            messages: vec![PubsubMessage {
+                topic: "presence/test".into(),
+                message: "online".to_owned(),
+                published_at: geth_types::UnixMillis(1),
+            }],
+            allowed: true,
+            reason: "direct grant".to_owned(),
+            note: "lossy".to_owned(),
+        };
+        assert_eq!(
+            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
+            response
+        );
+
         let response = ControlResponse::PipeRemoteConnected {
             peer_node_id: "node:peer".to_owned(),
             peer_agent_id: "agent:peer".to_owned(),
@@ -1303,6 +1360,50 @@ mod tests {
             response
         );
 
+        let request = PeerControlRequest::PubsubSubscribe {
+            peer_card: PeerCard {
+                node_id: "node:caller".into(),
+                agent_id: "agent:caller".into(),
+                endpoints: Vec::new(),
+                issued_at: geth_types::UnixMillis(1),
+                signature: geth_discovery::SignatureMetadata {
+                    namespace: "geth.peer-card.v1@geth.local".to_owned(),
+                    signer: "agent:caller".to_owned(),
+                    public_key: "key".to_owned(),
+                    signature: "sig".to_owned(),
+                },
+            },
+            topic: "presence/test".to_owned(),
+            nonce: "nonce".to_owned(),
+        };
+        assert_eq!(
+            decode_peer_request(&encode_peer_request(&request).expect("encode")).expect("decode"),
+            request
+        );
+
+        let response = PeerControlResponse::PubsubSubscribed {
+            node_id: "node:peer".to_owned(),
+            agent_id: "agent:peer".to_owned(),
+            endpoint_id: "endpoint:peer".to_owned(),
+            remote_endpoint_id: "endpoint:caller".to_owned(),
+            topic: "presence/test".to_owned(),
+            messages: vec![PubsubMessage {
+                topic: "presence/test".into(),
+                message: "online".to_owned(),
+                published_at: geth_types::UnixMillis(1),
+            }],
+            allowed: true,
+            reason: "direct grant".to_owned(),
+            evaluated_ops: 1,
+            nonce: "nonce".to_owned(),
+            note: "lossy".to_owned(),
+        };
+        assert_eq!(
+            decode_peer_response(&encode_peer_response(&response).expect("encode"))
+                .expect("decode"),
+            response
+        );
+
         let response = PeerControlResponse::PipeConnected {
             node_id: "node:peer".to_owned(),
             agent_id: "agent:peer".to_owned(),

crates/geth-node/src/lib.rs

@@ -273,6 +273,10 @@ pub async fn handle_request_async(
             message,
             node: Some(peer_node),
         } => pubsub_publish_to_peer(node, &peer_node, topic, message).await,
+        ControlRequest::PubsubSub {
+            topic,
+            node: Some(peer_node),
+        } => pubsub_subscribe_from_peer(node, &peer_node, topic).await,
         ControlRequest::PipeConnect {
             target,
             node: Some(peer_node),
@@ -560,6 +564,7 @@ async fn peer_ping(node: &LocalNode, peer_node: &str) -> Result<ControlResponse,
         | PeerControlResponse::SshRevocationSynced { .. }
         | PeerControlResponse::KvSynced { .. }
         | PeerControlResponse::PubsubPublished { .. }
+        | PeerControlResponse::PubsubSubscribed { .. }
         | PeerControlResponse::PipeConnected { .. }
         | PeerControlResponse::DocumentSynced { .. }
         | PeerControlResponse::DbSynced { .. } => Err(NodeError::IrohPeer(
@@ -673,6 +678,7 @@ async fn peer_auth_check(
         | PeerControlResponse::SshRevocationSynced { .. }
         | PeerControlResponse::KvSynced { .. }
         | PeerControlResponse::PubsubPublished { .. }
+        | PeerControlResponse::PubsubSubscribed { .. }
         | PeerControlResponse::PipeConnected { .. }
         | PeerControlResponse::DocumentSynced { .. }
         | PeerControlResponse::DbSynced { .. } => Err(NodeError::IrohPeer(
@@ -815,6 +821,7 @@ async fn cas_fetch_from_peer(
         | PeerControlResponse::SshRevocationSynced { .. }
         | PeerControlResponse::KvSynced { .. }
         | PeerControlResponse::PubsubPublished { .. }
+        | PeerControlResponse::PubsubSubscribed { .. }
         | PeerControlResponse::PipeConnected { .. }
         | PeerControlResponse::DocumentSynced { .. }
         | PeerControlResponse::DbSynced { .. } => Err(NodeError::IrohPeer(
@@ -1093,6 +1100,51 @@ async fn pubsub_publish_to_peer(
     }
 }
 
+async fn pubsub_subscribe_from_peer(
+    node: &LocalNode,
+    peer_node: &str,
+    topic: String,
+) -> Result<ControlResponse, NodeError> {
+    geth_pubsub::validate_topic(&topic)?;
+    let response = request_peer_control(node, peer_node, "pubsub-subscribe", |peer_card, nonce| {
+        PeerControlRequest::PubsubSubscribe {
+            peer_card,
+            topic: topic.clone(),
+            nonce,
+        }
+    })
+    .await?;
+    match response {
+        PeerControlResponse::PubsubSubscribed {
+            node_id,
+            agent_id,
+            endpoint_id,
+            topic: response_topic,
+            messages,
+            allowed,
+            reason,
+            note,
+            ..
+        } if response_topic == topic => Ok(ControlResponse::PubsubRemoteMessages {
+            peer_node_id: node_id,
+            peer_agent_id: agent_id,
+            endpoint_id,
+            topic: response_topic,
+            messages,
+            allowed,
+            reason,
+            note,
+        }),
+        PeerControlResponse::PubsubSubscribed { .. } => Err(NodeError::IrohPeer(
+            "peer pubsub subscribe response did not match request".to_owned(),
+        )),
+        PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
+        _ => Err(NodeError::IrohPeer(
+            "peer returned wrong response type to pubsub subscribe".to_owned(),
+        )),
+    }
+}
+
 async fn pipe_connect_to_peer(
     node: &LocalNode,
     peer_node: &str,
@@ -1549,6 +1601,10 @@ async fn request_peer_control(
             nonce: response_nonce,
             ..
         }
+        | PeerControlResponse::PubsubSubscribed {
+            nonce: response_nonce,
+            ..
+        }
         | PeerControlResponse::PipeConnected {
             nonce: response_nonce,
             ..
@@ -2082,6 +2138,52 @@ async fn handle_iroh_control_connection(
                 note: "pubsub publish authenticated endpoint/card binding and required pubsub.publish on the remote topic resource; pubsub is lossy".to_owned(),
             }
         }
+        PeerControlRequest::PubsubSubscribe {
+            peer_card,
+            topic,
+            nonce,
+        } => {
+            geth_pubsub::validate_topic(&topic)?;
+            peer_card.validate_candidate()?;
+            ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?;
+            let discovered = DiscoveredPeer::candidate(
+                peer_card.clone(),
+                UnixMillis(geth_store::now_ms()),
+                DiscoverySource::PeerExchange,
+            )?;
+            let store = Store::open(&node.paths.metadata_db())?;
+            store.upsert_peer_card(&StoredPeerCard {
+                peer_id: peer_card.node_id.to_string(),
+                card_json: serde_json::to_string(&peer_card)?,
+                updated_at_ms: discovered.discovered_at.0,
+            })?;
+            let resource = format!("resource:pubsub:{topic}");
+            let capability = "pubsub.subscribe".to_owned();
+            let explanation = geth_auth::explain_auth_ops(
+                &load_auth_ops_for_resource(&store, &resource)?,
+                PrincipalId::new(peer_card.node_id.to_string()),
+                ResourceId::new(resource),
+                Capability::new(capability),
+            );
+            let messages = if explanation.allowed {
+                pubsub_messages_for_topic(&node, &topic)?
+            } else {
+                Vec::new()
+            };
+            PeerControlResponse::PubsubSubscribed {
+                node_id: node.node_id.clone(),
+                agent_id: node.agent_id.clone(),
+                endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(),
+                remote_endpoint_id,
+                topic,
+                messages,
+                allowed: explanation.allowed,
+                reason: explanation.reason,
+                evaluated_ops: explanation.evaluated_ops,
+                nonce,
+                note: "pubsub subscribe authenticated endpoint/card binding and required pubsub.subscribe on the remote topic resource; pubsub is lossy daemon-lifetime state".to_owned(),
+            }
+        }
         PeerControlRequest::PipeConnect {
             peer_card,
             target,
@@ -3266,25 +3368,16 @@ pub fn handle_request(
             Ok(ControlResponse::PubsubPublished { message })
         }
         ControlRequest::PubsubPub { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable),
-        ControlRequest::PubsubSub { topic } => {
+        ControlRequest::PubsubSub { topic, node: None } => {
             geth_pubsub::validate_topic(&topic)?;
-            let runtime = node
+            let messages = pubsub_messages_for_topic(node, &topic)?;
-                .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::PubsubSub { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable),
         ControlRequest::PipeListen { name } => {
             geth_pipe::validate_pipe_name(&name)?;
             let listener = PipeListener {
@@ -3428,6 +3521,23 @@ fn record_pubsub_message(
     Ok(message)
 }
 
+fn pubsub_messages_for_topic(
+    node: &LocalNode,
+    topic: &str,
+) -> Result<Vec<PubsubMessage>, NodeError> {
+    let runtime = node
+        .runtime
+        .pubsub
+        .lock()
+        .map_err(|_| NodeError::RuntimeLockPoisoned)?;
+    Ok(runtime
+        .messages
+        .iter()
+        .filter(|message| message.topic.as_str() == topic)
+        .cloned()
+        .collect())
+}
+
 fn record_pipe_connection(
     node: &LocalNode,
     target: String,
@@ -4319,6 +4429,29 @@ mod tests {
             other => panic!("unexpected denied pubsub publish response: {other:?}"),
         }
 
+        let denied_pubsub_subscribe = handle_request_async(
+            &left,
+            ControlRequest::PubsubSub {
+                topic: "presence/test".to_owned(),
+                node: Some(right_card.node_id.to_string()),
+            },
+        )
+        .await
+        .expect("denied remote pubsub subscribe");
+        match denied_pubsub_subscribe {
+            ControlResponse::PubsubRemoteMessages {
+                allowed,
+                messages,
+                reason,
+                ..
+            } => {
+                assert!(!allowed);
+                assert!(messages.is_empty());
+                assert!(reason.contains("no active direct or group grant"));
+            }
+            other => panic!("unexpected denied pubsub subscribe response: {other:?}"),
+        }
+
         let denied_pipe = handle_request_async(
             &left,
             ControlRequest::PipeConnect {
@@ -4421,6 +4554,16 @@ mod tests {
             },
         )
         .expect("grant left pubsub publish");
+        handle_request(
+            &right,
+            ControlRequest::AuthGrant {
+                subject: left.node_id.clone(),
+                resource: "resource:pubsub:presence/test".to_owned(),
+                capability: "pubsub.subscribe".to_owned(),
+                grant_id: Some("grant:left-pubsub-subscribe".to_owned()),
+            },
+        )
+        .expect("grant left pubsub subscribe");
         handle_request(
             &right,
             ControlRequest::AuthGrant {
@@ -4578,6 +4721,7 @@ mod tests {
             &right,
             ControlRequest::PubsubSub {
                 topic: "presence/test".to_owned(),
+                node: None,
             },
         )
         .expect("right pubsub sub after remote publish");
@@ -4589,6 +4733,32 @@ mod tests {
             other => panic!("unexpected remote pubsub messages response: {other:?}"),
         }
 
+        let subscribed = handle_request_async(
+            &left,
+            ControlRequest::PubsubSub {
+                topic: "presence/test".to_owned(),
+                node: Some(right_card.node_id.to_string()),
+            },
+        )
+        .await
+        .expect("allowed remote pubsub subscribe");
+        match subscribed {
+            ControlResponse::PubsubRemoteMessages {
+                allowed,
+                messages,
+                reason,
+                note,
+                ..
+            } => {
+                assert!(allowed);
+                assert_eq!(messages.len(), 1);
+                assert_eq!(messages[0].message, "hello");
+                assert!(reason.contains("direct grant"));
+                assert!(note.contains("pubsub.subscribe"));
+            }
+            other => panic!("unexpected allowed pubsub subscribe response: {other:?}"),
+        }
+
         let remote_pipe = handle_request_async(
             &left,
             ControlRequest::PipeConnect {

crates/geth/tests/bootstrap.rs

@@ -1241,6 +1241,7 @@ fn pubsub_pub_sub_uses_lossy_in_memory_runtime() {
         &node,
         geth_control::ControlRequest::PubsubSub {
             topic: "presence/laptop".to_owned(),
+            node: None,
         },
     )
     .expect("subscribe snapshot");
@@ -1263,6 +1264,7 @@ fn pubsub_pub_sub_uses_lossy_in_memory_runtime() {
         &reopened,
         geth_control::ControlRequest::PubsubSub {
             topic: "presence/laptop".to_owned(),
+            node: None,
         },
     )
     .expect("subscribe reopened snapshot");

docs/architecture.md

@@ -177,8 +177,10 @@ presence channel, not authoritative storage. `geth pubsub pub <topic> <message>
 --node <node-id>` can publish to an imported peer over the protected Iroh
 control ALPN. The remote daemon validates endpoint/card binding and requires
 `pubsub.publish` on `resource:pubsub:<topic>` before recording the message in
-its local ring buffer. Iroh-gossip replication and private topics are future
+its local ring buffer. `geth pubsub sub <topic> --node <node-id>` can read an
-work.
+authorized peer's current snapshot for that topic over the same protected path
+when the caller has `pubsub.subscribe` on `resource:pubsub:<topic>`. Iroh-gossip
+replication and private topics are future work.
 
 `geth-pipe` currently supports `pipe listen/connect` against a daemon-lifetime
 registry. `geth pipe connect <name> --node <node-id>` sends an authorized remote

docs/roadmap.md

@@ -294,8 +294,12 @@ authorization and durable-state boundaries clear.
     imported peer over Iroh.
   - `[x]` Remote pubsub publish requires `pubsub.publish` on
     `resource:pubsub:<topic>`.
+  - `[x]` `geth pubsub sub <topic> --node <node-id>` reads an authorized peer
+    snapshot over Iroh.
+  - `[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.
-  - `[ ]` Pubsub subscribe works across local test nodes over Iroh or gossip.
   - `[ ]` Docs and tests keep durable state in CAS/KV/document/db instead.
 
 ## Phase 4: Pipes, SSH Proxy, And SSH Distribution