Fix subtle unhandledRejection case (#84)

* Fix subtle unhandledRejection case for iterables that reject when
encountering the first input rejection (merge, all, race).  This bug
could cause a spurious unhandledRejection under specific conditions.

Rename vars and params in ErrorHandler to be more intuitive.

* Fix indents

src/ErrorHandler.js

@@ -5,41 +5,41 @@ const HANDLED_REJECTION = 'rejectionHandled'
 
 export default class ErrorHandler {
 	constructor (emitEvent, reportError) {
-		this.errors = []
+		this.rejections = []
 		this.emit = emitEvent
 		this.reportError = reportError
 	}
 
-	track (e) {
-		if (!this.emit(UNHANDLED_REJECTION, e, e.value)) {
+	track (rejected) {
+		if (!this.emit(UNHANDLED_REJECTION, rejected, rejected.value)) {
 			/* istanbul ignore else */
-			if (this.errors.length === 0) {
-				setTimeout(reportErrors, 1, this.reportError, this.errors)
+			if (this.rejections.length === 0) {
+				setTimeout(reportErrors, 1, this.reportError, this.rejections)
 			}
-			this.errors.push(e)
+			this.rejections.push(rejected)
 		}
 	}
 
-	untrack (e) {
-		silenceError(e)
-		this.emit(HANDLED_REJECTION, e)
+	untrack (rejected) {
+		silenceError(rejected)
+		this.emit(HANDLED_REJECTION, rejected)
 	}
 }
 
-function reportErrors (report, errors) {
+function reportErrors (report, rejections) {
 	try {
-		reportAll(errors, report)
+		reportAll(rejections, report)
 	} finally {
-		errors.length = 0
+		rejections.length = 0
 	}
 }
 
-function reportAll (errors, report) {
-	for (let i = 0; i < errors.length; ++i) {
-		const e = errors[i]
+function reportAll (rejections, report) {
+	for (let i = 0; i < rejections.length; ++i) {
+		const rejected = rejections[i]
 		/* istanbul ignore else */
-		if (!isHandled(e)) {
-			report(e)
+		if (!isHandled(rejected)) {
+			report(rejected)
 		}
 	}
 }

src/Merge.js

@@ -1,3 +1,5 @@
+import { silenceError } from './inspect'
+
 export default class Merge {
 	constructor (mergeHandler, results) {
 		this.pending = 0
@@ -15,7 +17,9 @@ export default class Merge {
 	}
 
 	rejectAt (p, i, promise) {
-		promise._become(p)
+		// In the case where the result promise has been resolved
+		// need to silence all subsequently seen rejections
+		promise._isResolved() ? silenceError(p) : promise._become(p)
 	}
 
 	complete (total, promise) {

src/Race.js

@@ -1,3 +1,5 @@
+import { silenceError } from './inspect'
+
 export default class Race {
 	constructor (never) {
 		this.never = never
@@ -12,7 +14,9 @@ export default class Race {
 	}
 
 	rejectAt (p, i, promise) {
-		promise._become(p)
+		// In the case where the result promise has been resolved
+		// need to silence all subsequently seen rejections
+		promise._isResolved() ? silenceError(p) : promise._become(p)
 	}
 
 	complete (total, promise) {

src/inspect.js

@@ -50,9 +50,7 @@ export function silenceError (p) {
 // implements Action
 const silencer = {
 	fulfilled () {},
-	rejected: setHandled
-}
-
-function setHandled (rejected) {
-	rejected._state |= HANDLED
+	rejected (p) {
+		p._state |= HANDLED
+	}
 }

test/unhandledRejection-node.js

@@ -0,0 +1,45 @@
+import { describe, it } from 'mocha'
+import { race, reject, all, future } from '../src/main'
+import { isHandled } from '../src/inspect'
+
+const delayRejectWith = (ms, e) => {
+	const { resolve, promise } = future()
+	setTimeout((resolve, e) => resolve(reject(e)), ms, resolve, e)
+	return promise
+}
+
+const expectOne = aggregate => done => {
+	const expected = new Error('expected')
+
+	// Arrange for 2 delayed rejections. The first will cause the promise
+	// returned by aggregate() to reject, and the second should be silenced.
+	// After the first rejects, arrange to intercept the second via unhandledRejection.
+	// After another delay, giving the second promise enough time to have been silenced,
+	// check to see if it was indeed silenced.
+	aggregate([delayRejectWith(10, expected), delayRejectWith(15, new Error('Unsilenced rejection'))])
+		.catch(() => {
+			const rejections = []
+			const unhandledRejection = (e, p) => rejections.push(p)
+			const rejectionHandled = p => rejections.splice(rejections.indexOf(p), 1)
+
+			process.on('unhandledRejection', unhandledRejection)
+			process.on('rejectionHandled', rejectionHandled)
+
+			setTimeout(() => {
+				process.removeListener('unhandledRejection', unhandledRejection)
+				process.removeListener('rejectionHandled', rejectionHandled)
+
+				const remaining = rejections.filter(r => !isHandled(r))
+				if(remaining.length > 0) {
+					done(remaining.pop().value)
+				} else {
+					done()
+				}
+			}, 50)
+		})
+}
+
+describe('unhandledRejection-node', () => {
+	it('race should emit 1 unhandledRejection', expectOne(race))
+	it('all should emit 1 unhandledRejection', expectOne(all))
+})