fix: replace bad rid panics with errors (#2870)

cli/main.rs

@@ -45,6 +45,7 @@ mod signal;
 pub mod source_maps;
 mod startup_data;
 pub mod state;
+mod tokio_read;
 mod tokio_util;
 mod tokio_write;
 pub mod version;

cli/ops/files.rs

@@ -1,6 +1,5 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
 use super::dispatch_json::{Deserialize, JsonOp, Value};
-use crate::deno_error;
 use crate::fs as deno_fs;
 use crate::resources;
 use crate::state::ThreadSafeState;
@@ -104,13 +103,9 @@ pub fn op_close(
 ) -> Result<JsonOp, ErrBox> {
   let args: CloseArgs = serde_json::from_value(args)?;
 
-  match resources::lookup(args.rid as u32) {
-    None => Err(deno_error::bad_resource()),
-    Some(resource) => {
-      resource.close();
-      Ok(JsonOp::Sync(json!({})))
-    }
-  }
+  let resource = resources::lookup(args.rid as u32)?;
+  resource.close();
+  Ok(JsonOp::Sync(json!({})))
 }
 
 #[derive(Deserialize)]
@@ -129,17 +124,13 @@ pub fn op_seek(
 ) -> Result<JsonOp, ErrBox> {
   let args: SeekArgs = serde_json::from_value(args)?;
 
-  match resources::lookup(args.rid as u32) {
-    None => Err(deno_error::bad_resource()),
-    Some(resource) => {
-      let op = resources::seek(resource, args.offset, args.whence as u32)
-        .and_then(move |_| futures::future::ok(json!({})));
-      if args.promise_id.is_none() {
-        let buf = op.wait()?;
-        Ok(JsonOp::Sync(buf))
-      } else {
-        Ok(JsonOp::Async(Box::new(op)))
-      }
-    }
+  let resource = resources::lookup(args.rid as u32)?;
+  let op = resources::seek(resource, args.offset, args.whence as u32)
+    .and_then(move |_| futures::future::ok(json!({})));
+  if args.promise_id.is_none() {
+    let buf = op.wait()?;
+    Ok(JsonOp::Sync(buf))
+  } else {
+    Ok(JsonOp::Async(Box::new(op)))
   }
 }

cli/ops/io.rs

@@ -1,6 +1,7 @@
 use super::dispatch_minimal::MinimalOp;
 use crate::deno_error;
 use crate::resources;
+use crate::tokio_read;
 use crate::tokio_write;
 use deno::ErrBox;
 use deno::PinnedBuf;
@@ -14,10 +15,11 @@ pub fn op_read(rid: i32, zero_copy: Option<PinnedBuf>) -> Box<MinimalOp> {
     }
     Some(buf) => buf,
   };
+
   match resources::lookup(rid as u32) {
-    None => Box::new(futures::future::err(deno_error::bad_resource())),
-    Some(resource) => Box::new(
-      tokio::io::read(resource, zero_copy)
+    Err(e) => Box::new(futures::future::err(e)),
+    Ok(resource) => Box::new(
+      tokio_read::read(resource, zero_copy)
         .map_err(ErrBox::from)
         .and_then(move |(_resource, _buf, nread)| Ok(nread as i32)),
     ),
@@ -32,9 +34,10 @@ pub fn op_write(rid: i32, zero_copy: Option<PinnedBuf>) -> Box<MinimalOp> {
     }
     Some(buf) => buf,
   };
+
   match resources::lookup(rid as u32) {
-    None => Box::new(futures::future::err(deno_error::bad_resource())),
-    Some(resource) => Box::new(
+    Err(e) => Box::new(futures::future::err(e)),
+    Ok(resource) => Box::new(
       tokio_write::write(resource, zero_copy)
         .map_err(ErrBox::from)
         .and_then(move |(_resource, _buf, nwritten)| Ok(nwritten as i32)),

cli/ops/net.rs

@@ -1,6 +1,5 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
 use super::dispatch_json::{Deserialize, JsonOp, Value};
-use crate::deno_error;
 use crate::resolve_addr::resolve_addr;
 use crate::resources;
 use crate::resources::Resource;
@@ -28,28 +27,24 @@ pub fn op_accept(
   let args: AcceptArgs = serde_json::from_value(args)?;
   let server_rid = args.rid as u32;
 
-  match resources::lookup(server_rid) {
-    None => Err(deno_error::bad_resource()),
-    Some(server_resource) => {
-      let op = tokio_util::accept(server_resource)
-        .and_then(move |(tcp_stream, _socket_addr)| {
-          let local_addr = tcp_stream.local_addr()?;
-          let remote_addr = tcp_stream.peer_addr()?;
-          let tcp_stream_resource = resources::add_tcp_stream(tcp_stream);
-          Ok((tcp_stream_resource, local_addr, remote_addr))
-        })
-        .map_err(ErrBox::from)
-        .and_then(move |(tcp_stream_resource, local_addr, remote_addr)| {
-          futures::future::ok(json!({
-            "rid": tcp_stream_resource.rid,
-            "localAddr": local_addr.to_string(),
-            "remoteAddr": remote_addr.to_string(),
-          }))
-        });
+  let server_resource = resources::lookup(server_rid)?;
+  let op = tokio_util::accept(server_resource)
+    .and_then(move |(tcp_stream, _socket_addr)| {
+      let local_addr = tcp_stream.local_addr()?;
+      let remote_addr = tcp_stream.peer_addr()?;
+      let tcp_stream_resource = resources::add_tcp_stream(tcp_stream);
+      Ok((tcp_stream_resource, local_addr, remote_addr))
+    })
+    .map_err(ErrBox::from)
+    .and_then(move |(tcp_stream_resource, local_addr, remote_addr)| {
+      futures::future::ok(json!({
+        "rid": tcp_stream_resource.rid,
+        "localAddr": local_addr.to_string(),
+        "remoteAddr": remote_addr.to_string(),
+      }))
+    });
 
-      Ok(JsonOp::Async(Box::new(op)))
-    }
-  }
+  Ok(JsonOp::Async(Box::new(op)))
 }
 
 #[derive(Deserialize)]
@@ -105,22 +100,19 @@ pub fn op_shutdown(
 ) -> Result<JsonOp, ErrBox> {
   let args: ShutdownArgs = serde_json::from_value(args)?;
 
-  let rid = args.rid;
+  let rid = args.rid as u32;
   let how = args.how;
-  match resources::lookup(rid as u32) {
-    None => Err(deno_error::bad_resource()),
-    Some(mut resource) => {
-      let shutdown_mode = match how {
-        0 => Shutdown::Read,
-        1 => Shutdown::Write,
-        _ => unimplemented!(),
-      };
+  let mut resource = resources::lookup(rid)?;
 
-      // Use UFCS for disambiguation
-      Resource::shutdown(&mut resource, shutdown_mode)?;
-      Ok(JsonOp::Sync(json!({})))
-    }
-  }
+  let shutdown_mode = match how {
+    0 => Shutdown::Read,
+    1 => Shutdown::Write,
+    _ => unimplemented!(),
+  };
+
+  // Use UFCS for disambiguation
+  Resource::shutdown(&mut resource, shutdown_mode)?;
+  Ok(JsonOp::Sync(json!({})))
 }
 
 #[derive(Deserialize)]

cli/resources.rs

@@ -217,15 +217,13 @@ impl Resource {
 
   pub fn shutdown(&mut self, how: Shutdown) -> Result<(), ErrBox> {
     let mut table = RESOURCE_TABLE.lock().unwrap();
-    let maybe_repr = table.get_mut(&self.rid);
-    match maybe_repr {
-      None => panic!("bad rid"),
-      Some(repr) => match repr {
-        Repr::TcpStream(ref mut f) => {
-          TcpStream::shutdown(f, how).map_err(ErrBox::from)
-        }
-        _ => panic!("Cannot shutdown"),
-      },
+    let repr = table.get_mut(&self.rid).ok_or_else(bad_resource)?;
+
+    match repr {
+      Repr::TcpStream(ref mut f) => {
+        TcpStream::shutdown(f, how).map_err(ErrBox::from)
+      }
+      _ => Err(bad_resource()),
     }
   }
 }
@@ -236,22 +234,30 @@ impl Read for Resource {
   }
 }
 
-impl AsyncRead for Resource {
-  fn poll_read(&mut self, buf: &mut [u8]) -> Poll<usize, Error> {
+/// `DenoAsyncRead` is the same as the `tokio_io::AsyncRead` trait
+/// but uses an `ErrBox` error instead of `std::io:Error`
+pub trait DenoAsyncRead {
+  fn poll_read(&mut self, buf: &mut [u8]) -> Poll<usize, ErrBox>;
+}
+
+impl DenoAsyncRead for Resource {
+  fn poll_read(&mut self, buf: &mut [u8]) -> Poll<usize, ErrBox> {
     let mut table = RESOURCE_TABLE.lock().unwrap();
-    let maybe_repr = table.get_mut(&self.rid);
-    match maybe_repr {
-      None => panic!("bad rid"),
-      Some(repr) => match repr {
-        Repr::FsFile(ref mut f) => f.poll_read(buf),
-        Repr::Stdin(ref mut f) => f.poll_read(buf),
-        Repr::TcpStream(ref mut f) => f.poll_read(buf),
-        Repr::HttpBody(ref mut f) => f.poll_read(buf),
-        Repr::ChildStdout(ref mut f) => f.poll_read(buf),
-        Repr::ChildStderr(ref mut f) => f.poll_read(buf),
-        _ => panic!("Cannot read"),
-      },
-    }
+    let repr = table.get_mut(&self.rid).ok_or_else(bad_resource)?;
+
+    let r = match repr {
+      Repr::FsFile(ref mut f) => f.poll_read(buf),
+      Repr::Stdin(ref mut f) => f.poll_read(buf),
+      Repr::TcpStream(ref mut f) => f.poll_read(buf),
+      Repr::HttpBody(ref mut f) => f.poll_read(buf),
+      Repr::ChildStdout(ref mut f) => f.poll_read(buf),
+      Repr::ChildStderr(ref mut f) => f.poll_read(buf),
+      _ => {
+        return Err(bad_resource());
+      }
+    };
+
+    r.map_err(ErrBox::from)
   }
 }
 
@@ -265,24 +271,34 @@ impl Write for Resource {
   }
 }
 
-impl AsyncWrite for Resource {
-  fn poll_write(&mut self, buf: &[u8]) -> Poll<usize, Error> {
+/// `DenoAsyncWrite` is the same as the `tokio_io::AsyncWrite` trait
+/// but uses an `ErrBox` error instead of `std::io:Error`
+pub trait DenoAsyncWrite {
+  fn poll_write(&mut self, buf: &[u8]) -> Poll<usize, ErrBox>;
+
+  fn shutdown(&mut self) -> Poll<(), ErrBox>;
+}
+
+impl DenoAsyncWrite for Resource {
+  fn poll_write(&mut self, buf: &[u8]) -> Poll<usize, ErrBox> {
     let mut table = RESOURCE_TABLE.lock().unwrap();
-    let maybe_repr = table.get_mut(&self.rid);
-    match maybe_repr {
-      None => panic!("bad rid"),
-      Some(repr) => match repr {
-        Repr::FsFile(ref mut f) => f.poll_write(buf),
-        Repr::Stdout(ref mut f) => f.poll_write(buf),
-        Repr::Stderr(ref mut f) => f.poll_write(buf),
-        Repr::TcpStream(ref mut f) => f.poll_write(buf),
-        Repr::ChildStdin(ref mut f) => f.poll_write(buf),
-        _ => panic!("Cannot write"),
-      },
-    }
+    let repr = table.get_mut(&self.rid).ok_or_else(bad_resource)?;
+
+    let r = match repr {
+      Repr::FsFile(ref mut f) => f.poll_write(buf),
+      Repr::Stdout(ref mut f) => f.poll_write(buf),
+      Repr::Stderr(ref mut f) => f.poll_write(buf),
+      Repr::TcpStream(ref mut f) => f.poll_write(buf),
+      Repr::ChildStdin(ref mut f) => f.poll_write(buf),
+      _ => {
+        return Err(bad_resource());
+      }
+    };
+
+    r.map_err(ErrBox::from)
   }
 
-  fn shutdown(&mut self) -> futures::Poll<(), std::io::Error> {
+  fn shutdown(&mut self) -> futures::Poll<(), ErrBox> {
     unimplemented!()
   }
 }
@@ -295,10 +311,9 @@ fn new_rid() -> ResourceId {
 pub fn add_fs_file(fs_file: tokio::fs::File) -> Resource {
   let rid = new_rid();
   let mut tg = RESOURCE_TABLE.lock().unwrap();
-  match tg.insert(rid, Repr::FsFile(fs_file)) {
-    Some(_) => panic!("There is already a file with that rid"),
-    None => Resource { rid },
-  }
+  let r = tg.insert(rid, Repr::FsFile(fs_file));
+  assert!(r.is_none());
+  Resource { rid }
 }
 
 pub fn add_tcp_listener(listener: tokio::net::TcpListener) -> Resource {
@@ -354,6 +369,7 @@ pub fn post_message_to_worker(
       // unwrap here is incorrect, but doing it anyway
       wc.0.clone().send(buf)
     }
+    // TODO: replace this panic with `bad_resource`
     _ => panic!("bad resource"), // futures::future::err(bad_resource()).into(),
   }
 }
@@ -522,10 +538,13 @@ pub fn get_file(rid: ResourceId) -> Result<std::fs::File, ErrBox> {
   }
 }
 
-pub fn lookup(rid: ResourceId) -> Option<Resource> {
+pub fn lookup(rid: ResourceId) -> Result<Resource, ErrBox> {
   debug!("resource lookup {}", rid);
   let table = RESOURCE_TABLE.lock().unwrap();
-  table.get(&rid).map(|_| Resource { rid })
+  table
+    .get(&rid)
+    .ok_or_else(bad_resource)
+    .map(|_| Resource { rid })
 }
 
 pub fn seek(

cli/tokio_read.rs

@@ -0,0 +1,64 @@
+// Copyright (c) 2019 Tokio Contributors. All rights reserved. MIT license.
+// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+// Forked from: https://github.com/tokio-rs/tokio/blob/9b3f8564af4bb1aee07fab3c401eb412ca5eeac5/tokio-io/src/io/read.rs
+use crate::resources::DenoAsyncRead;
+use deno::ErrBox;
+use futures::{Future, Poll};
+use std::mem;
+
+/// This is almost the same implementation as in tokio, the only difference is
+/// that error type is `ErrBox` instead of `std::io::Error`.
+
+#[derive(Debug)]
+enum State<R, T> {
+  Pending { rd: R, buf: T },
+  Empty,
+}
+
+/// Tries to read some bytes directly into the given `buf` in asynchronous
+/// manner, returning a future type.
+///
+/// The returned future will resolve to both the I/O stream and the buffer
+/// as well as the number of bytes read once the read operation is completed.
+pub fn read<R, T>(rd: R, buf: T) -> Read<R, T>
+where
+  R: DenoAsyncRead,
+  T: AsMut<[u8]>,
+{
+  Read {
+    state: State::Pending { rd, buf },
+  }
+}
+
+/// A future which can be used to easily read available number of bytes to fill
+/// a buffer.
+///
+/// Created by the [`read`] function.
+#[derive(Debug)]
+pub struct Read<R, T> {
+  state: State<R, T>,
+}
+
+impl<R, T> Future for Read<R, T>
+where
+  R: DenoAsyncRead,
+  T: AsMut<[u8]>,
+{
+  type Item = (R, T, usize);
+  type Error = ErrBox;
+
+  fn poll(&mut self) -> Poll<(R, T, usize), ErrBox> {
+    let nread = match self.state {
+      State::Pending {
+        ref mut rd,
+        ref mut buf,
+      } => try_ready!(rd.poll_read(&mut buf.as_mut()[..])),
+      State::Empty => panic!("poll a Read after it's done"),
+    };
+
+    match mem::replace(&mut self.state, State::Empty) {
+      State::Pending { rd, buf } => Ok((rd, buf, nread).into()),
+      State::Empty => panic!("invalid internal state"),
+    }
+  }
+}

cli/tokio_write.rs

@@ -1,10 +1,8 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
-// TODO Submit this file upstream into tokio-io/src/io/write.rs
-use std::io;
-use std::mem;
-
+use crate::resources::DenoAsyncWrite;
+use deno::ErrBox;
 use futures::{Future, Poll};
-use tokio::io::AsyncWrite;
+use std::mem;
 
 /// A future used to write some data to a stream.
 ///
@@ -29,7 +27,7 @@ enum State<A, T> {
 /// buffer to get destroyed.
 pub fn write<A, T>(a: A, buf: T) -> Write<A, T>
 where
-  A: AsyncWrite,
+  A: DenoAsyncWrite,
   T: AsRef<[u8]>,
 {
   Write {
@@ -37,15 +35,17 @@ where
   }
 }
 
+/// This is almost the same implementation as in tokio, difference is
+/// that error type is `ErrBox` instead of `std::io::Error`.
 impl<A, T> Future for Write<A, T>
 where
-  A: AsyncWrite,
+  A: DenoAsyncWrite,
   T: AsRef<[u8]>,
 {
   type Item = (A, T, usize);
-  type Error = io::Error;
+  type Error = ErrBox;
 
-  fn poll(&mut self) -> Poll<(A, T, usize), io::Error> {
+  fn poll(&mut self) -> Poll<(A, T, usize), ErrBox> {
     let nwritten = match self.state {
       State::Pending {
         ref mut a,

tests/044_bad_resource.test

@@ -0,0 +1,4 @@
+args: run --reload --allow-read tests/044_bad_resource.ts
+output: tests/044_bad_resource.ts.out
+check_stderr: true
+exit_code: 1

tests/044_bad_resource.ts

@@ -0,0 +1,7 @@
+async function main(): Promise<void> {
+  const file = await Deno.open("Cargo.toml", "r");
+  file.close();
+  await file.seek(10, 0);
+}
+
+main();

tests/044_bad_resource.ts.out

@@ -0,0 +1,6 @@
+[WILDCARD]
+error: Uncaught BadResource: bad resource id
+[WILDCARD]dispatch_json.ts:[WILDCARD]
+    at DenoError ([WILDCARD]errors.ts:[WILDCARD])
+    at unwrapResponse ([WILDCARD]dispatch_json.ts:[WILDCARD])
+    at sendAsync ([WILDCARD]dispatch_json.ts:[WILDCARD])