Merge pull request #121 from advancedtelematic/bugfix/pro-1331/queued-packages

Check for in-flight updates

13 files changed, 178 insertions, 132 deletions

diff --git a/Cargo.toml b/Cargo.toml
index cbc8b2a..179e4fd 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -34,3 +34,9 @@ toml = "0.1.30"
 unix_socket = "0.5.0"
 url = "1.1.1"
 ws = "0.5.0"
+
+[profile.dev]
+panic = 'abort'
+
+[profile.release]
+panic = 'abort'
diff --git a/Makefile b/Makefile
index 9750616..8cdbfa4 100644
--- a/Makefile
+++ b/Makefile
@@ -68,8 +68,8 @@ clippy: ## Run clippy lint checks using the nightly compiler.
 		rustup run nightly cargo clippy -- -Dclippy
 
 client: rust-openssl src/ ## Compile a new release build of the client.
-	$(CARGO) build --release
-	@cp target/release/sota_client run/
+	$(CARGO) build --release --target=x86_64-unknown-linux-gnu
+	@cp target/x86_64-unknown-linux-gnu/release/sota_client run/
 
 image: client ## Build a Docker image for running the client.
 	@docker build --tag advancedtelematic/sota-client run
diff --git a/src/datatype/command.rs b/src/datatype/command.rs
index 0f59b1a..11707ea 100644
--- a/src/datatype/command.rs
+++ b/src/datatype/command.rs
@@ -3,9 +3,8 @@ use std::str;
 use std::str::FromStr;
 
 use nom::{IResult, space, eof};
-use datatype::{ClientCredentials, ClientId, ClientSecret, DownloadComplete, Error,
-               InstalledSoftware, Package, UpdateReport, UpdateRequestId,
-               UpdateResultCode};
+use datatype::{ClientCredentials, ClientId, ClientSecret, Error, InstalledSoftware,
+               Package, UpdateReport, UpdateRequestId, UpdateResultCode};
 
 
 /// System-wide commands that are sent to the interpreter.
