test(protocols): run the req/resp sim checks between two real hosts

Tests/HostPair.lean

@@ -0,0 +1,83 @@
+import Conc
+import Streams
+import Libp2p.Identity
+import Libp2p.Security
+import Libp2p.Mux
+import Libp2p.Host
+import Multiformats.Multiaddr
+
+/-!
+# Two hosts over the sim
+
+The protocol suites run their handlers between two real hosts, on the
+test values of root §3.1 (PRINCIPLES 1.4): the in-memory sim transport,
+the fixed-key Noise value with Ed25519 identities from one byte each,
+and a test clock. The muxer is yamux. The server listens on
+`/memory/4100`, and the client holds a session to it.
+-/
+
+set_option autoImplicit false
+
+namespace Tests.HostPair
+
+open Conc (Async Clock OnceRef TestClock)
+open Libp2p.Host (Host Failure)
+
+/-- A host on `net` and `clock`: the fixed-key Noise value with an
+Ed25519 identity from one byte, yamux, and the sim transport. The
+ephemeral keys come from an `Rng` seeded with 32 copies of
+`idByte + 50`. Its failures go to `failures`. -/
+def mkHost (net : Streams.Sim.Net) (clock : Clock) (idByte : UInt8)
+    (failures : OnceRef (List Failure)) : Async Host := do
+  let .ok identity := Libp2p.Identity.PrivKey.ofData? .ed25519 ⟨Array.replicate 32 idByte⟩
+    | throw (IO.userError "test identity does not build")
+  let some static := Libp2p.Security.Noise.Bytes32.ofByteArray? ⟨Array.replicate 32 (idByte + 100)⟩
+    | throw (IO.userError "test static key is not 32 bytes")
+  let noise := Libp2p.Security.Noise.fixedKeyProtocol { identity, static }
+    ⟨Array.replicate 32 (idByte + 50)⟩
+  match ← (Host.new {
+      identity
+      transports := [net.transport]
+      security := [noise]
+      muxers := [Libp2p.Mux.yamux]
+      clock
+      onFailure := fun f => failures.update (f :: ·) } : BaseIO _) with
+  | .ok h => return h
+  | .error _ => throw (IO.userError "test host does not build")
+
+/-- Two hosts over a new sim net: the server listens, and the client
+holds a session to it. -/
+structure Pair where
+  client : Host
+  server : Host
+  /-- The peer id the client proved for the server. -/
+  serverPeer : Libp2p.Identity.PeerId
+  /-- The address the server listens on. -/
+  serverAddr : Multiformats.Multiaddr
+  /-- The failures of the server's owned tasks, newest first. -/
+  failures : OnceRef (List Failure)
+  /-- The test clock both hosts run on. `Pair.close` closes it. -/
+  clock : TestClock
+
+/-- The pair. With `split`, the sim delivers each byte as its own read. -/
+def hostPair (split : Bool := false) : Async Pair := do
+  let net ← Streams.Sim.Net.new split
+  let tc ← TestClock.new
+  let failures ← OnceRef.new []
+  let client ← mkHost net tc.clock 1 (← OnceRef.new [])
+  let server ← mkHost net tc.clock 2 failures
+  let addr ← match Multiformats.Multiaddr.parse? "/memory/4100" with
+    | .ok a => pure a
+    | .error _ => throw (IO.userError "test address")
+  let .ok bound ← server.startListening addr | throw (IO.userError "listen failed")
+  let .ok serverPeer ← client.dial bound | throw (IO.userError "dial failed")
+  return { client, server, serverPeer, serverAddr := bound, failures, clock := tc }
+
+/-- Close both hosts, then the clock. Each close joins its own tasks
+(§5.5). -/
+def Pair.close (p : Pair) : Async Unit := do
+  p.client.close
+  p.server.close
+  p.clock.close
+
+end Tests.HostPair

Tests/ReqResp.lean

