-
Notifications
You must be signed in to change notification settings - Fork 124
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
fix(client): asyncEventEmitter to not silence unhandled exceptions raised in event handlers #1247
Changes from 2 commits
0e2ac33
8d40d05
aac0d54
eaf473d
dbe1d12
510a48f
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,5 @@ | ||
--- | ||
"electric-sql": patch | ||
--- | ||
|
||
Fix asyncEventEmitter to not silence unhandled exceptions raised in event handlers. |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -24,7 +24,7 @@ export class AsyncEventEmitter<Events extends EventMap> { | |
private eventQueue: Array< | ||
EmittedEvent<keyof Events, Parameters<Events[keyof Events]>> | ||
> = [] | ||
private processing = false // indicates whether the event queue is currently being processed | ||
private processing: Promise<PromiseSettledResult<void>[]> | false = false // indicates whether the event queue is currently being processed | ||
|
||
private getListeners<E extends keyof Events>(event: E): Array<Events[E]> { | ||
return this.listeners[event] ?? [] | ||
|
@@ -127,8 +127,6 @@ export class AsyncEventEmitter<Events extends EventMap> { | |
* and an 'error' event is emitted, the error is thrown. | ||
*/ | ||
private processQueue() { | ||
this.processing = true | ||
|
||
const emittedEvent = this.eventQueue.shift() | ||
if (emittedEvent) { | ||
// We call all listeners and process the next event when all listeners finished. | ||
|
@@ -148,15 +146,29 @@ export class AsyncEventEmitter<Events extends EventMap> { | |
// deep copy because once listeners mutate the `this.listeners` array as they remove themselves | ||
// which breaks the `map` which iterates over that same array while the contents may shift | ||
const ls = [...listeners] | ||
const listenerProms = ls.map(async (listener) => await listener(...args)) | ||
|
||
Promise | ||
// wait for all listeners to finish, | ||
// some may fail (i.e.return a rejected promise) | ||
// but that should not stop the queue from being processed | ||
// hence the use of `allSettled` rather than `all` | ||
.allSettled(listenerProms) | ||
.then(() => this.processQueue()) // only process the next event when all listeners have finished | ||
const listenerProms = ls.map(async (listener) => { | ||
try { | ||
await listener(...args) | ||
} catch (e) { | ||
// If a listener throws an error, we re-throw it asynchronously so that the queue can continue | ||
// to be processed, this ensures that the exception isn't swallowed by allSettled below. | ||
// It will likely be caught by a global error handler, or be logged to the console. | ||
queueMicrotask(() => { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is
So they are talking about browser environments. Perhaps we don't need this There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I'm pretty sure It does change the behaviour slightly though as it will only throw once everything has settled, whereas with There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I've merged the PR to avoid having it linger for too long as I think it's an important one to have in main, but if there's any disagreements on this last change let me know. |
||
throw e | ||
}) | ||
} | ||
}) | ||
|
||
// wait for all listeners to finish, | ||
// some may fail (i.e.return a rejected promise) | ||
// but that should not stop the queue from being processed | ||
// hence the use of `allSettled` rather than `all` | ||
const processingProm = Promise.allSettled(listenerProms) | ||
|
||
// only process the next event when all listeners have finished | ||
processingProm.then(() => this.processQueue()) | ||
|
||
this.processing = processingProm | ||
} else { | ||
// signal that the queue is no longer being processed | ||
this.processing = false | ||
|
@@ -251,4 +263,11 @@ export class AsyncEventEmitter<Events extends EventMap> { | |
this.maxListeners = maxListeners | ||
return this | ||
} | ||
|
||
/** | ||
* Wait for event queue to finish processing. | ||
*/ | ||
async waitForProcessing(): Promise<void> { | ||
await this.processing | ||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
really ugly fix but this is a symptom of how we've mixed and matched the architecture here - if
_applySubscriptionData
fails, it still resolves but fires_handleSubscriptionError
. However the error handler resets the state of thesubscriptionManager
, so theafterApply
call in lines 513-516 still fires because_applySubscriptionData
resolves without an issue but because the state of the subscription manager has been reset it attempts to resolve promises that are no longer present.I think satellite probably needs a bit of rearchitecting as we're seeing various issues w.r.t. timing and async.