fix(pubsub): return cancelled messages to the publisher's waiter

fix(pubsub): return cancelled messages to the publisher's waiter #5850