Add Lookahead, the dual to Pipelined

[nfrisby]

This PR supersedes PR #92. Same key idea: be able to receive (pipelined) messages possibly before sending response when you have the agency. But with better names, types, UX, etc.

The key semantic change was explained here: #92 (comment)

I've converted back to Draft because I think I realized why YieldPipelined is the way it is (ie why it "unnecessarily" couples "one Receiver with one send"): so that the Pipelining n is counting the number of pipelined messages. Because a principal concept when pipelining is the depth, so it's nice and helpful and expected for the current depth to equal n (or, well, n can be greater than the current depth but only if you haven't tried Collecting).

I'm now thinking through whether we want that same thing to be true on the DualPipelined n server: should it count (subject to DualCollect calls), the number of pipelined messages that the server has read? IE the number of received messages the server still owes a response to (where "response" might involve sending 1 or more messages).

#92 (comment)

[nfrisby]

With the new interface, the motivating use case (originally listed here #92 (comment)) now looks as follows:

  leiosNotifyServerPeerLookahead ::
    forall m point announcement vote.
    MonadThrow m =>
    m Bool ->
    -- ^ increments the number of outstanding requests
    m (Message (LeiosNotify point announcement vote) StBusy StIdle) ->
    -- ^ blocks until the next reply (announcement\/offer\/vote) is ready
    PeerLookaheadFixedSender (LeiosNotify point announcement vote) AsServer StIdle m ()
  leiosNotifyServerPeerLookahead incr next =
      PeerLookaheadFixedSender responder start
    where
      responder :: Sender (LeiosNotify point announcement vote) AsServer VariableSender StBusy StIdle m
      responder = SenderEffect $ next <&> \msg -> SenderYield ReflServerAgency msg SenderDone

      -- StIdle with nothing outstanding: receive the first request (or done).
      -- We must use a plain Await here: AwaitLookahead defers the StBusy->StIdle
      -- send, so it is only usable once we hold a request (i.e. are at StBusy).
      start ::
        Peer (LeiosNotify point announcement vote) AsServer (Lookahead Z (FixedSender StBusy StIdle)) StIdle m ()
      start =
        Await ReflClientAgency $ \case
          MsgDone                         -> Done ReflNobodyAgency ()
          MsgLeiosNotificationRequestNext ->
            Effect $ do
              incr >>= \b -> when b $ throwIO MkExnLeiosNotifyExcessiveRequests
              pure $ busy Zero

      -- StBusy with @n@ outstanding: hand this reply off to the responder and
      -- look ahead to the next request (or done), in one step.
      busy :: forall n.
        Nat n ->
        Peer (LeiosNotify point announcement vote) AsServer (Lookahead n (FixedSender StBusy StIdle)) StBusy m ()
      busy n =
        AwaitLookahead TheSender ReflClientAgency $ \case
          MsgDone                         -> drain (Succ n)
          MsgLeiosNotificationRequestNext ->
            Effect $ do
              incr >>= \b -> when b $ throwIO MkExnLeiosNotifyExcessiveRequests
              pure $ busy (Succ n)

      -- on termination, flush the sends we've handed off, then Done.
      drain :: forall n.
        Nat n ->
        Peer (LeiosNotify point announcement vote) AsServer (Lookahead n (FixedSender StBusy StIdle)) StDone m ()
      drain = \case
        Zero   -> Done ReflNobodyAgency ()
        Succ j -> FlushSender Nothing (drain j)