Add error handling for minimal dispatch (#3176)

Cargo.lock

@@ -271,6 +271,7 @@ version = "0.21.0"
 dependencies = [
  "ansi_term 0.12.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "atty 0.2.13 (registry+https://github.com/rust-lang/crates.io-index)",
+ "byteorder 1.3.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "clap 2.33.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "deno 0.21.0",
  "deno_typescript 0.21.0",

cli/Cargo.toml

@@ -27,6 +27,7 @@ deno_typescript = { path = "../deno_typescript", version = "0.21.0" }
 
 ansi_term = "0.12.1"
 atty = "0.2.13"
+byteorder = "1.3.2"
 clap = "2.33.0"
 dirs = "2.0.2"
 futures = "0.1.29"

cli/js/dispatch_minimal.ts

@@ -1,13 +1,17 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
 import * as util from "./util.ts";
 import { core } from "./core.ts";
+import { TextDecoder } from "./text_encoding.ts";
+import { ErrorKind, DenoError } from "./errors.ts";
 
-const promiseTableMin = new Map<number, util.Resolvable<number>>();
+const promiseTableMin = new Map<number, util.Resolvable<RecordMinimal>>();
 // Note it's important that promiseId starts at 1 instead of 0, because sync
 // messages are indicated with promiseId 0. If we ever add wrap around logic for
 // overflows, this should be taken into account.
 let _nextPromiseId = 1;
 
+const decoder = new TextDecoder();
+
 function nextPromiseId(): number {
   return _nextPromiseId++;
 }
@@ -17,23 +21,51 @@ export interface RecordMinimal {
   opId: number; // Maybe better called dispatchId
   arg: number;
   result: number;
+  err?: {
+    kind: ErrorKind;
+    message: string;
+  };
 }
 
 export function recordFromBufMinimal(
   opId: number,
-  buf32: Int32Array
+  ui8: Uint8Array
 ): RecordMinimal {
-  if (buf32.length != 3) {
-    throw Error("Bad message");
+  const header = ui8.slice(0, 12);
+  const buf32 = new Int32Array(
+    header.buffer,
+    header.byteOffset,
+    header.byteLength / 4
+  );
+  const promiseId = buf32[0];
+  const arg = buf32[1];
+  const result = buf32[2];
+  let err;
+
+  if (arg < 0) {
+    const kind = result as ErrorKind;
+    const message = decoder.decode(ui8.slice(12));
+    err = { kind, message };
+  } else if (ui8.length != 12) {
+    err = { kind: ErrorKind.InvalidData, message: "Bad message" };
   }
+
   return {
-    promiseId: buf32[0],
+    promiseId,
     opId,
-    arg: buf32[1],
-    result: buf32[2]
+    arg,
+    result,
+    err
   };
 }
 
+function unwrapResponse(res: RecordMinimal): number {
+  if (res.err != null) {
+    throw new DenoError(res.err!.kind, res.err!.message);
+  }
+  return res.result;
+}
+
 const scratch32 = new Int32Array(3);
 const scratchBytes = new Uint8Array(
   scratch32.buffer,
@@ -43,15 +75,14 @@ const scratchBytes = new Uint8Array(
 util.assert(scratchBytes.byteLength === scratch32.length * 4);
 
 export function asyncMsgFromRust(opId: number, ui8: Uint8Array): void {
-  const buf32 = new Int32Array(ui8.buffer, ui8.byteOffset, ui8.byteLength / 4);
-  const record = recordFromBufMinimal(opId, buf32);
-  const { promiseId, result } = record;
+  const record = recordFromBufMinimal(opId, ui8);
+  const { promiseId } = record;
   const promise = promiseTableMin.get(promiseId);
   promiseTableMin.delete(promiseId);
-  promise!.resolve(result);
+  promise!.resolve(record);
 }
 
-export function sendAsyncMinimal(
+export async function sendAsyncMinimal(
   opId: number,
   arg: number,
   zeroCopy: Uint8Array
@@ -60,22 +91,19 @@ export function sendAsyncMinimal(
   scratch32[0] = promiseId;
   scratch32[1] = arg;
   scratch32[2] = 0; // result
-  const promise = util.createResolvable<number>();
+  const promise = util.createResolvable<RecordMinimal>();
   const buf = core.dispatch(opId, scratchBytes, zeroCopy);
   if (buf) {
-    const buf32 = new Int32Array(
-      buf.buffer,
-      buf.byteOffset,
-      buf.byteLength / 4
-    );
-    const record = recordFromBufMinimal(opId, buf32);
+    const record = recordFromBufMinimal(opId, buf);
     // Sync result.
-    promise.resolve(record.result);
+    promise.resolve(record);
   } else {
     // Async result.
     promiseTableMin.set(promiseId, promise);
   }
-  return promise;
+
+  const res = await promise;
+  return unwrapResponse(res);
 }
 
 export function sendSyncMinimal(
@@ -86,7 +114,6 @@ export function sendSyncMinimal(
   scratch32[0] = 0; // promiseId 0 indicates sync
   scratch32[1] = arg;
   const res = core.dispatch(opId, scratchBytes, zeroCopy)!;
-  const res32 = new Int32Array(res.buffer, res.byteOffset, 3);
-  const resRecord = recordFromBufMinimal(opId, res32);
-  return resRecord.result;
+  const resRecord = recordFromBufMinimal(opId, res);
+  return unwrapResponse(resRecord);
 }

cli/ops/dispatch_minimal.rs

@@ -4,6 +4,8 @@
 //! alternative to flatbuffers using a very simple list of int32s to lay out
 //! messages. The first i32 is used to determine if a message a flatbuffer
 //! message or a "minimal" message.
+use crate::deno_error::GetErrorKind;
+use byteorder::{LittleEndian, WriteBytesExt};
 use deno::Buf;
 use deno::CoreOp;
 use deno::ErrBox;
@@ -31,6 +33,44 @@ impl Into<Buf> for Record {
   }
 }
 
+pub struct ErrorRecord {
+  pub promise_id: i32,
+  pub arg: i32,
+  pub error_code: i32,
+  pub error_message: Vec<u8>,
+}
+
+impl Into<Buf> for ErrorRecord {
+  fn into(self) -> Buf {
+    let v32: Vec<i32> = vec![self.promise_id, self.arg, self.error_code];
+    let mut v8: Vec<u8> = Vec::new();
+    for n in v32 {
+      v8.write_i32::<LittleEndian>(n).unwrap();
+    }
+    let mut message = self.error_message;
+    // Align to 32bit word, padding with the space character.
+    message.resize((message.len() + 3usize) & !3usize, b' ');
+    v8.append(&mut message);
+    v8.into_boxed_slice()
+  }
+}
+
+#[test]
+fn test_error_record() {
+  let expected = vec![
+    1, 0, 0, 0, 255, 255, 255, 255, 10, 0, 0, 0, 69, 114, 114, 111, 114, 32,
+    32, 32,
+  ];
+  let err_record = ErrorRecord {
+    promise_id: 1,
+    arg: -1,
+    error_code: 10,
+    error_message: "Error".to_string().as_bytes().to_owned(),
+  };
+  let buf: Buf = err_record.into();
+  assert_eq!(buf, expected.into_boxed_slice());
+}
+
 pub fn parse_min_record(bytes: &[u8]) -> Option<Record> {
   if bytes.len() % std::mem::size_of::<i32>() != 0 {
     return None;
@@ -85,15 +125,18 @@ pub fn minimal_op(
       match result {
         Ok(r) => {
           record.result = r;
+          Ok(record.into())
         }
         Err(err) => {
-          // TODO(ry) The dispatch_minimal doesn't properly pipe errors back to
-          // the caller.
-          debug!("swallowed err {}", err);
-          record.result = -1;
+          let error_record = ErrorRecord {
+            promise_id: record.promise_id,
+            arg: -1,
+            error_code: err.kind() as i32,
+            error_message: err.to_string().as_bytes().to_owned(),
+          };
+          Ok(error_record.into())
         }
       }
-      Ok(record.into())
     }));
 
     if is_sync {