@@ -1,7 +1,7 @@
 import Tests.Support
+import Tests.HostPair
 import Libp2p.Protocols.ReqResp.Exchange
-import Libp2p.Protocols.Mock.Muxer
-import Libp2p.Protocols.Mock.HostStream
+import Libp2p.Host
 import Multiformats.Multiaddr
 
 /-!
@@ -18,8 +18,8 @@ encoder shape writes through go's snappy.
 The sections follow the layers: the checksum, the chunk codec, the
 length prefix, the reader core as transcripts with no clock and no
 task, the writer, the concurrency limit, the exchange over a stream
-pair with the test clock, and the exchange over the sim transport
-through the mock muxer and the host stream mock.
+pair with the test clock, and the exchange between two hosts over the
+sim transport, with the fixed-key Noise value and yamux.
 -/
 
 set_option autoImplicit false
@@ -607,35 +607,17 @@ def limiterChecks (r : Runner) : IO Unit := do
 
 /-! ## Over the sim transport -/
 
-open Libp2p.Protocols.Mock.Muxer (muxer)
-open Libp2p.Protocols.Mock.HostStream (Host)
-
-/-- The peer id of a fixed Ed25519 seed of 32 copies of `v`. -/
-private def seedPeer (v : UInt8) : Async Libp2p.Identity.PeerId := do
-  let seed : ByteArray := ⟨Array.replicate 32 v⟩
-  match (do Libp2p.Identity.PeerId.ofPubKey (← (← Libp2p.Identity.PrivKey.ofData? .ed25519 seed).toPublic)
-      : Except Libp2p.Identity.Error Libp2p.Identity.PeerId) with
-  | .ok p => return p
-  | .error _ => throw (IO.userError "test: the fixed seed derives no peer id")
-
-/-- A connected pair of hosts over a new sim net and the mock muxer.
-Each end's handler context names the other end's peer and its own
-connection's remote address. -/
-def hostPair (split : Bool := false) : Async (Host × Host) := do
-  let net ← Streams.Sim.Net.new split
-  let addr ← match Multiformats.Multiaddr.parse? "/ip4/127.0.0.1/tcp/4100" with
-    | .ok a => pure a
-    | .error _ => throw (IO.userError "test address")
-  let l ← net.transport.listen addr
-  let some c ← net.transport.dial l.addr | throw (IO.userError "dial failed")
-  let some s ← l.accept | throw (IO.userError "accept failed")
-  l.close
-  let .ok cs ← (muxer.«open» .initiator c ByteArray.empty).run | throw (IO.userError "open failed")
-  let .ok ss ← (muxer.«open» .responder s ByteArray.empty).run | throw (IO.userError "open failed")
-  let client ← seedPeer 1
-  let server ← seedPeer 2
-  return (← Host.new cs { remotePeer := server, remoteAddr := c.remoteAddr },
-    ← Host.new ss { remotePeer := client, remoteAddr := s.remoteAddr })
+open Tests.HostPair (Pair hostPair)
+
+/-- The host's stream call for the server, as the `OpenStream` of
+`requestPeer`. A host failure that wraps a seam error reaches the
+requester as that error. The host renders its own refusals into
+`protocol`. -/
+def openStreamOf (p : Pair) : OpenStream := fun ids => do
+  match ← p.client.newStream p.serverPeer ids with
+  | .ok x => return .ok x
+  | .error (.stream e) | .error (.upgrade e) | .error (.session e) => return .error e
+  | .error e => return .error (.protocol (toString e))
 
 def simChecks (r : Runner) : Async Unit := do
   r.section "over the sim" "phase0 Req/Resp interaction; ARCHITECTURE §5.4; PRINCIPLES 1.4"
@@ -644,83 +626,69 @@ def simChecks (r : Runner) : Async Unit := do
   let answer : List Chunk := [⟨.success, digest, noise 1 70000⟩, ⟨.success, digest, noise 2 5⟩]
   for split in [false, true] do
     let tc ← TestClock.new
