Streaming API must accept more than one resource path to efficiently read CSV files when more than one channel from that CSV is requested

[Apollo3zehn]

[Apollo3zehn]

API-Design: Multi-Channel-Streaming über HTTP/2

Ziel

Effizientes Abfragen mehrerer Datenkanäle (z. B. 30 Spalten aus einer CSV) in
einem logischen Request, ohne dass:

• die zugrunde liegende Datei mehrfach gelesen wird,
• Server oder Client GB-weise Daten zwischenpuffern müssen.

Das Backend verfügt bereits über die Fähigkeit, mehrere angefragte Kanäle
gleichzeitig aus denselben Quelldateien zu befüllen, unter Verwendung von
System.IO.Pipelines (Pipe, PipeReader/PipeWriter) mit Backpressure:
Ist eine Pipe voll, wartet der schreibende Read-Prozess, bis der Consumer sie
geleert hat.

Architekturidee

Backpressure wird über zwei Ebenen durchgereicht:

Datei-Read → Pipe (Backend-intern) → HTTP/2 Flow-Control-Window → Client

HTTP/2 hat pro logischem Response-Stream ein eigenes Flow-Control-Window.
Liest der Client eine Response nicht aus, blockiert der Server beim Schreiben
nur für diesen einen Stream, alle anderen Streams derselben Connection laufen
unabhängig weiter. Das ist strukturell identisch zum vorhandenen
Pipe-Backpressure-Mechanismus, nur eine Ebene höher — beide werden einfach
verkettet:

await pipeReader.CopyToAsync(Response.BodyWriter, cancellationToken);

Kann Kestrel wegen HTTP/2-Flow-Control nicht weiterschreiben, hängt
CopyToAsync, der PipeReader wird nicht weiter advanced, die Pipe bleibt
voll, und der Backend-seitige Multi-Read blockiert exakt so, wie er es heute
schon für einen vollen Consumer tut.

Ablauf

1. Batch-Request (ein Call, registriert die Session)

public record BatchRequest(IReadOnlyList<string> ResourcePaths, DateTime Begin, DateTime End);
public record BatchResponse(Guid SessionId, IReadOnlyList<ChannelRef> Channels);
public record ChannelRef(Guid ChannelId, string ResourcePath);

[HttpPost("streams/batch")]
public ActionResult<BatchResponse> CreateBatch([FromBody] BatchRequest request)
{
    var session = _streamSessionManager.CreateSession(
        request.ResourcePaths,
        request.Begin.ToUniversalTime(),
        request.End.ToUniversalTime());

    return Ok(new BatchResponse(session.Id, session.Channels));
}

_streamSessionManager legt pro angefragtem Kanal eine Pipe-Instanz an und
hält sie unter der SessionId vor. Der eigentliche Multi-Read startet erst,
sobald alle erwarteten Channel-Requests (Schritt 2) eingetroffen sind — die
genaue Anzahl ist durch die BatchResponse bekannt, kein Timing-Window nötig.

2. Channel-Requests (N parallele GET-Requests, gleiche HTTP/2-Connection)

[HttpGet("streams/batch/{sessionId}/channel/{channelId}")]
public async Task<IActionResult> GetChannelAsync(
    Guid sessionId, Guid channelId, CancellationToken cancellationToken)
{
    var pipeReader = _streamSessionManager.GetPipeReader(sessionId, channelId);

    Response.ContentType = "application/octet-stream";
    await pipeReader.CopyToAsync(Response.BodyWriter, cancellationToken);
    return new EmptyResult();
}

Jeder Kanal ist eine eigenständige HTTP-Response. Ein Fehler in Kanal 5 von 30
(z. B. NotFound, Validation, Forbidden) betrifft nur diesen einen
Request/Response, die übrigen 29 laufen unbeeinflusst weiter.

3. Client: paralleles Lesen ist zwingend

var batch = await httpClient.PostAsJsonAsync("streams/batch", new BatchRequest(paths, begin, end));
var batchResult = await batch.Content.ReadFromJsonAsync<BatchResponse>();

var tasks = batchResult.Channels.Select(async channel =>
{
    using var response = await httpClient.GetAsync(
        $"streams/batch/{batchResult.SessionId}/channel/{channel.ChannelId}",
        HttpCompletionOption.ResponseHeadersRead,
        cancellationToken);

    await using var stream = await response.Content.ReadAsStreamAsync(cancellationToken);
    await ProcessChannelDataAsync(channel.ResourcePath, stream, cancellationToken);
});

await Task.WhenAll(tasks);

Werden die Channel-Requests nicht gleichzeitig ausgelesen (z. B. sequenziell
awaited statt mit Task.WhenAll), blockiert das Backend am vollen Kanal und
das Gesamtsystem verliert seinen Geschwindigkeitsvorteil.

Offene Punkte

• Session-Timeout / Cleanup: Ruft ein Client einen registrierten Channel
  nie ab oder bricht ab, muss die zugehörige Session inkl. Pipes und
  Multi-Read nach einem Timeout automatisch verworfen werden, sonst blockiert
  ein toter Consumer die übrigen Kanäle derselben Session dauerhaft.
  CancellationToken pro Session ist Pflicht.
• HTTP/1.1-Fallback: Das Modell setzt echtes HTTP/2-Multiplexing voraus.
  Unter HTTP/1.1 (typischerweise ~6 parallele Connections pro Host) verliert
  der Ansatz seinen Vorteil bzw. verhält sich schlechter als N unabhängige
  Requests. Muss geklärt werden, ob HTTP/1.1-Clients/Proxys im Zielumfeld
  ausgeschlossen werden können.
• Session-Storage: ConcurrentDictionary als einfachster Startpunkt für
  _streamSessionManager; bei Bedarf (Skalierung über mehrere Instanzen,
  Persistenz) durch robusteren Mechanismus ersetzbar.
• Fehlerformat pro Kanal: Statuscode-Mapping (404/403/422) pro
  Channel-Request beibehalten, analog zum bisherigen Single-Channel-Endpoint.