Review a real-time subscription lifecycle

Review exercise · TypeScript

Advanced45 min

Review brief

A real-time gateway is adding per-client broker subscriptions and forwarding tasks. Connections are long lived, individual clients may stop reading, and the process must shut down without retaining owned work. Review how capacity, queues, locks, disconnects, and shutdown interact.

Behavioral contract

At most 500 accepted or handshaking client sessions may exist. A capacity permit is owned from acquisition until that session's full cleanup completes.
Each client has an outbound queue bounded at 64 messages. A further message disconnects that slow client and triggers normal cleanup rather than growing memory or blocking other clients. Calling `queues.create()` without a capacity creates an unbounded queue; `push` returns `full` when a bounded queue has no remaining capacity.
The clients lock protects only map membership and snapshots. No network send, connection close, or task join may be awaited while holding it.
Orderly and abrupt disconnects both remove the client, unsubscribe its broker callback, close its queue, cancel if needed, and join its forwarding task before releasing capacity.
Shutdown rejects new sessions, closes all current sessions, and cancels and joins all forwarding work within five seconds.
Message order is preserved within one client, and one slow or failed client must not delay delivery or cleanup for another client.
`ClientConnection` safely serializes concurrent `send` calls internally; transport write serialization itself is out of scope.

Review instructions

1Review the changed gateway against the capacity, backpressure, lock, disconnect, and shutdown contract.
2Trace ownership from handshake through both orderly and abrupt termination, including background forwarding work.
3Prioritize no more than six findings, including test feedback only when lifecycle behavior is unprotected.

Learning objectives

• Tie permits and background work to connection lifetime
• Identify backpressure and lock-scope risks
• Verify cleanup for disconnect and shutdown

src/subscription-gateway-types.tsTypeScript

@@ -0,0 +1,37 @@

1+export interface RealtimeMessage {

2+ topic: string;

3+ data: string;

4+}

6+export interface ClientConnection {

7+ id: string;

8+ accept(): Promise<void>;

9+ send(message: RealtimeMessage): Promise<void>;

10+ close(reason: "server-shutdown" | "slow-client"): Promise<void>;

11+ closed: Promise<{ orderly: boolean }>;

12+}

13+

14+export interface MessageQueue extends AsyncIterable<RealtimeMessage> {

15+ push(message: RealtimeMessage): "queued" | "full";

16+ close(): void;

17+}

18+

19+export interface Subscription {

20+ unsubscribe(): void;

21+}

22+

23+export interface SubscriptionGatewayDependencies {

24+ capacity: {

25+ acquire(): Promise<{ release(): void }>;

26+ };

27+ broker: {

28+ subscribe(topics: string[], onMessage: (message: RealtimeMessage) => void): Subscription;

29+ };

30+ queues: {

31+ create(capacity?: number): MessageQueue;

32+ };

33+ clientsLock: {

34+ runExclusive<T>(work: () => Promise<T> | T): Promise<T>;

35+ };

36+ shutdownTimeoutMilliseconds: number;

37+}

Reference solution

The findings below are ordered by priority.

Findings

1. Session capacity is released immediately after the handshake

Location: src/subscription-gateway.ts:23-56

The acquired permit is released as soon as the connection is accepted and registered. Long-lived sessions therefore consume no capacity, so the gateway can exceed its 500-session limit even though every handshake individually acquired a permit.

Hold the permit for the entire session lifetime and release it exactly once in the common cleanup path after the client is removed and its forwarding resources are joined.

2. Every client receives an unbounded outbound queue

Location: src/subscription-gateway-types.ts:30-32 and src/subscription-gateway.ts:33-37

The queue factory treats an omitted capacity as unbounded, and connect calls it without one. Broker callbacks can therefore keep appending while a slow client blocks in send. The callback also ignores push's overflow result, so it cannot invoke the required slow-client cleanup even if the queue were bounded.

Create each queue with the required capacity of 64. When push returns full, close that connection with the slow-client reason and run its normal subscription, queue, forwarder, and permit cleanup without affecting others.

3. Broadcast holds the clients lock while waiting for network delivery

Location: src/subscription-gateway.ts:69-76

The exclusive lock remains held through every awaited connection.send. One slow client blocks registration, cleanup, shutdown snapshots, and delivery to every later matching client. Direct sends also bypass the per-client queue's ordering and overflow policy.

Take a stable snapshot of matching sessions under the lock, release it, and enqueue to each session outside the critical section. If an enqueue reports full, invoke that client's slow-client cleanup independently.

4. Abrupt disconnect or forwarding failure skips session cleanup

Location: src/subscription-gateway.ts:46,58-66,89-93

Removal, unsubscribe, queue closure, and forwarder joining occur only when close.orderly is true. An abrupt network loss therefore retains the session. A rejected forwarding task is also not observed while connect waits for connection.closed, allowing an unhandled rejection and retaining the same owned resources until some unrelated close.

Observe both transport closure and forwarding completion, initiate the appropriate connection close when forwarding fails, and run one idempotent cleanup path in a finally block for every outcome. Close the queue and consume the forwarder's result before releasing capacity.

5. Shutdown misses handshakes and does not join forwarding work

Location: src/subscription-gateway.ts:23-55,79-93

The shutdown snapshot contains only registered clients. A connection can pass the early stopping check, remain in accept, and register after that snapshot, escaping shutdown entirely. For captured sessions, shutdown only awaits transport close calls; it never closes queues, unsubscribes, joins forwarding work, or enforces shutdownTimeoutMilliseconds.

Track handshakes from permit acquisition, recheck shutdown before registration, and include accepted or handshaking sessions in the shutdown join. Initiate the common cleanup for every owned session and await connection closure and every forwarder within the five-second deadline.

6. The test covers only an already-orderly close

Location: src/subscription-gateway.test.ts:17-49

The connection's close promise and empty queue complete immediately. The test cannot detect early permit release, queue overflow, a slow broadcast under lock, abrupt disconnect leaks, or shutdown with a blocked forwarder.

Add controlled tests that keep sessions open, fill a 64-entry queue, block one client send while another progresses, resolve an abrupt close, and hold forwarding work across shutdown and its deadline.

Reasonable non-findings

The handshake failure path unsubscribes, closes the queue, and releases its permit before rethrowing.
Rechecking stopping after acquiring capacity correctly prevents a waiter from becoming a new session during shutdown.
Filtering broadcasts by subscribed topic is required and correctly implemented.
Sequential sends within one client's forwarding loop preserve that client's message order.
The transport safely serializes concurrent send calls by contract; the issue is waiting for sends under the shared clients lock.