-    let (client, server) ← hostPair split
-    server.setHandler (handler m tc.clock 1000 fun _ => return answer)
-    server.serve
+    let p ← hostPair split
+    p.server.setHandler (handler m tc.clock 1000 fun _ => return answer)
     let limiter ← Limiter.new
     let got : OnceRef (List Chunk) ← OnceRef.new []
-    let res ← requestPeer client.newStream limiter m tc.clock 1000 (some (noise 3 200))
+    let res ← requestPeer (openStreamOf p) limiter m tc.clock 1000 (some (noise 3 200))
       fun c => got.update (· ++ [c])
     r.check s!"sim (split {split}): a request through the host gets its chunks"
       (succeedsWith () res && (← got.get).map (·.payload) == answer.map (·.payload))
-    r.check s!"sim (split {split}): the handler saw no failure" ((← server.failures.get).isEmpty)
+    r.check s!"sim (split {split}): the handler saw no failure" ((← p.failures.get).isEmpty)
     r.check s!"sim (split {split}): the limiter slot came back" ((← limiter.take m.id) && (← limiter.take m.id))
-    client.close
-    server.close
-    client.join
-    server.join
+    p.close
     tc.close
-  -- An id the peer does not serve: the stream is reset.
+  -- An id the peer does not serve: multistream-select answers `na`, so
+  -- the stream never opens for it.
   do
     let tc ← TestClock.new
-    let (client, server) ← hostPair
-    server.serve
-    let res ← requestPeer client.newStream (← Limiter.new) m tc.clock 1000 (some (noise 3 4))
+    let p ← hostPair
+    let res ← requestPeer (openStreamOf p) (← Limiter.new) m tc.clock 1000 (some (noise 3 4))
       fun _ => pure ()
-    r.check "sim: an id with no handler fails the request with a reset"
-      (failsWith (.stream .reset) res)
-    client.close
-    server.close
-    client.join
-    server.join
+    r.check "sim: an id with no handler fails the open"
+      (match res with | .error (.openFailed _) => true | _ => false)
+    p.close
     tc.close
   -- A silent peer over the sim: the deadline resets the stream, and the
   -- connection stays for the next exchange.
   do
     let tc ← TestClock.new
-    let (client, server) ← hostPair
+    let p ← hostPair
     -- The handler takes one read and ends with no answer and no close.
-    server.setHandler ⟨m.id, fun _ s => ExceptT.mk do
+    p.server.setHandler ⟨m.id, fun _ s => ExceptT.mk do
       discard <| s.read
       return .ok ()⟩
-    server.serve
     let owner ← TaskSet.new
     let res : OnceRef (Option (Except Error Unit)) ← OnceRef.new none
     owner.spawn do
