nit: What do you think of renaming "resumed" to "running"?

Possibly, but I definitely do not want to do that in this PR. That change is not trivial, and requires more design thought. Also note that this structure looks similar for now, but with the connection migration work in, it will be different because the processors will need to do different operations (e.g. setting various variables), though you can argue that some kind of callbacks may work here.

Let me think about the suggestion that you mentioned above. There was a reason why I did the current approach, but I don't recall anymore.

nit: The manual p.lock() and p.unlock() pairings are error prone.

Option 1:
Extract these into helpers. I'm thinking something like:
setResumed func() error
setInPeek func() (suspended bool, error)
setForwarding func() (suspended bool, error)

Option 2:
Use function blocks.

  err := func() error {
  p.lock()
  defer p.unlock()
  if p.mu.resumed {
	  // Already resumed - don't start another one.
	  return errProcessorResumed
  }
  p.mu.resumed = true
      return nil
  }()
  if err != nil {
      return err
  }

I don't think this is possible. The root problem is that SetReadDeadline is asynchronous:

// A deadline is an absolute time after which I/O operations
// fail instead of blocking. The deadline applies to all future
// and pending I/O, not just the immediately following call to
// Read or Write. After a deadline has been exceeded, the
// connection can be refreshed by setting a deadline in the future.

Without setting this, I see that we get races in two conditions:

1. PeekMsg exits with a deadline error when suspendReq = false
2. ForwardMsg exits with a deadline error when that's not the intention

If multiple resumption and suspension operations were called at the same time (as illustrated in one of the tests), both cases are possible. My understanding is that Go uses non-blocking reads under the hood, and it really depends on what the runtime is doing when SetReadDeadline is called.

Consider the following case (where we don't reset deadlines within transfer loop):

_, _, err := p.f.clientConn.PeekMsg()

p.lock()
suspend := p.mu.suspendReq
p.mu.suspendReq = false
p.mu.inPeek = false
p.unlock()
// Broadcast
... 
// ForwardMsg

When we signal the blocked wait calls, there's a possibility where a context switch has not occurred, and the control continues to ForwardMsg. In that case, ForwardMsg may be unblocked, but we don't want that to happen. This is just one case, but I've seen other possible cases, where we return from the initial PeekMsg with a timeout error, but suspendReq is false. If ForwardMsg gets unblocked when it has started forwarding, the connection is non-recoverable.

Hm, I guess doing this also means that we can remove waitUntilSuspended. I'll look into this, and see how this ties back to the connection migration.

nit: can we remove the error from enterPeek and exitPeek? They always return nil.

nit: this context check is unecessary