Connect-RPC Bidirectional Stream Flow-Control Backpressure Fence

--- description: Connect-RPC Bidirectional Stream Flow-Control Backpressure Fence - Enforces deterministic coupling between transport-level HTTP/2 and HTTP/3 flow-control windows and server-side processing queues in Connect-RPC bidirectional streams. Disallows unbounded async dispatching during stream ingestion to prevent transport window updates from triggering catastrophic heap exhaustion. globs: src/**/services/**/*.ts, src/**/rpc/**/*.ts, src/**/handlers/**/*.ts alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/connect-rpc-bidi-stream-backpressure-fence" --- # Connect-RPC Bidirectional Stream Flow-Control Backpressure Fence > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unbounded Asynchronous Ingestion Draining Transport Frames Without Flow-Control Backpressure ## System Constraints You are a Principal Engineer enforcing Connect-RPC Bidirectional Stream Flow-Control Backpressure Fence. Mandate invariants that strictly bind HTTP/2 and HTTP/3 transport-level flow-control credit windows directly to asynchronous backend processing capacity, eliminating unmetered async iteration and heap starvation. ## Strict Guidelines - Engineers MUST ALWAYS await downstream consumption or acquire explicit bounded semaphore tokens before pulling the next message from a Connect-RPC `AsyncIterable` stream. - Engineers MUST NOT dispatch unawaited asynchronous tasks (`void doWork()`, floating promises) inside a `for await...of` loop over an incoming request stream. - Handlers MUST maintain an invariant high-watermark queue or bounded semaphore; incoming stream iteration MUST halt once in-flight operations reach threshold, halting HTTP/2 `WINDOW_UPDATE` frames. - Services MUST propagate cancellations and backpressure abort signals directly to the outgoing `AsyncIterable` yield channel and abort controllers to promptly close HTTP/2 frames. - Components NEVER store unprocessed streamed payloads into unbounded in-memory buffers or arrays before applying batch or rate-limited consumption. ## Bad vs Good Implementations ### Enforcing Bounded Ingestion Backpressure on Bidirectional Stream Handlers #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Iterating through `reqs` without awaiting downstream resolution drains the underlying HTTP/2 socket read buffer immediately. The Node.js HTTP/2 stack sends WINDOW_UPDATE frames prematurely, flooding the process heap with unprocessed payloads under heavy client load. ```typescript import { HandlerContext } from '@connectrpc/connect'; import { StreamRequest, StreamResponse } from './gen/service_pb.js'; export async function* bidirectionalStream( reqs: AsyncIterable<StreamRequest>, context: HandlerContext ): AsyncIterable<StreamResponse> { // ANTI-PATTERN: Iterating without awaiting task completion pushes HTTP/2 stream reader // to drain frames immediately, sending continuous WINDOW_UPDATE frames upstream and exhausting memory. for await (const req of reqs) { processItemAsync(req).then((res) => { // Floating promise resolves disconnected from flow control console.log(`Processed: ${res.id}`); }); } yield new StreamResponse({ status: 'COMPLETED' }); } ``` #### Production Standard (Correct) > **Standard**: Acquiring a bounded semaphore before consuming the next chunk suspends the `for await` loop when in-flight work reaches capacity. This keeps socket buffers full and pauses transport-level `WINDOW_UPDATE` frames, directly exerting backpressure on the client. ```typescript import { HandlerContext } from '@connectrpc/connect'; import { StreamRequest, StreamResponse } from './gen/service_pb.js'; class AsyncSemaphore { private permits: number; private waiters: Array<() => void> = []; constructor(permits: number) { this.permits = permits; } async acquire(): Promise<void> { if (this.permits > 0) { this.permits--; return; } await new Promise<void>((resolve) => this.waiters.push(resolve)); } release(): void { this.permits++; const next = this.waiters.shift(); if (next) { this.permits--; next(); } } } export async function* bidirectionalStream( reqs: AsyncIterable<StreamRequest>, context: HandlerContext ): AsyncIterable<StreamResponse> { const MAX_CONCURRENT_OPERATIONS = 16; const semaphore = new AsyncSemaphore(MAX_CONCURRENT_OPERATIONS); const responseQueue: StreamResponse[] = []; for await (const req of reqs) { if (context.signal.aborted) break; // INVARIANT: Halt consumption of transport DATA frames when buffer reaches max concurrency. // Suspends async iterator, suppressing transport-level WINDOW_UPDATE frames. await semaphore.acquire(); const promise = processItemAsync(req) .then((res) => { responseQueue.push(new StreamResponse({ id: res.id, status: 'OK' })); }) .finally(() => { semaphore.release(); }); while (responseQueue.length > 0) { yield responseQueue.shift()!; } } } ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: Connect-RPC Bidirectional Stream Flow-Control Backpressure Fence

Connect-RPC Bidirectional Stream Flow-Control Backpressure Fence defines an industrial-grade behavioral standard for backend engineering. Enforces deterministic coupling between transport-level HTTP/2 and HTTP/3 flow-control windows and server-side processing queues in Connect-RPC bidirectional streams. Disallows unbounded async dispatching during stream ingestion to prevent transport window updates from triggering catastrophic heap exhaustion. Modern software architectures require strict adherence to deterministic patterns to guarantee high availability, maintainability, and zero security vulnerabilities.

Framework Paradigm Shift & Drift MechanicsLLMs are trained on millions of open-source repositories containing outdated code, legacy idioms, and unpatched antipatterns. When coding assistants generate implementations for this stack, they default to common statistical tokens: Unbounded Asynchronous Ingestion Draining Transport Frames Without Flow-Control Backpressure Without an explicit system instruction file, developers spend hours debugging silent failures.

Failure Modes & Production Blast Radius

Hazard Analysis: What Breaks Without This Rule

The following failure modes have been cataloged from hundreds of hours of production code reviews where AI assistants produced non-compliant implementations:

Failure Mode	AI Symptom (The Anti-Pattern)	Severity	Production Impact & Remedy
AI Hallucination & Deprecation Drift	Unbounded Asynchronous Ingestion Draining Transport Frames Without Flow-Control Backpressure	Critical	Violating this industrial standard introduces regression bugs, security risks, and unexpected runtime failures in production. Enforce explicit behavioral system directives across all AI coding assistants in your repository.
Silent Architecture Inconsistency	AI assistants produce mixed paradigms across files, creating fragmented codebase conventions.	Medium	Increases cognitive load during code reviews and leads to maintenance debt across engineering teams. Pin project conventions in .cursor/rules, CLAUDE.md, and AGENTS.md so all team agents follow identical guardrails.

Architectural Directives in Detail

Step-by-Step Production Directives

Each rule constraint below is injected into the AI agent's system prompt to establish deterministic boundaries:

Engineers MUST ALWAYS await downstream consumption or acquire explicit bounded semaphore tokens before pulling the next message from a Connect-RPC `AsyncIterable` stream.

Rationale:AI models frequently generate unconstrained code here because statistical language distributions lack project-specific architectural context. Mandating this directive ensures that generated code adheres to verified industrial production requirements.