comms/uniflow/tcp: fail loudly on the paths that quietly did the wrong thing (#3921)

[cppccppccppc]

Summary:

Folded from two adjacent diffs; each half is stated separately below so the two
arguments stay reviewable on their own terms.

--- outbound: a failed operation must not reach the peer (was D117927060) ----

One invariant, enforced from the two ends it was missing: the peer must not do work for
an operation whose caller has already been told it failed. put() has held that line since
its pre-flight/commit split; these are the two places that did not. Folded together
because they are halves of the same argument and each is small -- the second is the one
that made the first insufficient.

--- Before the first frame is queued (was D117888775) --------------------------

get() looked up each request's remote handle in the loop that also admits and
queues frames, so a multi-request get() whose second request carries no TCP handle
failed after the first request's ReadRequests were already queued. The peer serviced
reads for an operation the caller had already been told had failed.

put() settles everything that can fail before its first frame is queued and says so
at the top of its pre-flight loop. get() did not hold that line. This moves the
lookup into the loop that already walks requests to count chunks, and carries the
resolved segIds forward, so a rejected get() leaves the peer untouched.

Not a correctness fix in the load-bearing sense and not a performance one. The caller
already saw the same InvalidArgument, and the destination buffer was never at risk:
fail() marks the op done, so a late ReadReply fails tryBeginWrite() and skips the
copy rather than writing into a buffer the caller may have released. What changed is
that the peer no longer does work for a rejected operation, and those inflight_ slots
are no longer held for a round trip.

No hot-path effect either way: findRemoteHandle() was already called once per request
rather than once per chunk, so this reorders the same work instead of removing any.
Benchmarked regardless, because the get path is measured rather than argued about.

--- After teardown has swept (was D117927060) ----------------------------------

The three admission points disagreed about connBroken_. admitInflight() and
recvImpl() re-test it under their container mutex, because the caller's entry check is
not enough on its own -- failAllPending() can land in the gap between that check and
the insert. The three enqueue paths listed connBroken_ only in their wait predicate, as
a wake condition, and after waking checked outClosed alone.

failAllPending() sets connBroken_ and clears every lane queue, but never sets outClosed:
the only writer is a sender that has died. handleFrame's exception containment sweeps
without closing the connection on purpose, so the reachable state is connBroken_ set,
reader stopped, connection open, sender alive indefinitely. A frame admitted just before
that sweep then lands in the just-cleared queue and the live sender transmits it.

For put and get that means a Write reaching the peer's segment for an operation whose
caller has already been resolved with ConnectionFailed -- a partial write at offsets
nobody is told about, which is the case put()'s pre-flight/commit split exists to
prevent. For send it means the promise completes successfully on a transport that has
failed everything else and stopped reading.

Now all three check connBroken_ alongside outClosed and route it down the path they
already had for refusal: enqueueFrame/enqueueFrames report false so the caller fails the
op, and enqueueSendFrame takes ownership of the promise and fails it.

senderLoop's own outClosed check is deliberately left alone. It has to keep draining
whatever is already queued; refusing there would abandon frames rather than admit them.

--- inbound: a mismatched reply must not complete an operation (was D117932250) ---

TcpInflight::isRead records whether a chunk came from get() or put(), and the
reply handler only consulted it on the ReadReply branch. The Ack branch did not, so an
Ack naming a get chunk fell through to completeOne(): the chunk resolved Ok with
entry.dst never written, and the caller read back whatever its destination buffer
already held.

That is the one peer-supplied dimension on this path that failed silently. segId,
offset, len and payload size are all checked and all produce an error; a crossed reply
kind produced success with wrong data.

Both directions are now settled in one place, immediately after the entry lookup, and
the check is exhaustive over the three ops that reach it: an Ack must name a write and
a ReadReply must name a read. Error stays exempt because it is kind-agnostic by design.

Consolidating also fixes a diagnostic that was wrong before: !entry.isRead used to
report "read reply size mismatch", so a skewed peer sending a ReadReply for a put chunk
sent the reader chasing a length bug that did not exist. That branch is now purely a
size check and its message is accurate.

Not reachable from a same-version peer -- Ack answers a Write, ReadReply answers a
ReadRequest, and reqIds are unique per chunk -- so this is version skew or a hostile
peer, the same bar as the oversized-ReadRequest check.

Reviewed By: rmahidhar

Differential Revision: D117927060

[meta-codesync]

@cppccppccppc has exported this pull request. If you are a Meta employee, you can view the originating Diff in D117927060.

[meta-codesync]

This pull request has been merged in 93f9e43.