-      res.set (some (← requestPeer client.newStream (← Limiter.new) m tc.clock 300 (some (noise 3 4))
+      res.set (some (← requestPeer (openStreamOf p) (← Limiter.new) m tc.clock 300 (some (noise 3 4))
         fun _ => pure ()))
     r.check "sim: a silent peer, the deadline arms" (← tc.nextTimer)
     tc.advance 300
     owner.joinAll
     r.check "sim: a silent peer fails the request at its deadline" ((← res.get).any (failsWith .timeout))
     tc.close
-    client.close
-    server.close
-    client.join
-    server.join
+    p.close
   -- The connection closes under a parked read: the request ends.
   do
     let tc ← TestClock.new
-    let (client, server) ← hostPair
+    let p ← hostPair
     let parked : Chan Unit ← Chan.new 1
-    server.setHandler ⟨m.id, fun _ _ => ExceptT.mk do
+    p.server.setHandler ⟨m.id, fun _ _ => ExceptT.mk do
       discard <| parked.send ()
       return .ok ()⟩
-    server.serve
     let owner ← TaskSet.new
     let res : OnceRef (Option (Except Error Unit)) ← OnceRef.new none
     owner.spawn do
-      res.set (some (← requestPeer client.newStream (← Limiter.new) m tc.clock 60000 (some (noise 3 4))
+      res.set (some (← requestPeer (openStreamOf p) (← Limiter.new) m tc.clock 60000 (some (noise 3 4))
         fun _ => pure ()))
     -- The handler ran, so the request is past its write and parks on the read.
     discard <| parked.recv
-    client.close
+    p.client.close
     owner.joinAll
     r.check "sim: a connection closed under a parked read ends the request"
       ((← res.get).any fun e => match e with | .error _ => true | .ok _ => false)
-    server.close
-    client.join
-    server.join
+    p.server.close
     tc.close
 
 def run (r : Runner) : IO Unit := do

packages/libp2p/host/Libp2p/Host.lean

@@ -5,6 +5,8 @@ import Libp2p.Host.Dedup
 import Libp2p.Host.Peerstore
 import Libp2p.Host.Policy
 import Libp2p.Host.Gate
+import Libp2p.Host.Upgrade
+import Libp2p.Host.StreamSelect
 import Libp2p.Host.Swarm
 import Libp2p.Host.Shutdown
 

packages/libp2p/host/Libp2p/Host/StreamSelect.lean

@@ -0,0 +1,82 @@
+import Streams
+import Bytes
+import Libp2p.Core
+import Libp2p.Negotiate
+
+/-!
+# Protocol selection on a muxed stream
+
+Every stream the muxer opens starts with its own multistream-select
+1.0.0 exchange (the connection spec, "Opening new streams over a
+multiplexed connection"): the initiator proposes the protocol ids in
+order, and the responder answers with the first one in its handler
+table. This module runs that exchange on a `Streams.Stream`.
+
+The negotiate package drives the same pure core over a
+`Streams.Connection` (`Negotiate.negotiate`), for the security and muxer
+stages of the upgrade. A muxed stream is a different seam: it reads and
+writes through `Except` answers, and it carries a read-back buffer for
+the leftover. The host owns this shell, and it reuses the core
+unchanged: `initiatorStart`, `responderStart`, `drive`, and
+`mapNegotiateError` all come from `Libp2p.Negotiate`.
+
+The bytes that follow the selection on the same read belong to the
+protocol. `Stream.prependBytes` hands them on to the handler
+(ARCHITECTURE §5.6).
+
+The read parks for as long as the peer stays silent, so no measure
+bounds the loop (root §3.3). The `repeat` form keeps the definition free
+of a `partial` marker, as in `Negotiate.negotiate`.
+-/
+
+set_option autoImplicit false
+
+namespace Libp2p.Host
+
+open Libp2p.Negotiate (initiatorStart responderStart drive mapNegotiateError)
+
+/-- Write each frame payload of the core to the stream, in order. A
+failed write is the `closed` arm: the stream is gone. -/
+private def sendWrites (stream : Streams.Stream) (ws : List (List Nat)) :
+    ExceptT Libp2p.Core.Error Conc.Async Unit := do
+  for w in ws do
+    if let .error e ← stream.write (Bytes.ofNatList w) then
+      throw (.closed s!"stream negotiation write failed: {repr e}")
+
+/-- Negotiate a protocol on a muxed stream. The answer is the selected
+id and the stream with the bytes after the selection put back in front
+of its reads. -/
+def negotiateStream (role : Libp2p.Core.Role) (stream : Streams.Stream)
+    (ids : List Libp2p.Core.ProtocolId) :
+    ExceptT Libp2p.Core.Error Conc.Async (Libp2p.Core.ProtocolId × Streams.Stream) := do
+  let start := match role with
+    | .initiator => initiatorStart ids
+    | .responder => responderStart ids
+  let (initial, first) ← match start with
+    | .error e => throw (mapNegotiateError e)
+    | .ok pair => pure pair
+  sendWrites stream first
+  let mut state := initial
+  let mut input : List Nat := []
+  repeat
+    match drive 1 state input with
+    | .error e => throw (mapNegotiateError e)
+    | .ok (.selected p _ rest writes) =>
+        sendWrites stream writes
+        return (p, ← Streams.Stream.prependBytes stream (Bytes.ofNatList rest))
+    | .ok (.running s _ rest writes) =>
+        sendWrites stream writes
+        state := s
+        input := rest
+    | .ok (.waiting s _ rest writes) =>
+        sendWrites stream writes
+        state := s
+        match ← stream.read with
+        | .error e => throw (.closed s!"stream negotiation read failed: {repr e}")
+        | .ok none => throw (.closed "stream ended during negotiation")
+        | .ok (some bytes) => input := rest ++ Bytes.toNatList bytes
+  -- Every arm above returns or loops, so the `repeat` never falls
+  -- through. The tail gives the `do` block its result type.
+  throw (.closed "stream negotiation ended without a selection")
+
+end Libp2p.Host

packages/libp2p/host/Libp2p/Host/Swarm.lean

@@ -3,8 +3,9 @@ import Conc
 import Streams
 import Libp2p.Identity
 import Libp2p.Core
-import Libp2p.Negotiate.Legacy
 import Libp2p.Host.Error
+import Libp2p.Host.StreamSelect
+import Libp2p.Host.Upgrade
 import Libp2p.Host.Registry
 import Libp2p.Host.Dedup
 import Libp2p.Host.Peerstore
@@ -36,7 +37,7 @@ ARCHITECTURE, "Refinements fixed at implementation (unit 1b)"):
 `allowDial` before a dial, `allowInbound` after an accept, the
 resource manager's `openConnection` before the upgrade, and
 `allowSecured` and `ConnectionScope.setPeer` inside the upgrade, after
-the handshake and before the muxer stage (`Negotiate.upgradeWith`). A
+the handshake and before the muxer stage (`Libp2p.Host.upgrade`). A
 refusal is `Error.gated` with the checkpoint.
 
 ## Ownership
@@ -69,7 +70,7 @@ set_option autoImplicit false
 
 namespace Libp2p.Host
 
-open Conc Streams Multiformats Libp2p.Identity Libp2p.Core Libp2p.Negotiate.Legacy
+open Conc Streams Multiformats Libp2p.Identity Libp2p.Core
 
 /-- A failure of a task that no caller awaits: an inbound upgrade, a
 session that ended badly, an inbound stream. -/
@@ -242,9 +243,9 @@ private def admit (h : Host) (role : Role) (scope : ConnectionScope) (conn : Con
 /-! ## The upgrade -/
 
 /-- Run the upgrade of `conn` as a pending connection: registered for
-the shutdown (§5.5), under the deadline of the upgrader, with the
-checkpoint. Every failure releases `scope`; the upgrader has already
-closed the connection. -/
+the shutdown (§5.5), under the deadline of the upgrade, with the
+checkpoint. Every failure releases `scope`. On a failure, the upgrade
+already closed the connection. -/
 private def upgradeConn (h : Host) (role : Role) (conn : Connection) (expected : Option PeerId)
     (scope : ConnectionScope) : Async (Except Error Upgraded) := do
   let key ← (h.draw : BaseIO Nat)
@@ -257,7 +258,7 @@ private def upgradeConn (h : Host) (role : Role) (conn : Connection) (expected :
     conn.closeQuietly
   let unregister : BaseIO Unit := h.pending.update (·.map (·.erase key))
   let r ← try
-      (upgradeAdmit h.config.clock role conn .empty expected (h.admit role scope conn)
+      (upgrade h.config.clock role conn expected (h.admit role scope conn)
         h.config.security h.config.muxers).run
     catch e =>
       unregister