@@ -16,17 +15,20 @@ pub enum Command {
     /// Shutdown the client immediately.
     Shutdown,
 
-    /// Check for any new updates.
-    GetNewUpdates,
+    /// Check for any pending updates.
+    GetPendingUpdates,
+    /// Check for any in-flight updates.
+    GetInFlightUpdates,
+
     /// List the installed packages on the system.
     ListInstalledPackages,
     /// List the system information.
     ListSystemInfo,
 
-    /// Start downloading one or more updates.
-    StartDownload(Vec<UpdateRequestId>),
-    /// Start installing an update
-    StartInstall(DownloadComplete),
+    /// Start downloading an update.
+    StartDownload(UpdateRequestId),
+    /// Start installing an update.
+    StartInstall(UpdateRequestId),
 
     /// Send a list of packages to the Core server.
     SendInstalledPackages(Vec<Package>),
@@ -40,7 +42,11 @@ pub enum Command {
 
 impl Display for Command {
     fn fmt(&self, f: &mut Formatter) -> FmtResult {
-        write!(f, "{:?}", self)
+        let text = match *self {
+            Command::SendInstalledPackages(_) => "SendInstalledPackages(...)".to_string(),
+            _                                 => format!("{:?}", self)
+        };
+        write!(f, "{}", text)
     }
 }
 
@@ -61,8 +67,10 @@ named!(command <(Command, Vec<&str>)>, chain!(
     ~ cmd: alt!(
         alt_complete!(tag!("Authenticate") | tag!("auth"))
             => { |_| Command::Authenticate(None) }
-        | alt_complete!(tag!("GetNewUpdates") | tag!("new"))
-            => { |_| Command::GetNewUpdates }
+        | alt_complete!(tag!("GetPendingUpdates") | tag!("getpend"))
+            => { |_| Command::GetPendingUpdates }
+        | alt_complete!(tag!("GetInFlightUpdates") | tag!("getflight"))
+            => { |_| Command::GetInFlightUpdates }
         | alt_complete!(tag!("ListInstalledPackages") | tag!("ls"))
             => { |_| Command::ListInstalledPackages }
         | alt_complete!(tag!("ListSystemInfo") | tag!("info"))
@@ -78,9 +86,9 @@ named!(command <(Command, Vec<&str>)>, chain!(
         | alt_complete!(tag!("SendUpdateReport") | tag!("sendup"))
             => { |_| Command::SendUpdateReport(UpdateReport::default()) }
         | alt_complete!(tag!("StartDownload") | tag!("dl"))
-            => { |_| Command::StartDownload(Vec::new()) }
+            => { |_| Command::StartDownload("".to_string()) }
         | alt_complete!(tag!("StartInstall") | tag!("inst"))
-            => { |_| Command::StartInstall(DownloadComplete::default()) }
+            => { |_| Command::StartInstall("".to_string()) }
     )
         ~ args: arguments
         ~ alt!(eof | tag!("\r") | tag!("\n") | tag!(";")),
@@ -112,9 +120,14 @@ fn parse_arguments(cmd: Command, args: Vec<&str>) -> Result<Command, Error> {
             _ => Err(Error::Command(format!("unexpected Authenticate args: {:?}", args))),
         },
 
-        Command::GetNewUpdates => match args.len() {
-            0 => Ok(Command::GetNewUpdates),
-            _ => Err(Error::Command(format!("unexpected GetNewUpdates args: {:?}", args))),
+        Command::GetPendingUpdates => match args.len() {
+            0 => Ok(Command::GetPendingUpdates),
+            _ => Err(Error::Command(format!("unexpected GetPendingUpdates args: {:?}", args))),
+        },
+
+        Command::GetInFlightUpdates => match args.len() {
+            0 => Ok(Command::GetInFlightUpdates),
+            _ => Err(Error::Command(format!("unexpected GetInFlightUpdates args: {:?}", args))),
         },
 
         Command::ListInstalledPackages => match args.len() {
@@ -170,12 +183,14 @@ fn parse_arguments(cmd: Command, args: Vec<&str>) -> Result<Command, Error> {
         },
 
         Command::StartDownload(_) => match args.len() {
-            0 => Err(Error::Command("usage: dl [<id>]".to_string())),
-            _ => Ok(Command::StartDownload(args.iter().map(|arg| String::from(*arg)).collect())),
+            0 => Err(Error::Command("usage: dl <id>".to_string())),
+            1 => Ok(Command::StartDownload(args[0].to_string())),
+            _ => Err(Error::Command(format!("unexpected StartInstall args: {:?}", args))),
         },
 
         Command::StartInstall(_) => match args.len() {
-            // FIXME(PRO-1160): args
+            0 => Err(Error::Command("usage: inst <id>".to_string())),
+            1 => Ok(Command::StartInstall(args[0].to_string())),
             _ => Err(Error::Command(format!("unexpected StartInstall args: {:?}", args))),
         },
 
@@ -196,7 +211,7 @@ mod tests {
         assert_eq!(command(&b"auth foo bar"[..]),
                    IResult::Done(&b""[..], (Command::Authenticate(None), vec!["foo", "bar"])));
         assert_eq!(command(&b"dl 1"[..]),
-                   IResult::Done(&b""[..], (Command::StartDownload(Vec::new()), vec!["1"])));
+                   IResult::Done(&b""[..], (Command::StartDownload("".to_string()), vec!["1"])));
         assert_eq!(command(&b"ls;\n"[..]),
                    IResult::Done(&b"\n"[..], (Command::ListInstalledPackages, Vec::new())));
     }
@@ -226,10 +241,17 @@ mod tests {
     }
 
     #[test]
-    fn get_new_updates_test() {
-        assert_eq!("GetNewUpdates".parse::<Command>().unwrap(), Command::GetNewUpdates);
-        assert_eq!("new".parse::<Command>().unwrap(), Command::GetNewUpdates);
-        assert!("new old".parse::<Command>().is_err());
+    fn get_pending_updates_test() {
+        assert_eq!("GetPendingUpdates".parse::<Command>().unwrap(), Command::GetPendingUpdates);
+        assert_eq!("getpend".parse::<Command>().unwrap(), Command::GetPendingUpdates);
+        assert!("getpend old".parse::<Command>().is_err());
+    }
+
+    #[test]
+    fn get_in_flight_updates_test() {
+        assert_eq!("GetInFlightUpdates".parse::<Command>().unwrap(), Command::GetInFlightUpdates);
+        assert_eq!("getflight".parse::<Command>().unwrap(), Command::GetInFlightUpdates);
+        assert!("getflight old".parse::<Command>().is_err());
     }
 
     #[test]
@@ -301,15 +323,16 @@ mod tests {
 
     #[test]
     fn start_download_test() {
-        assert_eq!("StartDownload this".parse::<Command>().unwrap(),
-                   Command::StartDownload(vec!["this".to_string()]));
-        assert_eq!("dl some more".parse::<Command>().unwrap(),
-                   Command::StartDownload(vec!["some".to_string(), "more".to_string()]));
+        assert_eq!("StartDownload this".parse::<Command>().unwrap(), Command::StartDownload("this".to_string()));
+        assert_eq!("dl that".parse::<Command>().unwrap(), Command::StartDownload("that".to_string()));
+        assert!("StartDownload this and that".parse::<Command>().is_err());
         assert!("dl".parse::<Command>().is_err());
     }
 
     #[test]
     fn start_install_test() {
+        assert_eq!("StartInstall 123".parse::<Command>().unwrap(), Command::StartInstall("123".to_string()));
+        assert_eq!("inst this".parse::<Command>().unwrap(), Command::StartInstall("this".to_string()));
         assert!("StartInstall".parse::<Command>().is_err());
         assert!("inst more than one".parse::<Command>().is_err());
     }
diff --git a/src/datatype/event.rs b/src/datatype/event.rs
index 63ae850..6a493f0 100644
--- a/src/datatype/event.rs
+++ b/src/datatype/event.rs
@@ -1,7 +1,7 @@
 use std::fmt::{Display, Formatter, Result as FmtResult};
 
-use datatype::{DownloadComplete, Package, UpdateAvailable, UpdateReport,
-               UpdateRequestId};
+use datatype::{DownloadComplete, Package, PendingUpdateRequest, UpdateAvailable,
+               UpdateReport, UpdateRequestId};
 
 
 /// System-wide events that are broadcast to all interested parties.
@@ -15,12 +15,17 @@ pub enum Event {
     /// An operation failed because we are not currently authenticated.
     NotAuthenticated,
 
-    /// There are new updates available.
-    NewUpdatesReceived(Vec<UpdateRequestId>),
-    /// A notification from RVI of a new update.
-    NewUpdateAvailable(UpdateAvailable),
-    /// There are no new updates available.
-    NoNewUpdates,
+    /// There are pending updates available.
+    PendingUpdatesReceived(Vec<UpdateRequestId>),
+    /// A notification from RVI of a pending update.
+    PendingUpdateAvailable(UpdateAvailable),
+    /// There are no pending updates.
+    NoPendingUpdates,
+
+    /// There are in-flight updates available.
+    InFlightUpdatesReceived(Vec<PendingUpdateRequest>),
+    /// There are no in-flight updates.
+    NoInFlightUpdates,
 
     /// The following packages are installed on the device.
     FoundInstalledPackages(Vec<Package>),
diff --git a/src/gateway/dbus.rs b/src/gateway/dbus.rs
index 3bd41ea..ea807f6 100644
--- a/src/gateway/dbus.rs
+++ b/src/gateway/dbus.rs
@@ -47,7 +47,7 @@ impl Gateway for DBus {
 
     fn pulse(&self, event: Event) {
         match event {
-            Event::NewUpdateAvailable(avail) => {
+            Event::PendingUpdateAvailable(avail) => {
                 let msg = self.new_message("updateAvailable", &[
                     MessageItem::from(avail.update_id),
                     MessageItem::from(avail.signature),
@@ -144,7 +144,7 @@ fn handle_initiate_download(itx: &Sender<Interpret>, msg: &mut Message) -> Metho
     let mut args = msg.get_items().into_iter();
     let arg_id   = try!(args.next().ok_or(dbus::missing_arg()));
     let update_id: &String = try!(FromMessageItem::from(&arg_id).or(Err(dbus::malformed_arg())));
-    send(itx, Command::StartDownload(vec![update_id.clone()]));
+    send(itx, Command::StartDownload(update_id.clone()));
 
     Ok(vec![])
 }
diff --git a/src/gateway/http.rs b/src/gateway/http.rs
index 990a1fc..6ccc2b5 100644
--- a/src/gateway/http.rs
+++ b/src/gateway/http.rs
@@ -107,9 +107,9 @@ mod tests {
             loop {
                 let interpret = irx.recv().expect("itx is closed");
                 match interpret.command {
-                    Command::StartDownload(ids) => {
+                    Command::StartDownload(id) => {
                         let tx = interpret.response_tx.unwrap();
-                        tx.lock().unwrap().send(Event::FoundSystemInfo(ids.first().unwrap().to_owned()));
+                        tx.lock().unwrap().send(Event::FoundSystemInfo(id));
                     }
                     _ => panic!("expected AcceptUpdates"),
                 }
@@ -119,7 +119,7 @@ mod tests {
         crossbeam::scope(|scope| {
             for id in 0..10 {
                 scope.spawn(move || {
-                    let cmd     = Command::StartDownload(vec!(format!("{}", id)));
+                    let cmd     = Command::StartDownload(format!("{}", id));
                     let client  = AuthClient::default();
                     let url     = "http://127.0.0.1:8888".parse().unwrap();
                     let body    = json::encode(&cmd).unwrap();
diff --git a/src/gateway/socket.rs b/src/gateway/socket.rs
index 295cf6e..af496c9 100644
--- a/src/gateway/socket.rs
+++ b/src/gateway/socket.rs
@@ -149,9 +149,9 @@ mod tests {
             loop {
                 let interpret = irx.recv().expect("gtx is closed");
                 match interpret.command {
-                    Command::StartDownload(ids) => {
+                    Command::StartDownload(id) => {
                         let tx = interpret.response_tx.unwrap();
-                        tx.lock().unwrap().send(Event::FoundSystemInfo(ids.first().unwrap().to_owned()));
+                        tx.lock().unwrap().send(Event::FoundSystemInfo(id));
                     }
                     _ => panic!("expected AcceptUpdates"),
                 }
diff --git a/src/gateway/websocket.rs b/src/gateway/websocket.rs
index eb5e040..72e0889 100644
--- a/src/gateway/websocket.rs
+++ b/src/gateway/websocket.rs
@@ -140,9 +140,9 @@ mod tests {
             loop {
                 let interpret = irx.recv().expect("gtx is closed");
                 match interpret.command {
-                    Command::StartDownload(ids) => {
+                    Command::StartDownload(id) => {
                         let tx = interpret.response_tx.unwrap();
-                        tx.lock().unwrap().send(Event::FoundSystemInfo(ids.first().unwrap().to_owned()));
+                        tx.lock().unwrap().send(Event::FoundSystemInfo(id));
                     }
                     _ => panic!("expected AcceptUpdates"),
                 }
diff --git a/src/interpreter.rs b/src/interpreter.rs
index 3495def..9797456 100644
--- a/src/interpreter.rs
+++ b/src/interpreter.rs
@@ -4,7 +4,7 @@ use std;
 use std::borrow::Cow;
 
 use datatype::{AccessToken, Auth, ClientId, ClientSecret, Command, Config,
-               Error, Event, Package, UpdateRequestId};
+               Error, Event, Package, UpdateReport, UpdateRequestId, UpdateResultCode};
 use gateway::Interpret;
 use http::{AuthClient, Client};
 use oauth2::authenticate;
@@ -51,17 +51,35 @@ impl Interpreter<Event, Command> for EventInterpreter {
                 ctx.send(Command::Authenticate(None));
             }
 
-            Event::NewUpdatesReceived(ids) => {
-                ctx.send(Command::StartDownload(ids));
+            Event::InFlightUpdatesReceived(requests) => {
+                if self.package_manager != PackageManager::Off {
+                    self.package_manager.installed_packages().map(|packages| {
+                        for request in requests {
+                            let id = request.requestId.clone();
+                            if packages.contains(&request.packageId) {
+                                let report = UpdateReport::single(id, UpdateResultCode::OK, "".to_string());
+                                ctx.send(Command::SendUpdateReport(report));
+                            } else {
+                                ctx.send(Command::StartDownload(id));
+                            }
+                        }
+                    }).unwrap_or_else(|err| error!("couldn't get a list of packages: {}", err));
+                }
+            }
+
+            Event::PendingUpdatesReceived(ids) => {
+                for id in ids {
+                    ctx.send(Command::StartDownload(id));
+                }
             }
 
             Event::DownloadComplete(dl) => {
                 if self.package_manager != PackageManager::Off {
-                    ctx.send(Command::StartInstall(dl));
+                    ctx.send(Command::StartInstall(dl.update_id.clone()));
                 }
             }
 
-            Event::InstallComplete(report) => {
+            Event::InstallComplete(report) | Event::InstallFailed(report) => {
                 ctx.send(Command::SendUpdateReport(report));
             }
 
@@ -149,14 +167,23 @@ impl<'t> GlobalInterpreter<'t> {
         match cmd {
             Command::Authenticate(_) => etx.send(Event::Authenticated),
 
-            Command::GetNewUpdates => {
+            Command::GetInFlightUpdates => {
+                let updates = try!(sota.get_in_flight_updates());
+                if updates.is_empty() {
+                    etx.send(Event::NoInFlightUpdates);
+                } else {
+                    etx.send(Event::InFlightUpdatesReceived(updates));
+                }
+            }
+
+            Command::GetPendingUpdates => {
                 let mut updates = try!(sota.get_pending_updates());
                 if updates.is_empty() {
-                    etx.send(Event::NoNewUpdates);
+                    etx.send(Event::NoPendingUpdates);
                 } else {
                     updates.sort_by_key(|u| u.installPos);
                     let ids = updates.iter().map(|u| u.requestId.clone()).collect::<Vec<UpdateRequestId>>();
-                    etx.send(Event::NewUpdatesReceived(ids))
+                    etx.send(Event::PendingUpdatesReceived(ids));
                 }
             }
 
@@ -203,22 +230,20 @@ impl<'t> GlobalInterpreter<'t> {
                 etx.send(Event::UpdateReportSent);
             }
 
-            Command::StartDownload(ref ids) => {
-                for id in ids {
-                    etx.send(Event::DownloadingUpdate(id.clone()));
-                    if let Some(ref rvi) = self.rvi {
-                        let _ = rvi.remote.lock().unwrap().send_download_started(id.clone());
-                    } else {
-                        let _ = sota.download_update(id.clone())
-                            .map(|dl| etx.send(Event::DownloadComplete(dl)))
-                            .map_err(|err| etx.send(Event::DownloadFailed(id.clone(), format!("{}", err))));
-                    }
+            Command::StartDownload(id) => {
+                etx.send(Event::DownloadingUpdate(id.clone()));
+                if let Some(ref rvi) = self.rvi {
+                    let _ = rvi.remote.lock().unwrap().send_download_started(id);
+                } else {
+                    let _ = sota.download_update(id.clone())
+                        .map(|dl| etx.send(Event::DownloadComplete(dl)))
+                        .map_err(|err| etx.send(Event::DownloadFailed(id, format!("{}", err))));
                 }
             }
 
-            Command::StartInstall(dl) => {
-                etx.send(Event::InstallingUpdate(dl.update_id.clone()));
-                let _ = sota.install_update(dl)
+            Command::StartInstall(id) => {
+                etx.send(Event::InstallingUpdate(id.clone()));
+                let _ = sota.install_update(id)
                     .map(|report| etx.send(Event::InstallComplete(report)))
                     .map_err(|report| etx.send(Event::InstallFailed(report)));
             }
@@ -242,7 +267,8 @@ impl<'t> GlobalInterpreter<'t> {
                 etx.send(Event::Authenticated);
             }
 
-            Command::GetNewUpdates            |
+            Command::GetInFlightUpdates       |
+            Command::GetPendingUpdates        |
             Command::ListInstalledPackages    |
             Command::ListSystemInfo           |
             Command::SendInstalledPackages(_) |
@@ -321,20 +347,14 @@ mod tests {
         let pkg_mgr    = PackageManager::new_tpm(true);
         let (ctx, erx) = new_interpreter(replies, pkg_mgr);
 
-        ctx.send(Command::StartDownload(vec!["1".to_string(), "2".to_string()]));
+        ctx.send(Command::StartDownload("1".to_string()));
         assert_rx(erx, &[
             Event::DownloadingUpdate("1".to_string()),
             Event::DownloadComplete(DownloadComplete {
                 update_id:    "1".to_string(),
                 update_image: "/tmp/1".to_string(),
                 signature:    "".to_string()
-            }),
-            Event::DownloadingUpdate("2".to_string()),
-            Event::DownloadComplete(DownloadComplete {
-                update_id:    "2".to_string(),
-                update_image: "/tmp/2".to_string(),
-                signature:    "".to_string()
-            }),
+            })
         ]);
     }
 
@@ -344,11 +364,7 @@ mod tests {
         let pkg_mgr    = PackageManager::new_tpm(true);
         let (ctx, erx) = new_interpreter(replies, pkg_mgr);
 
-        ctx.send(Command::StartInstall(DownloadComplete {
-            update_id:    "1".to_string(),
-            update_image: "/tmp/1".to_string(),
-            signature:    "".to_string()
-        }));
+        ctx.send(Command::StartInstall("1".to_string()));
         assert_rx(erx, &[
             Event::InstallingUpdate("1".to_string()),
             Event::InstallComplete(
@@ -363,11 +379,7 @@ mod tests {
         let pkg_mgr    = PackageManager::new_tpm(false);
         let (ctx, erx) = new_interpreter(replies, pkg_mgr);
 
-        ctx.send(Command::StartInstall(DownloadComplete {
-            update_id:    "1".to_string(),
-            update_image: "/tmp/1".to_string(),
-            signature:    "".to_string()
-        }));
+        ctx.send(Command::StartInstall("1".to_string()));
         assert_rx(erx, &[
             Event::InstallingUpdate("1".to_string()),
             Event::InstallFailed(
diff --git a/src/main.rs b/src/main.rs
index 3fd5667..3fd7797 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -51,7 +51,7 @@ fn start_update_poller(interval: u64, itx: Sender<Interpret>) {
     loop {
         let _ = tick.recv();
         itx.send(Interpret {
-            command:     Command::GetNewUpdates,
+            command:     Command::GetPendingUpdates,
             response_tx: Some(Arc::new(Mutex::new(etx.clone())))
         });
         let _ = erx.recv();
diff --git a/src/package_manager/deb.rs b/src/package_manager/deb.rs
index bba86e6..845c2b8 100644
--- a/src/package_manager/deb.rs
+++ b/src/package_manager/deb.rs
@@ -5,7 +5,7 @@ use package_manager::package_manager::{InstallOutcome, parse_package};
 
 
 /// Returns a list of installed DEB packages with
-/// `dpkg-query -f='${Package} ${Version}\n -W`.
+/// `dpkg-query -f='${Package} ${Version}\n' -W`.
 pub fn installed_packages() -> Result<Vec<Package>, Error> {
     Command::new("dpkg-query").arg("-f='${Package} ${Version}\n'").arg("-W")
         .output()
diff --git a/src/rvi/parameters.rs b/src/rvi/parameters.rs
index f48ea44..fc5d663 100644
--- a/src/rvi/parameters.rs
+++ b/src/rvi/parameters.rs
@@ -22,7 +22,7 @@ pub struct Notify {
 impl Parameter for Notify {
     fn handle(&self, remote: &Mutex<RemoteServices>, _: &Mutex<Transfers>) -> Result<Option<Event>, String> {
         remote.lock().unwrap().backend = Some(self.services.clone());
-        Ok(Some(Event::NewUpdateAvailable(self.update_available.clone())))
+        Ok(Some(Event::PendingUpdateAvailable(self.update_available.clone())))
     }
 }
 
diff --git a/src/sota.rs b/src/sota.rs
index 3dec33d..5759b9f 100644
--- a/src/sota.rs
+++ b/src/sota.rs
@@ -22,39 +22,43 @@ impl<'c, 'h> Sota<'c, 'h> {
     }
 
     /// Takes a path and returns a new endpoint of the format
-    /// `<Core server>/api/v1/device_updates/<uuid>/<path>`.
-    pub fn endpoint(&self, path: &str) -> Url {
-        let endpoint = if path.is_empty() {
-            format!("/api/v1/device_updates/{}", self.config.device.uuid)
-        } else {
-            format!("/api/v1/device_updates/{}/{}", self.config.device.uuid, path)
-        };
+    /// `<Core server>/api/v1/device_updates/<device-id>$path`.
+    fn endpoint(&self, path: &str) -> Url {
+        let endpoint = format!("/api/v1/device_updates/{}{}", self.config.device.uuid, path);
         self.config.core.server.join(&endpoint).expect("couldn't build endpoint url")
     }
 
-    /// Query the Core server to identify any new package updates available.
+    /// Returns the path to a package on the device.
+    fn package_path(&self, id: UpdateRequestId) -> Result<String, Error> {
+        let mut path = PathBuf::new();
+        path.push(&self.config.device.packages_dir);
+        path.push(id);
+        Ok(try!(path.to_str().ok_or(Error::Parse(format!("Path is not valid UTF-8: {:?}", path)))).to_string())
+    }
+
+    /// Query the Core server for any pending package updates.
     pub fn get_pending_updates(&mut self) -> Result<Vec<PendingUpdateRequest>, Error> {
         let resp_rx = self.client.get(self.endpoint(""), None);
-        let resp    = resp_rx.recv().expect("no get_package_updates response received");
-        let data    = try!(resp);
-        let text    = try!(String::from_utf8(data));
+        let resp    = try!(resp_rx.recv().ok_or(Error::Client("couldn't get pending updates".to_string())));
+        let text    = try!(String::from_utf8(try!(resp)));
+        Ok(try!(json::decode::<Vec<PendingUpdateRequest>>(&text)))
+    }
+
+    /// Query the Core server for any in-flight package updates.
+    pub fn get_in_flight_updates(&mut self) -> Result<Vec<PendingUpdateRequest>, Error> {
+        let resp_rx = self.client.get(self.endpoint("/queued"), None);
+        let resp    = try!(resp_rx.recv().ok_or(Error::Client("couldn't get in-flight updates".to_string())));
+        let text    = try!(String::from_utf8(try!(resp)));
         Ok(try!(json::decode::<Vec<PendingUpdateRequest>>(&text)))
     }
 
     /// Download a specific update from the Core server.
     pub fn download_update(&mut self, id: UpdateRequestId) -> Result<DownloadComplete, Error> {
-        let resp_rx = self.client.get(self.endpoint(&format!("{}/download", id)), None);
-        let resp    = resp_rx.recv().expect("no download_package_update response received");
-        let data    = try!(resp);
-
-        let mut path = PathBuf::new();
-        path.push(&self.config.device.packages_dir);
-        path.push(id.clone()); // TODO: Use Content-Disposition filename from request?
-        let mut file = try!(File::create(path.as_path()));
-
-        let _    = io::copy(&mut &*data, &mut file);
-        let path = try!(path.to_str().ok_or(Error::Parse(format!("Path is not valid UTF-8: {:?}", path))));
-
+        let resp_rx  = self.client.get(self.endpoint(&format!("/{}/download", id)), None);
+        let resp     = try!(resp_rx.recv().ok_or(Error::Client("couldn't download update".to_string())));
+        let path     = try!(self.package_path(id.clone()));
+        let mut file = try!(File::create(&path));
+        let _        = io::copy(&mut &*try!(resp), &mut file);
         Ok(DownloadComplete {
             update_id:    id,
             update_image: path.to_string(),
@@ -63,26 +67,22 @@ impl<'c, 'h> Sota<'c, 'h> {
     }
 
     /// Install an update using the package manager.
-    pub fn install_update(&mut self, download: DownloadComplete) -> Result<UpdateReport, UpdateReport> {
+    pub fn install_update(&mut self, id: UpdateRequestId) -> Result<UpdateReport, UpdateReport> {
         let ref pacman = self.config.device.package_manager;
-        pacman.install_package(&download.update_image).and_then(|(code, output)| {
-            Ok(UpdateReport::single(download.update_id.clone(), code, output))
+        let path       = self.package_path(id.clone()).expect("install_update expects a valid path");
+        pacman.install_package(&path).and_then(|(code, output)| {
+            Ok(UpdateReport::single(id.clone(), code, output))
         }).or_else(|(code, output)| {
-            Err(UpdateReport::single(download.update_id.clone(), code, output))
+            Err(UpdateReport::single(id.clone(), code, output))
         })
     }
 
-    /// Get a list of the currently installed packages from the package manager.
-    pub fn get_installed_packages(&mut self) -> Result<Vec<Package>, Error> {
-        Ok(try!(self.config.device.package_manager.installed_packages()))
-    }
-
     /// Send a list of the currently installed packages to the Core server.
     pub fn send_installed_packages(&mut self, packages: &Vec<Package>) -> Result<(), Error> {
         let body    = try!(json::encode(packages));
-        let resp_rx = self.client.put(self.endpoint("installed"), Some(body.into_bytes()));
-        let _       = resp_rx.recv().expect("no update_installed_packages response received")
-                             .map_err(|err| error!("update_installed_packages failed: {}", err));
+        let resp_rx = self.client.put(self.endpoint("/installed"), Some(body.into_bytes()));
+        let resp    = try!(resp_rx.recv().ok_or(Error::Client("couldn't send installed packages".to_string())));
+        let _       = resp.map_err(|err| error!("send_installed_packages failed: {}", err));
         Ok(())
     }
 
@@ -90,18 +90,18 @@ impl<'c, 'h> Sota<'c, 'h> {
     pub fn send_update_report(&mut self, update_report: &UpdateReport) -> Result<(), Error> {
         let report  = DeviceReport::new(&self.config.device.uuid, update_report);
         let body    = try!(json::encode(&report));
-        let url     = self.endpoint(report.device);
+        let url     = self.endpoint(&format!("/{}", report.device));
         let resp_rx = self.client.post(url, Some(body.into_bytes()));
-        let resp    = resp_rx.recv().expect("no send_install_report response received");
-        let _       = try!(resp);
+        let resp    = try!(resp_rx.recv().ok_or(Error::Client("couldn't send update report".to_string())));
+        let _       = resp.map_err(|err| error!("send_update_report failed: {}", err));
         Ok(())
     }
 
     /// Send system information from the device to the Core server.
     pub fn send_system_info(&mut self, body: &str) -> Result<(), Error> {
-        let resp_rx = self.client.put(self.endpoint("system_info"), Some(body.as_bytes().to_vec()));
-        let resp    = resp_rx.recv().expect("no send_system_info response received");
-        let _       = try!(resp);
+        let resp_rx = self.client.put(self.endpoint("/system_info"), Some(body.as_bytes().to_vec()));
+        let resp    = try!(resp_rx.recv().ok_or(Error::Client("couldn't send system info".to_string())));
+        let _       = resp.map_err(|err| error!("send_system_info failed: {}", err));
         Ok(())
     }
 }