Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@src/valkey/valkey.zig`:
- Around line 673-690: The code currently uses a locally-computed subscription
count via p._subscription_ctx.channelsSubscribedToCount(globalThis) before
resolving the SUBSCRIBE promise; change this to use the server-reported count in
the response by extracting frame.data[1] (via frame.data[1].asInteger()) and
pass that value into req_pair.promise.promise.resolve(...) so that
addSubscription(), onValkeySubscribe(), and the promise resolution use the
server's authoritative count for multi-channel SUBSCRIBE responses.

In `@test/regression/issue/29042.test.ts`:
- Around line 47-60: The test awaits (the Promise created via
Promise.withResolvers and other async assertions around sub.subscribe /
pub.send) can hang; wrap those awaits with a bounded timeout helper (e.g.,
withTimeout) so failures are immediate and diagnostic. Replace direct awaits of
promise/received (and the other places flagged: the other awaits around pub.send
/ subscriber resolution) with await withTimeout(promiseOrOperation, TIMEOUT_MS)
and keep using Promise.withResolvers, resolve, reject inside sub.subscribe to
trigger the resolver; ensure you apply this pattern to the other two spots
referenced (the awaits at the other subscribe/publish assertions).

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@test/regression/issue/29042.test.ts`:
- Around line 57-89: The test leaks Redis sockets because pub.close() and
sub.close() only run on the happy path; change the setup to use scoped disposal
so clients are always closed even on assertion/timeout failures: create the
clients with the Disposable pattern (use await using or using) around
RedisClient and its connect() result (e.g., replace manual const pub = new
RedisClient(...); await pub.connect(); ... pub.close() with an await using
(const pub = await RedisClient.connect("redis://localhost:6379") or equivalent)
and similarly for sub; ensure the test body uses the scoped variables (sub, pub)
and remove the explicit close() calls so close() runs automatically on scope
exit.

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@test/regression/issue/29042.test.ts`:
- Around line 19-28: The helper redisReachable creates a RedisClient (variable
c) but only calls c.close() on the success path so a thrown connect() leaves a
stray connection; change the function redisReachable to ensure the client is
always closed by moving c.close() into a finally block (guarding for c being
defined) and await the close if it returns a promise so the connection is
cleaned up regardless of whether connect() throws.

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@src/valkey/valkey.zig`:
- Around line 844-847: The branch inside valkey.zig that checks if (value.* ==
.Error) currently calls this.fail(value.Error,
protocol.RedisError.InvalidResponse) after the subscriber request pair may have
already been popped from in_flight, which causes global rejection of remaining
queues and leaves the original caller unresolved; instead, detect whether
pair_maybe was removed and if so reject only the specific promise tied to that
subscriber (or return the appropriate error to the caller) rather than invoking
this.fail; update the logic around pair_maybe/in_flight handling so you either:
1) resolve/reject the popped pair directly when value.* == .Error, or 2) if the
pair is already removed, propagate the error only to that subscriber's promise,
and only call this.fail when the error should close the whole client.
- Around line 801-810: Restore the guard that prevents unsupported RESP3 Push
frames from entering the normal reply queue: before calling readItem() check if
the incoming value is a RESP3 Push (value.kind == .Push) and that
protocol.SubscriptionPushMessage.asFrame(value) returns null; if so, treat it as
an unsupported push (route to the subscription/push handler or drop/log it)
instead of letting it fall through into the usual reply handling/promise pair
consumption. Ensure this same guard is applied in the other affected block
mentioned (the code region around the existing SubscriptionPushMessage.asFrame
usage and the similar code at the other spot around lines 834–856) so only
non-push RESP2 replies are allowed to proceed into the normal reply queue.

Verify each finding against the current code and only fix it if needed.

Outside diff comments:
In `@src/valkey/valkey.zig`:
- Around line 823-886: The code currently returns silently when pair_maybe is
null which causes unsupported RESP2 async Array notifications to be dropped;
change the path where pair_maybe orelse { return; } to surface an error when
we're in subscriber context and received a non-subscribe async Array with no
in-flight pair: instead of returning, call this.fail(...) with an appropriate
protocol.RedisError (e.g., InvalidResponse or a new UnsupportedAsyncNotification
error) passing the received value so the client is notified; reference the
symbols pair_maybe, subscription_frame, this.parent().isSubscriber(), value, and
this.fail to locate where to implement this explicit error handling.