Track forwarded requests to notify if link dies

Keep track of requests that have been forwarded over a link in a similar
manner to requests that were initiated on a link in order to properly
return an error response to the sender of the request if the link goes
down while the request is still pending at the forwarded node.

packages/controller/src/ControllerRouter.js

@@ -84,7 +84,11 @@ class ControllerRouter {
 		}
 
 		if (nextHop) {
-			nextHop.connector.send(message);
+			if (message.type === "request") {
+				nextHop.forwardRequest(message, origin);
+			} else {
+				nextHop.connector.send(message);
+			}
 		} else {
 			this.warnUnrouted(message, msg);
 		}

packages/host/src/Host.js

@@ -151,7 +151,7 @@ class HostRouter {
 		}
 
 		if (nextHop) {
-			this.sendMessage(nextHop, message);
+			this.sendMessage(nextHop, message, origin);
 		} else {
 			this.warnUnrouted(message, msg);
 		}
@@ -208,9 +208,13 @@ class HostRouter {
 		});
 	}
 
-	sendMessage(nextHop, message) {
+	sendMessage(nextHop, message, origin) {
 		try {
-			nextHop.connector.send(message);
+			if (message.type === "request") {
+				nextHop.forwardRequest(message, origin);
+			} else {
+				nextHop.connector.send(message);
+			}
 		} catch (err) {
 			if (message.type === "request") {
 				origin.connector.sendResponseError(

packages/lib/data/messages_core.js

@@ -98,6 +98,14 @@ class Address {
 		return `[Address ${type}:${this.id}:${this.requestId}]`;
 	}
 
+	/**
+	 * Returns a string suitable for indexing the address in a Map
+	 * @returns {string} unique string for this address
+	 */
+	index() {
+		return `${this.type}:${this.id}:${this.requestId}`;
+	}
+
 	/**
 	 * Returns true if this address targets the given destination
 	 * @param {module:lib/Data.Address} dst - Destination to check

packages/lib/link/link.js

@@ -30,6 +30,7 @@ class Link {
 		this._eventHandlers = new Map();
 		this._eventSnoopers = new Map();
 		this._pendingRequests = new Map();
+		this._forwardedRequests = new Map();
 		this._nextRequestId = 1;
 
 		this.handle(libData.PingRequest, () => {});
@@ -65,22 +66,29 @@ class Link {
 		});
 
 		connector.on("invalidate", () => {
-			for (let pending of this._pendingRequests.values()) {
-				pending.reject(new libErrors.SessionLost("Session Lost"));
-			}
-
-			this._pendingRequests.clear();
+			this._clearPendingRequests(new libErrors.SessionLost("Session Lost"));
 		});
 
 		connector.on("close", () => {
-			for (let pending of this._pendingRequests.values()) {
-				pending.reject(new libErrors.SessionLost("Session Closed"));
-			}
-
-			this._pendingRequests.clear();
+			this._clearPendingRequests(new libErrors.SessionLost("Session Closed"));
 		});
 	}
 
+	_clearPendingRequests(err) {
+		for (let pending of this._pendingRequests.values()) {
+			pending.reject(err);
+		}
+		for (let pending of this._forwardedRequests.values()) {
+			if (pending.origin.connector.hasSession) {
+				pending.origin.connector.sendResponseError(
+					new libData.ResponseError(err.message, err.code), pending.src, pending.dst
+				);
+			}
+		}
+		this._pendingRequests.clear();
+		this._forwardedRequests.clear();
+	}
+
 	/**
 	 * Process a received message on the link
 	 *
@@ -115,6 +123,9 @@ class Link {
 		}
 
 		if (!message.dst.addressedTo(this.connector.src)) {
+			if (message.type === "response" || message.type === "responseError") {
+				this._forwardedRequests.delete(message.dst.index());
+			}
 			this._routeMessage(message, entry);
 			return;
 		}
@@ -411,6 +422,24 @@ class Link {
 		return pending.promise;
 	}
 
+	forwardRequest(message, origin) {
+		if (this.validateSent) {
+			let entry = this.constructor._requestsByName.get(message.name);
+			if (!entry) {
+				throw new Error(`Attempt to forward unregistered Request ${message.name}`);
+			}
+			entry.requestFromJSON(message.data);
+		}
+
+		let pending = {
+			origin,
+			src: message.src,
+			dst: message.dst,
+		};
+		this._forwardedRequests.set(message.src.index(), pending);
+		this.connector.send(message);
+	}
+
 	sendEvent(event, dst) {
 		let entry = this.constructor._eventsByClass.get(event.constructor);
 		if (!entry) {