Feature: closing the subscription stream without clearing the local inventory and flushing ACKs

[ide]

Is your feature request related to a problem? Please describe.

I am looking to gracefully shutdown a service (ex: during rolling deployments) by draining remaining Pub/Sub messages without accepting new ones. Currently, this library offers an async subscription.close() method, which (1) closes the subscription stream, (2) clears the inventory of locally buffered messages, and (3) waits to flush all ACKs/NACKs:

nodejs-pubsub/src/subscriber.ts

Lines 297 to 308 in f8203ef

	async close(): Promise<void> {
	if (!this.isOpen) {
	return;
	}
	this.isOpen = false;
	this._stream.destroy();
	this._inventory.clear();
	await this._waitForFlush();
	this.emit('close');

The issue with close() is that the local inventory (in the LeaseManager instance) is cleared, which means those messages won't get handled until their ack deadline expires and the Pub/Sub service redelivers the messages to a new subscriber.

Describe the solution you'd like

I would like to have an API to help with gracefully draining the Pub/Sub subscription. Here are two proposals:

1. New method -- subscription.pause(): pauses the underlying pull stream so that it doesn't fetch new messages from the Pub/Sub service. The caller of this API can then pause the subscription, handle the remaining messages until the inventory is drained (perhaps expose another event to learn when the inventory is empty), and then call close() to flush all the ACKs.
2. Make subscription.close() drain the inventory: calling close() would close the underlying pull stream, wait for the existing inventory to become empty instead of clearing it, and then wait for the ACKs to be flushed.

Describe alternatives you've considered
In the meantime, I have a separate leasing system that deduplicates messages across Pub/Sub subscriber instances. (This was necessary because Pub/Sub has at-least-once semantics.) I could reduce the ack deadline for messages to something short, and rely on the leasing system to guard against the higher likelihood of duplicate messages.

With the shorter ack deadline, calling close() will still cause messages in the orphaned inventory to wait until their ack deadlines expire, but the deadlines will be shorter, reducing the impact of this issue.

[callmehiphop]

@ide thanks for the detailed write up! I think I would be in favor of adding an alternative method to close that would wait for all the messages to be acked/nacked before releasing the lock on them.

@kamalaboulhosn @bcoe any thoughts here?

[meredithslota]

(Just shifting assignee, will leave to @feywind to determine path forward.)

[feywind]

I'd like to take a further look at this after we've got 2.x out the door.

[feywind]

Just bumping this as still in the queue.

[michaelwittig]

I was looking for the same problem today.

Our workaround for now is this:

this.subscription = this.topic.subscription(PUBSUB_SUBSCRIPTION_NAME, {
      flowControl: { maxMessages: 10 }, // tradeoff performance against number of "stuck" messages during deployments
    });

Helps us a bit to keep the number of locally queued messages low but it also says that pubsub might just ignore that value. So... :)

[feywind]

Moving this into our internal backlog, but feel free to keep adding comments if needed here.

[kaushlakers]

Any update on this? Is there any way to gracefully drain a subscription in the library now? Either of the solutions proposed here would be great.

[ervin-pactum]

I think it is great idea, indeed as user of library i would expect library that has some internal state, automatically listen and obey SIGTERM by at least preparing for shutdown and releasing any locally queued unprocessed messages cleanly, so that only message that was currently processed would end with NACK and go to incremental backoff.

[WesCossick]

A slightly hacky workaround to allow the inventory to drain without retrieving new Pub/Sub messages is to do the following:

subscription._subscriber._stream.resume = () => {};
subscription._subscriber._stream.pause();

This works by:

1. Replacing the Stream's .resume() function with a function that does nothing. This is necessary because there are several places that emit a 'free' event, which will trigger calling the .resume() function. But, to drain the inventory, we don't want this to happen.
2. Pausing the stream, which prevents new messages from being received.

Then, if needed, you can wait until the inventory is empty by checking its .size property:

while (subscription._subscriber._inventory._messages.size > 0) {
    await sleep(1000);
}