Introduce TransportWriter

yuriks pointed out that the std::io::Write trait is inadequate for the
contract that users of Transport::put must abide by, so this commit
introduces a new trait, TransportWriter, which extends Write with
explicit complete_upload and cancel_upload methods. This commit also
resolves a bug with idle connection timeouts in Rusoto and Hyper.

facilitator/Cargo.toml

@@ -12,12 +12,13 @@ base64 = "0.12.3"
 chrono = "0.4"
 clap = "2.33.3"
 derivative = "2.1.1"
+hyper = "0.13.8"
+hyper-rustls = "0.21.0"
 prio = "0.2"
 rand = "0.7"
 ring = { version = "0.16.15", features = ["std"] }
-rusoto_core = { version="0.45.0", default_features=false, features=["rustls"] }
-rusoto_mock = { version="0.45.0", default_features=false, features=["rustls"] }
-rusoto_s3 = { version="0.45.0", default_features=false, features=["rustls"] }
+rusoto_core = { version = "0.45.0", default_features = false, features = ["rustls"] }
+rusoto_s3 = { version = "0.45.0", default_features = false, features = ["rustls"] }
 serde = { version = "1.0", features = ["derive"] }
 tempfile = "3.1.0"
 thiserror = "1.0"
@@ -26,3 +27,6 @@ uuid = { version = "0.8", features = ["serde", "v4"] }
 
 [build-dependencies]
 vergen = "3"
+
+[dev-dependencies]
+rusoto_mock = { version = "0.45.0", default_features = false, features = ["rustls"] }

facilitator/src/batch.rs

@@ -1,6 +1,6 @@
 use crate::{
     idl::{Header, Packet},
-    transport::Transport,
+    transport::{Transport, TransportWriter},
     DigestWriter, SidecarWriter, DATE_FORMAT,
 };
 use anyhow::{anyhow, Context, Result};
@@ -51,13 +51,10 @@ impl Batch {
         is_first: bool,
     ) -> Batch {
         let batch_path = format!(
-            "{}/{}",
+            "{}/{}-{}",
             aggregation_name,
-            format!(
-                "{}-{}",
-                aggregation_start.format(DATE_FORMAT),
-                aggregation_end.format(DATE_FORMAT)
-            )
+            aggregation_start.format(DATE_FORMAT),
+            aggregation_end.format(DATE_FORMAT)
         );
         let filename = format!("sum_{}", if is_first { 0 } else { 1 });
 
@@ -258,7 +255,10 @@ impl<'a, H: Header, P: Packet> BatchWriter<'a, H, P> {
         let mut sidecar_writer =
             SidecarWriter::new(self.transport.put(self.batch.header_key())?, Vec::new());
         header.write(&mut sidecar_writer)?;
-        sidecar_writer.flush().context("failed to flush writer")?;
+        sidecar_writer
+            .writer
+            .complete_upload()
+            .context("failed to complete batch header upload")?;
 
         let header_signature = key
             .sign(&SystemRandom::new(), &sidecar_writer.sidecar)
@@ -274,7 +274,7 @@ impl<'a, H: Header, P: Packet> BatchWriter<'a, H, P> {
     /// content written by the operation.
     pub fn packet_file_writer<F>(&mut self, operation: F) -> Result<Digest>
     where
-        F: FnOnce(&mut Writer<SidecarWriter<Box<dyn Write>, DigestWriter>>) -> Result<()>,
+        F: FnOnce(&mut Writer<SidecarWriter<Box<dyn TransportWriter>, DigestWriter>>) -> Result<()>,
     {
         let mut writer = Writer::new(
             &self.packet_schema,
@@ -284,10 +284,23 @@ impl<'a, H: Header, P: Packet> BatchWriter<'a, H, P> {
             ),
         );
 
-        operation(&mut writer)?;
+        let result = operation(&mut writer);
+        let mut sidecar_writer = writer
+            .into_inner()
+            .with_context(|| format!("failed to flush Avro writer ({:?})", result))?;
+
+        if let Err(e) = result {
+            sidecar_writer
+                .writer
+                .cancel_upload()
+                .with_context(|| format!("Encountered while handling: {}", e))?;
+            return Err(e);
+        }
 
-        let mut sidecar_writer = writer.into_inner().context("failed to flush Avro writer")?;
-        sidecar_writer.flush().context("failed to flush writer")?;
+        sidecar_writer
+            .writer
+            .complete_upload()
+            .context("failed to complete packet file upload")?;
         Ok(sidecar_writer.sidecar.finish())
     }
 
@@ -298,7 +311,9 @@ impl<'a, H: Header, P: Packet> BatchWriter<'a, H, P> {
         writer
             .write_all(signature.as_ref())
             .context("failed to write signature")?;
-        writer.flush().context("failed to flush signature")
+        writer
+            .complete_upload()
+            .context("failed to complete signature upload")
     }
 }
 
@@ -391,7 +406,7 @@ mod tests {
         // Verify file layout is as expected
         for extension in filenames {
             transport
-                .get(format!("{}.{}", base_path, extension).as_ref())
+                .get(&format!("{}.{}", base_path, extension))
                 .expect(&format!("could not get batch file {}", extension));
         }