[AsyncStreaming] Add support for final elements

[FranzBusch]

Motivation

Real transports deliver structured data alongside end-of-stream and need to fuse the last write with the close. HTTP trailers and gRPC status are the obvious cases. Neither can be expressed by an empty-buffer terminator, and both lose H2/H3/QUIC's DATA+END_STREAM coalescing without a fused call.

Modification

Adds FinalElement: ~Copyable = Void as a primary associated type on all four protocols. AsyncReader delivers it via a consuming FinalElement? closure parameter; CallerAsyncReader returns it. Both writers gain a consuming finish carrying the last chunk and the payload in one call.

forEachBuffer, collect, and the pipe variants thread the payload through. collect becomes consuming and gains a Void-final overload returning just the result.

Adds an "Alternatives considered" entry covering what would break without this: HTTP body, gRPC, and fused close on H2/H3/QUIC.

Result

The four protocols can back HTTP body, gRPC streaming, and similar shapes without giving up the fused close. Default Void keeps simple conformers unchanged; Never marks infinite streams. Custom and ~Copyable payloads work end to end.

[guoye-zhang]

Should the final element be delivered to the closure?

[FranzBusch]

I intend to change this API in a future PR to be spelled collect(into: &buffer) and that should return the FinalElement as well.

[FranzBusch]

There are a few tests where I had to drop the expectations since pipe is now consuming. However, I have added them back in a follow up PR #425. I needed a few more fundamental building blocks to properly write those tests.

[guoye-zhang]

We could simplify CallerAsyncWriter into a single method that takes a buffer and an optional final element, since the code to write the buffer can usually be shared between those methods.

[gjcairo]

LGTM overall just a few questions.

[gjcairo]

What's keeping us from doing this now? Also shouldn't it be a RangeReplaceableContainer?

[FranzBusch]

I want to keep the scope of the PR limited to one change at a time

[gjcairo]

Sure. Should it not be a Container though instead of a Collection? Out of curiosity, you don't have to update the comment.

[FranzBusch]

Ah yeah it should. My bad :D

[gjcairo]

If the second error in EitherError is Never, can't we call unwrap() and have the type system realise it only throws the first? Or is it not that smart that we need this?

[FranzBusch]

Sadly no. The problem is that would require extensions like this

extension EitherError where First == Never {
  public func unwrap() throws(Second) {
    switch self {
    case .first:
      fatalError()
    case .second(let second):
      throw second
    }
  }
}

extension EitherError where Second == Never {
  public func unwrap() throws(First) {
    switch self {
    case .first(let first):
      throw first
    case .second:
      fatalError()
    }
  }
}

which result in ambiguity at the call site since you can't overload on the thrown type

[gjcairo]

This would allow empty buffers being received but not meaning end of stream, right? Is that a concern?

[FranzBusch]

Some protocols have it as valid to send empty buffers for liveliness probes so this would be a supported pattern.

[gjcairo]

Nit: do we want to annotate forEachBuffer as @discardableResult?

[FranzBusch]

I wanted to see more adoption before doing that.

[FranzBusch]

We could simplify CallerAsyncWriter into a single method that takes a buffer and an optional final element, since the code to write the buffer can usually be shared between those methods.

I thought about this but I think this is worse for two reasons:

1. finish is consuming right now and makes it impossible to produce more writes after finishing
2. The ergonomics become a lot worse when always having an optional final element on write

[guoye-zhang]

1. finish is consuming right now and makes it impossible to produce more writes after finishing

Yeah not being consuming is unfortunate, and would necessitate a separate function.

2. The ergonomics become a lot worse when always having an optional final element on write

That's somewhat debatable. Suppose we are writing a middleware that transforms the body:

func write(buffer: Buffer) {
    let processed = // process body
    self.next.write(buffer: processed)
}

func finish(buffer: Buffer, finalElement: Void) {
    let processed = // process body
    self.next.finish(buffer: processed, finalElement: ())
}

Implementations have to either duplicate code or make both functions call a common helper