Unhandled promises in Topic.flush and PubSub.close prevent proper shutdown

Both methods Topic.flush and PubSub.close indicate that it is ok to call them with no parameter and await their completion. Unfortunately when passing no callback, the function internals don't properly await and instead have unhandled promises which can in certain cases lead to message not being sent correctly.

This could be a reason to the recently observed increase in Total timeout of API google.pubsub.v1.Publisher exceeded 60000 milliseconds before any response was received. errors on Google Cloud Functions and Cloud Run because those runtime instances assumingly switch to idle before all messages have been send.

Take a look at Topic.flush:

nodejs-pubsub/src/topic.ts

Lines 193 to 199 in 38fba8b

    
           flush(): Promise<void>; 
        
           flush(callback: EmptyCallback): void; 
        
           flush(callback?: EmptyCallback): Promise<void> | void { 
        
             // It doesn't matter here if callback is undefined; the Publisher 
        
             // flush() will handle it. 
        
             this.publisher.flush(callback!); 
        
           }

The call to this.publisher.flush(callback!); is not awaited nor returned so there is no way for the caller of Topic.flush to await the internal Promise completion without passing a callback.

Similar issue exists in PubSub.close.

After a little more inspection of the Topic.flush code path, I think the problem with unhandled promises is actually more wide-spread. Take a look at the following line of code from Publisher.flush:

nodejs-pubsub/src/publisher/index.ts

Lines 110 to 124 in 38fba8b

    
           flush(callback?: EmptyCallback): Promise<void> | void { 
        
             const definedCallback = callback ? callback : () => {}; 
        
             const publishes = [promisify(this.queue.publish).bind(this.queue)()]; 
        
             Array.from(this.orderedQueues.values()).forEach(q => 
        
               publishes.push(promisify(q.publish).bind(q)()) 
        
             ); 
        
             const allPublishes = Promise.all(publishes); 
        
             allPublishes 
        
               .then(() => { 
        
                 definedCallback(null); 
        
               }) 
        
               .catch(definedCallback); 
        
           }

In line 113 the internal Queue.publish method is called to publish the remaining messages. The callback from line 110 isn't passed down into Queue.publish. Instead Promise.all is used to wait for the queue to publish. But the Queue.publish method itself is void; it is not returning a Promise that can be awaited.

nodejs-pubsub/src/publisher/message-queues.ts

Lines 162 to 173 in 38fba8b

    
           publish(callback?: PublishDone): void { 
        
             const {messages, callbacks} = this.batch; 
        
             this.batch = new MessageBatch(this.batchOptions); 
        
             if (this.pending) { 
        
               clearTimeout(this.pending); 
        
               delete this.pending; 
        
             } 
        
             this._publish(messages, callbacks, callback); 
        
           }

With this, it seems currently impossible to reliably await the complete publication of all message with this library, no?

@cdupuis

@cdupuis Thanks for the report. That does look wrong at first glance, but I wonder if this code was relying on promisify(), and I missed updating it or something.

I'm moving this to our internal backlog to look at along with other unhandled promise and graceful shutdown issues.

Re-opening this upon request from TSEs.

This got some changes from the exactly once delivery work, and it looks like the current unit tests should cover it, so I think this is finished. Please comment if you're still having the issue.

feat: Bigtable authorized views requests on the Data plane (googleapi…

Navigation Menu

Repository navigation

Unhandled promises in Topic.flush and PubSub.close prevent proper shutdown #1463

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	flush(): Promise<void>;
	flush(callback: EmptyCallback): void;
	flush(callback?: EmptyCallback): Promise<void> \| void {
	// It doesn't matter here if callback is undefined; the Publisher
	// flush() will handle it.
	this.publisher.flush(callback!);
	}

	flush(callback?: EmptyCallback): Promise<void> \| void {
	const definedCallback = callback ? callback : () => {};

	const publishes = [promisify(this.queue.publish).bind(this.queue)()];
	Array.from(this.orderedQueues.values()).forEach(q =>
	publishes.push(promisify(q.publish).bind(q)())
	);
	const allPublishes = Promise.all(publishes);

	allPublishes
	.then(() => {
	definedCallback(null);
	})
	.catch(definedCallback);
	}

	publish(callback?: PublishDone): void {
	const {messages, callbacks} = this.batch;

	this.batch = new MessageBatch(this.batchOptions);

	if (this.pending) {
	clearTimeout(this.pending);
	delete this.pending;
	}

	this._publish(messages, callbacks, callback);
	}

Unhandled promises in Topic.flush and PubSub.close prevent proper shutdown #1463

Description

Activity

cdupuis commented on Jan 15, 2022

feywind commented on Jan 17, 2022

feywind commented on Mar 24, 2022

feywind commented on Mar 28, 2022

feywind commented on Feb 8, 2023

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions