fix(microservices): route message handler rejections to handleError

[hktitof]

PR Checklist

Please check if your PR fulfills the following requirements:

• The commit message follows our guidelines: https://github.com/nestjs/nest/blob/master/CONTRIBUTING.md
• Tests for the changes have been added (for bug fixes / features)
• Docs have been added / updated (for bug fixes / features)

PR Type

What kind of change does this PR introduce?

• Bugfix
• Feature
• Code style update (formatting, local variables)
• Refactoring (no functional changes, no api changes)
• Build related changes
• CI related changes
• Other... Please describe:

What is the current behavior?

Context: Sibling to #17737 (which routed the tcp and rmq handleMessage rejections to handleError bcs an unhandled promise rejection terminates the process). This addresses the mqtt, nats and redis servers, which hand the same async handler to their client libraries with no catch.

#17737 fixed the unhandled rejection on tcp and rmq, the same shape is still open on the other three transports:

• mqtt: mqttClient.on('message', this.getMessageHandler(mqttClient).bind(this)), an EventEmitter listener, the returned promise is never observed
• nats: client.subscribe(channel, { callback: ... }), nats-core calls sub.callback(null, msg) without awaiting it
• redis: subClient.on('message' | 'pmessage', this.getMessageHandler(pubClient).bind(this)), same EventEmitter story

handleMessage rejects on paths that get hit in normal operation: the deserializer await runs with nothing around it, and the handler await sits inside onProcessingStartHook whose default is (id, ctx, done) => done(), so a rejection propagates straight out of the base handleEvent too, meaning one malformed packet from the wire, or an event handler that rejects, is enough to kill the whole server process

Issue Number: N/A

Steps to reproduce

1. take the spec files from this PR, then run git checkout master -- packages/microservices/server/server-mqtt.ts packages/microservices/server/server-nats.ts packages/microservices/server/server-redis.ts and npx vitest run packages/microservices/test/server/server-mqtt.spec.ts packages/microservices/test/server/server-nats.spec.ts packages/microservices/test/server/server-redis.spec.ts -t "leaving them unhandled"
2. Expected: the registered handlers route a handleMessage rejection to handleError, like the tcp and rmq handlers already do since fix: absolute-form middleware bypass (fastify) and unserializable pattern crash (microservices) #17737
3. Actual (raw output on untouched master):

⎯⎯⎯⎯⎯⎯⎯ Failed Tests 3 ⎯⎯⎯⎯⎯⎯⎯

 FAIL  packages/microservices/test/server/server-mqtt.spec.ts > ServerMqtt > listen > should route "handleMessage" rejections to "handleError" instead of leaving them unhandled
Error: unexpected
 ❯ packages/microservices/test/server/server-mqtt.spec.ts:57:21
     55|     });
     56|     it('should route "handleMessage" rejections to "handleError" inste…
     57|       const error = new Error('unexpected');
     |                     ^
     58|       vi.spyOn(server, 'handleMessage').mockRejectedValue(error);
     59|       const handleErrorSpy = vi

 FAIL  packages/microservices/test/server/server-nats.spec.ts > ServerNats > getMessageHandler > handler > should route "handleMessage" rejections to "handleError" instead of leaving them unhandled
Error: unexpected
 ❯ packages/microservices/test/server/server-nats.spec.ts:170:23
    168|       });
    169|       it('should route "handleMessage" rejections to "handleError" ins…
    170|         const error = new Error('unexpected');
    |                       ^
    171|         vi.spyOn(server, 'handleMessage').mockRejectedValue(error);
    172|         const handleErrorSpy = vi

 FAIL  packages/microservices/test/server/server-redis.spec.ts > ServerRedis > handleConnection > should route "handleMessage" rejections to "handleError" instead of leaving them unhandled
Error: unexpected
 ❯ packages/microservices/test/server/server-redis.spec.ts:115:21
    113|     });
    114|     it('should route "handleMessage" rejections to "handleError" inste…
    115|       const error = new Error('unexpected');
    |                     ^
    116|       vi.spyOn(server, 'handleMessage').mockRejectedValue(error);
    117|       const handleErrorSpy = vi

 Test Files  3 failed (3)
      Tests  3 failed | 84 skipped (87)

What is the new behavior?

the mqtt, nats and redis message handlers route rejections through handleError exactly like tcp and rmq do since #17737, so a bad packet logs an error instead of crashing the process

Does this PR introduce a breaking change?

• Yes
• No

Other information

same bug class as #17737, just the transports it left out, kafka is deliberately not touched bcs kafkajs reports handler failures itself, same reason #17737 skipped it, successful messages behave exactly as before, only the paths that used to crash the process now log an error

[kamilmysliwiec]

lgtm