refactor(protocols): run req/resp over the seam types of libp2p/core

Tests/ReqResp.lean

@@ -1,5 +1,6 @@
 import Tests.Support
 import Libp2p.Protocols.ReqResp.Exchange
+import Libp2p.Protocols.Mock.Muxer
 import Libp2p.Protocols.Mock.HostStream
 import Multiformats.Multiaddr
 

packages/libp2p/protocols/Libp2p/Protocols.lean

@@ -1,4 +1,5 @@
 import Libp2p.Protocols.Constants
+import Libp2p.Protocols.StreamError
 import Libp2p.Protocols.Identity
 import Libp2p.Protocols.Identify
 import Libp2p.Protocols.Ping

packages/libp2p/protocols/Libp2p/Protocols/Constants.lean

@@ -1,3 +1,4 @@
+import Libp2p.Core.ProtocolId
 import Libp2p.Protocols.Mock.Core
 
 /-!
@@ -44,7 +45,9 @@ def pingSize : Nat := 32
 /-- The protocol version the responder writes, the spike's constant. -/
 def identifyProtocolVersion : String := "/ipfs/1.0.0"
 
-open Libp2p.Protocols.Mock.Core
+-- Req/resp takes its ids from `libp2p/core`. The identify and ping ids
+-- above stay on the mock until their swap (issue #55).
+open Libp2p.Core (ProtocolId)
 
 /-! ## Req/resp: the protocol id -/
 

packages/libp2p/protocols/Libp2p/Protocols/Mock/HostStream.lean

@@ -1,20 +1,20 @@
 import Conc
 import Streams
-import Libp2p.Protocols.Mock.Core
-import Libp2p.Protocols.Mock.Muxer
+import Libp2p.Core.Handler
+import Libp2p.Core.Mux
+import Libp2p.Protocols.StreamError
 
 /-!
 # Host stream mock
 
-`Libp2p.Host` (W5.6, issue #19) is not merged. The protocols docs have
-the req/resp client open its stream through the host, and the host pair
-names the inbound side, `setHandler` over the registry of §3.10. Its
-surface lists no outbound call yet; go-libp2p's is `NewStream`
-(`vendor/go-libp2p/p2p/host/basic/basic_host.go`, line 441), which
-opens a stream to the peer and negotiates one of the caller's ids. This
-mock fixes minimal signatures for both sides over one mock muxer
-session, as #45's peerstore mock does for the names the host docs
-freeze (RULES.md, "Working ahead of the tree").
+`Libp2p.Host` (W5.6) merged, but it still runs its upgrade through a
+mock of `libp2p/negotiate` (issue #70), so no host opens a session yet.
+The protocols docs have the req/resp client open its stream through
+the host, with `newStream`, and serve through `setHandler` over the
+registry of §3.10; go-libp2p's outbound call is `NewStream`
+(`vendor/go-libp2p/p2p/host/basic/basic_host.go`, line 441). This
+mock fixes minimal signatures for both sides over one muxer session of
+the real `MuxSession` seam (RULES.md, "Working ahead of the tree").
 
 ## What it leaves out
 
@@ -46,8 +46,7 @@ set_option autoImplicit false
 namespace Libp2p.Protocols.Mock.HostStream
 
 open Conc Streams
-open Libp2p.Protocols.Mock.Core (ProtocolId StreamContext StreamHandler)
-open Libp2p.Protocols.Mock.Muxer
+open Libp2p.Core (ProtocolId StreamContext StreamHandler MuxSession)
 
 /-- One end of a peer connection, with its handler registry. -/
 structure Host where
@@ -59,7 +58,7 @@ structure Host where
   /-- The registry, by protocol id; changed only through `update`. -/
   handlers : OnceRef (List StreamHandler)
   /-- The handler failures, newest first. -/
-  failures : OnceRef (List Core.Error)
+  failures : OnceRef (List Libp2p.Core.Error)
   /-- The tasks of the served streams and the serve loop. -/
   owner : TaskSet
 
@@ -78,14 +77,14 @@ def idHeader (id : ProtocolId) : Option ByteArray :=
 
 /-- Open a stream to the peer for the first id of `protocols`. -/
 def Host.newStream (h : Host) (protocols : List ProtocolId) :
-    Async (Except Core.Error (ProtocolId × Stream)) := do
+    Async (Except Libp2p.Core.Error (ProtocolId × Stream)) := do
   let some id := protocols.head? | return .error (.protocol "newStream: no protocol id")
   let some header := idHeader id | return .error (.protocol "newStream: an id over 255 bytes")
-  match ← h.session.openStream.run with
+  match ← (h.session.openStream id).run with
   | .error e => return .error e
   | .ok s =>
     match ← s.write header with
-    | .error e => s.reset; return .error (Core.mapStreamError e)
+    | .error e => s.reset; return .error (mapStreamError e)
     | .ok () => return .ok (id, s)
 
 /-- Read the id header from `s`: the id and the bytes after it. `none`
@@ -135,9 +134,11 @@ def Host.serve (h : Host) : BaseIO Unit := do
 /-- Close the session. -/
 def Host.close (h : Host) : IO Unit := h.session.close
 
-/-- Wait for the serve loop, its streams, and the session loop. -/
+/-- Wait for the serve loop, its streams, and the session loop. The
+session of `Mock.Muxer.muxer` always joins with `.ok`, so the answer
+goes unread. -/
 def Host.join (h : Host) : Async Unit := do
   h.owner.joinAll
-  h.session.done
+  discard h.session.join.run
 
 end Libp2p.Protocols.Mock.HostStream

packages/libp2p/protocols/Libp2p/Protocols/Mock/Muxer.lean

@@ -1,19 +1,17 @@
 import Std.Data.HashMap
 import Conc
 import Streams
-import Libp2p.Protocols.Mock.Core
+import Libp2p.Core.Mux
 
 /-!
 # Muxer mock
 
-`Libp2p.Core` (W5.2, PR #37) and the yamux `Muxer` value (W5.4) are not
-merged. This mock declares the seam shapes that root ARCHITECTURE §4.9
-and the core pair freeze: `Role`, `MuxSession` (`openStream`, `accept`,
-`close`, and the completion its owner joins), and `Muxer` with `open :
-Role → Connection → ByteArray → ExceptT Error Async MuxSession`. The
-value `muxer` runs over any `Connection`, so the req/resp suite runs over
-the sim transport (PRINCIPLES 1.4; RULES.md, "Working ahead of the
-tree").
+The muxer values of `libp2p/mux` took the real core seams while this
+branch ran (PR #84, issue #77 closed), so the swap of this mock is due
+now (issue #78). Until it lands, this module stands in with one value,
+`muxer`, of the real `Muxer` seam of `libp2p/core`. It runs over any
+`Connection`, so the req/resp suite runs over the sim transport
+(PRINCIPLES 1.4; RULES.md, "Working ahead of the tree").
 
 ## The wire of the mock
 
@@ -36,52 +34,25 @@ yamux window of W5.4 is what bounds them in production.
 
 ## Tasks and parks
 
-One reader task per session, in a `TaskSet` the session owns; `done`
+One reader task per session, in a `TaskSet` the session owns; `join`
 joins it. The reader parks on `Connection.read`, which `close` wakes.
 Writers take a one-token channel around each frame write, so two
 streams never interleave the bytes of their frames; the token's park
 wakes when the writer before it sends the token back.
 
 ## Integration
 
-Delete this module. Import `Libp2p.Core` for `Role`, `MuxSession`, and
-`Muxer`, and use `Mux.yamux` (W5.4) where the suite takes `muxer`. The
-suite needs no other change: it uses the seam fields only.
+Delete this module, and use `Mux.test` or `Mux.yamux` where the suite
+takes `muxer`. The suite needs no other change: it uses the seam fields
+only.
 -/
 
 set_option autoImplicit false
 
 namespace Libp2p.Protocols.Mock.Muxer
 
 open Conc Streams
-open Libp2p.Protocols.Mock.Core (ProtocolId)
-
-/-- Which end of the connection a session serves. -/
-inductive Role where
-  /-- The end that dialed. -/
-  | initiator
-  /-- The end that accepted. -/
-  | responder
-  deriving DecidableEq, Repr
-
-/-- The uniform session view of the core seam. -/
-structure MuxSession where
-  /-- Open a stream to the peer. -/
-  openStream : ExceptT Core.Error Async Stream
-  /-- The next stream the peer opened; `none` after the session closed. -/
-  accept : Async (Option Stream)
-  /-- Close the session and its connection. -/
-  close : IO Unit
-  /-- The completion of the session loop, which the owner joins. -/
-  done : Async Unit
-
-/-- The muxer seam of the core. -/
-structure Muxer where
-  /-- The protocol id of the muxer. -/
-  id : ProtocolId
-  /-- Run a session over a connection. The `ByteArray` holds what the
-  negotiation read past its last message. -/
-  «open» : Role → Connection → ByteArray → ExceptT Core.Error Async MuxSession
+open Libp2p.Core (Role MuxSession Muxer)
 
 /-! ## Frames -/
 
@@ -219,7 +190,7 @@ partial def readLoop (sh : Shared) (peerParity : Nat) (buf : ByteArray) : Async
 
 /-- Run a mock session over `conn`. -/
 def openSession (role : Role) (conn : Connection) (leftover : ByteArray) :
-    ExceptT Core.Error Async MuxSession := do
+    ExceptT Libp2p.Core.Error Async MuxSession := do
   let token : Chan Unit ← Chan.new 1
   discard <| token.trySend ()
   let sh : Shared := {
@@ -232,14 +203,18 @@ def openSession (role : Role) (conn : Connection) (leftover : ByteArray) :
   let peerParity := if role = .initiator then 0 else 1
   owner.spawn (readLoop sh peerParity leftover)
   return {
-    openStream := do
+    -- The mock wire carries no protocol label, so the id goes unread,
+    -- as yamux drops it (the core `MuxSession.openStream` doc).
+    openStream := fun _ => do
       let sid ← (sh.next.updateGet fun n => (n, n + 2) : BaseIO Nat)
       let e ← newEntry
       sh.streams.update (·.insert sid e)
       mkStream sh sid e
     accept := sh.incoming.recv
     close := sh.conn.close
-    done := owner.joinAll }
+    -- The reader loop has no failure of its own, so the join answers
+    -- `.ok` once the loop ends.
+    join := ExceptT.lift owner.joinAll }
 
 /-- The mock muxer value. -/
 def muxer : Muxer := ⟨⟨"/leanp2p-mock-mux/1.0.0"⟩, openSession⟩

packages/libp2p/protocols/Libp2p/Protocols/ReqResp/Exchange.lean

@@ -1,6 +1,7 @@
 import Conc
 import Streams
-import Libp2p.Protocols.Mock.Core
+import Libp2p.Core.Handler
+import Libp2p.Protocols.StreamError
 import Libp2p.Protocols.ReqResp.Core
 import Libp2p.Protocols.ReqResp.Limiter
 
@@ -60,6 +61,9 @@ namespace Libp2p.Protocols.ReqResp
 open Libp2p.Protocols
 open Conc (Async)
 
+-- Inside this namespace a bare `Core` is the reader core,
+-- `ReqResp.Core`, so the seam types keep the full `Libp2p.Core` prefix.
+
 /-- Hand out the chunks of a list of actions. -/
 def deliverAll (deliver : Chunk → Async Unit) : List Core.Action → Async Unit
   | [] => pure ()
@@ -130,7 +134,8 @@ def request (s : Streams.Stream) (m : Method) (clock : Conc.Clock) (timeoutMs :
 /-- The host's call that opens a stream to one peer and negotiates one
 of the ids, as go-libp2p's `NewStream` does. The caller binds the peer. -/
 abbrev OpenStream :=
-  List Mock.Core.ProtocolId → Async (Except Mock.Core.Error (Mock.Core.ProtocolId × Streams.Stream))
+  List Libp2p.Core.ProtocolId →
+    Async (Except Libp2p.Core.Error (Libp2p.Core.ProtocolId × Streams.Stream))
 
 /-- Send one request to a peer: take a slot of the peer's `limiter`,
 open the stream for `m.id` through the host, and run `request` on it.
@@ -200,15 +205,15 @@ def serve (s : Streams.Stream) (m : Method) (clock : Conc.Clock) (timeoutMs : Na
     writeAnswer s m chunks
 
 /-- The one map from a req/resp error into the coarse core error. -/
-def mapError : Error → Mock.Core.Error
-  | .stream e => Mock.Core.mapStreamError e
+def mapError : Error → Libp2p.Core.Error
+  | .stream e => mapStreamError e
   | .timeout => .timeout "req/resp: the deadline expired"
   | .openFailed e => e
   | e => .protocol s!"req/resp: {repr e}"
 
 /-- The handler value the host registers under `m.id`. -/
 def handler (m : Method) (clock : Conc.Clock) (timeoutMs : Nat)
-    (respond : Option ByteArray → Async (List Chunk)) : Mock.Core.StreamHandler :=
+    (respond : Option ByteArray → Async (List Chunk)) : Libp2p.Core.StreamHandler :=
   ⟨m.id, fun _ s => ExceptT.mk do return (← serve s m clock timeoutMs respond).mapError mapError⟩
 
 end Libp2p.Protocols.ReqResp

packages/libp2p/protocols/Libp2p/Protocols/ReqResp/Limiter.lean

@@ -1,4 +1,5 @@
 import Conc
+import Libp2p.Core.ProtocolId
 import Libp2p.Protocols.Constants
 
 /-!
@@ -23,7 +24,7 @@ set_option autoImplicit false
 namespace Libp2p.Protocols.ReqResp
 
 open Libp2p.Protocols
-open Libp2p.Protocols.Mock.Core (ProtocolId)
+open Libp2p.Core (ProtocolId)
 
 /-- The open requests per protocol id. An id with no entry has none. -/
 abbrev Table := List (ProtocolId × Nat)

packages/libp2p/protocols/Libp2p/Protocols/ReqResp/Payload.lean

@@ -1,5 +1,7 @@
 import Varint
 import Snappy
+import Libp2p.Core.Error
+import Libp2p.Core.ProtocolId
 import Libp2p.Protocols.ReqResp.Crc32c
 import Libp2p.Protocols.ReqResp.SnappyFrame
 
@@ -102,7 +104,7 @@ inductive Error where
   the peer, so the spec forbids one more. -/
   | tooManyRequests
   /-- The host could not open the stream or negotiate the id. -/
-  | openFailed (e : Mock.Core.Error)
+  | openFailed (e : Libp2p.Core.Error)
   deriving DecidableEq, Repr, BEq
 
 /-- The `result` byte of a response chunk (phase0 "Responding side"). -/
@@ -204,7 +206,7 @@ def Bounds.admits (b : Bounds) (n : Nat) : Bool :=
 package frames the bytes; the method says how many. -/
 structure Method where
   /-- The protocol id, from `Constants.reqRespId`. -/
-  id : Mock.Core.ProtocolId
+  id : Libp2p.Core.ProtocolId
   /-- The bounds of the request payload. `none` for a method with "No
   Request Content" (GetMetaData): the request stream carries no byte. -/
   request : Option Bounds

packages/libp2p/protocols/Libp2p/Protocols/StreamError.lean

@@ -0,0 +1,28 @@
+import Streams
+import Libp2p.Core.Error
+
+/-!
+# The stream error at the handler seam
+
+A handler of this package reads and writes one negotiated `Stream`, and
+a failed read or write answers a `Streams.Error`. The handler seam of
+`libp2p/core` fails with the coarse `Core.Error` (ARCHITECTURE §3.5),
+so each handler maps the stream error once, through this function.
+
+Identify and ping still type over `Libp2p.Protocols.Mock.Core`, which
+keeps its own copy of this map until their swap (issue #55).
+-/
+
+set_option autoImplicit false
+
+namespace Libp2p.Protocols
+
+/-- The one map from a stream error to the coarse core error. `closed`
+stays closed, `reset` is a protocol fault, and `timeout` stays
+timeout. -/
+def mapStreamError : Streams.Error → Libp2p.Core.Error
+  | .closed => .closed "the stream closed"
+  | .reset => .protocol "the peer reset the stream"
+  | .timeout => .timeout "the stream deadline expired"
+
+end Libp2p.Protocols

packages/libp2p/protocols/docs/ARCHITECTURE.md

@@ -26,7 +26,9 @@ group is §7, row 10). The spike has no req/resp or gossipsub code.
 | `Libp2p/Protocols/ReqResp/Core.lean` | the reader state machine of §3.3: `State`, `Event`, `Action`, and the total `step`; it reads one request or a run of response chunks, and the output so far, the budget spent, the chunk count, and the error/EOF distinction live in the state |
 | `Libp2p/Protocols/ReqResp/Limiter.lean` | `MAX_CONCURRENT_REQUESTS`: the pure table rule with its bound theorem, and the per-peer `Limiter` over `OnceRef` |
 | `Libp2p/Protocols/ReqResp/Exchange.lean` | the shell over the core: the requester `request` and `requestPeer`, the responder `serve`, the handler value the registry stores, and the deadlines |
-| `Libp2p/Protocols/Mock/{Core,Muxer,HostStream}.lean` | the mocks of the dependencies this package still stands in for, W5.2, W5.4, and W5.6 (RULES.md, "Working ahead of the tree"); the integration deletes them |
+| `Libp2p/Protocols/StreamError.lean` | `mapStreamError`, the one map from a stream error into `Core.Error` that the req/resp handler uses |
+| `Libp2p/Protocols/Mock/Core.lean` | the mock of W5.2 that identify and ping still type over, until their swap (issue #55) |
+| `Libp2p/Protocols/Mock/{Muxer,HostStream}.lean` | the test values that stand in for a muxer value of W5.4 and the host of W5.6, over the real core seams; the muxer swap is due (`libp2p/mux` took the real core in PR #84, issue #77 closed, the swap owes on issue #78), and the host swap waits for the host's negotiate swap (issue #70) |
 | `Libp2p/Protocols/Gossipsub/Rpc.lean` | the RPC message with its `ProtoMessage` instance (§3.2) |
 | `Libp2p/Protocols/Gossipsub/Core.lean` | `State`, `Event`, `Action`, `ProtocolError`, and the total `step` that holds mesh, scoring, and heartbeat (§3.3) |
 | `Libp2p/Protocols/Gossipsub/Node.lean` | the loop task over the core, and the publish and subscribe surface at the seam level |
@@ -94,8 +96,9 @@ package; §4.9 names no echo, so no module line covers it.
   stream to the peer and negotiating the id is the host's job (W5.6),
   as go-libp2p's `NewStream` does it
   (`vendor/go-libp2p/p2p/host/basic/basic_host.go`, line 441). The host
-  pair's surface does not list that call yet, so the suite runs against
-  a mock of it. `requestPeer` takes the call as an argument, with a slot
+  names it `newStream`. The host still upgrades through a mock of
+  `libp2p/negotiate` (issue #70), so the suite runs against a mock of
+  the call. `requestPeer` takes the call as an argument, with a slot
   of the peer's `Limiter`, and runs `request` on the stream the call
   returns. `request` needs no host type: it writes
   the request, half-closes, and hands each response chunk to a callback
@@ -261,13 +264,16 @@ Both spike programs call `Ping.ping`.
   forbids the request and a wait would park on another request's
   deadline. The slot goes back on every path. The responder's side of
   the rule, "MAY penalise", is scoring, and stays with the client.
-- **The suite runs over the sim through two mocks.** The mock muxer
-  fills the frozen `Muxer` and `MuxSession` shapes of §4.9 over a sim
+- **The suite runs over the sim through two mocks.** The mock muxer is
+  one value of the `Muxer` seam of `libp2p/core` over a sim
   `Connection`, and the host stream mock fixes minimal signatures for
   the outbound stream call (go-libp2p's `NewStream`,
   `vendor/go-libp2p/p2p/host/basic/basic_host.go`, line 441) and for
-  `setHandler`. The W5.4 and W5.6 integrations replace them, and the
-  suite then runs over yamux and the host.
+  `setHandler`. Both type over the real core. `libp2p/mux` landed its
+  values over the real core (PR #84, issue #77 closed), so the muxer
+  swap is due (issue #78). The host swap waits for the host's
+  negotiate swap (issue #70). The suite then runs over yamux and the
+  host.
 
 ## What the package must not depend on