Merge pull request #119 from luser/handle-storage-get-failure

Handle storage get failure

src/compiler/compiler.rs

@@ -147,15 +147,15 @@ pub trait CompilerHasher<T>: fmt::Debug + Send + 'static
             });
 
             // Check the result of the cache lookup.
-            Box::new(cache_status.and_then(move |result| {
+            Box::new(cache_status.then(move |result| {
                 let duration = start.elapsed();
                 let pwd = Path::new(&cwd);
                 let outputs = compilation.outputs()
                     .map(|(key, path)| (key.to_string(), pwd.join(path)))
                     .collect::<HashMap<_, _>>();
 
                 let miss_type = match result {
-                    Some(Cache::Hit(mut entry)) => {
+                    Ok(Some(Cache::Hit(mut entry))) => {
                         debug!("[{}]: Cache hit in {}", out_file, fmt_duration_as_secs(&duration));
                         let mut stdout = io::Cursor::new(vec!());
                         let mut stderr = io::Cursor::new(vec!());
@@ -181,18 +181,25 @@ pub trait CompilerHasher<T>: fmt::Debug + Send + 'static
                             (result, output)
                         })) as SFuture<_>
                     }
-                    Some(Cache::Miss) => {
+                    Ok(Some(Cache::Miss)) => {
                         debug!("[{}]: Cache miss", out_file);
                         MissType::Normal
                     }
-                    Some(Cache::Recache) => {
+                    Ok(Some(Cache::Recache)) => {
                         debug!("[{}]: Cache recache", out_file);
                         MissType::ForcedRecache
                     }
-                    None => {
+                    Ok(None) => {
                         debug!("[{}]: Cache timed out", out_file);
                         MissType::TimedOut
                     }
+                    Err(err) => {
+                        error!("[{}]: Cache read error: {}", out_file, err);
+                        for e in err.iter() {
+                            error!("[{:?}] \t{}", out_file, e);
+                        }
+                        MissType::CacheReadError
+                    }
                 };
 
                 // Cache miss, so compile it.
@@ -315,6 +322,8 @@ pub enum MissType {
     ForcedRecache,
     /// Cache took too long to respond.
     TimedOut,
+    /// Error reading from cache
+    CacheReadError,
 }
 
 /// Information about a successful cache write.
@@ -575,7 +584,7 @@ pub fn get_compiler_info<T>(creator: &T, executable: &str, pool: &CpuPool)
 #[cfg(test)]
 mod test {
     use super::*;
-    use cache::Storage;
+    use cache::{CacheWrite,Storage};
     use cache::disk::DiskCache;
     use futures::Future;
     use futures_cpupool::CpuPool;
@@ -585,6 +594,7 @@ mod test {
     use std::sync::Arc;
     use std::time::Duration;
     use std::usize;
+    use test::mock_storage::MockStorage;
     use test::utils::*;
     use tokio_core::reactor::Core;
 
@@ -841,6 +851,72 @@ mod test {
         assert_eq!(COMPILER_STDERR, res.stderr.as_slice());
     }
 
+    #[test]
+    /// Test that a cache read that results in an error is treated as a cache
+    /// miss.
+    fn test_compiler_get_cached_or_compile_cache_error() {
+        use env_logger;
+        drop(env_logger::init());
+        let creator = new_creator();
+        let f = TestFixture::new();
+        let pool = CpuPool::new(1);
+        let core = Core::new().unwrap();
+        let handle = core.handle();
+        let storage = MockStorage::new();
+        let storage: Arc<MockStorage> = Arc::new(storage);
+        // Pretend to be GCC.
+        next_command(&creator, Ok(MockChild::new(exit_status(0), "gcc", "")));
+        let c = get_compiler_info(&creator,
+                                  f.bins[0].to_str().unwrap(),
+                                  &pool).wait().unwrap();
+        // The preprocessor invocation.
+        next_command(&creator, Ok(MockChild::new(exit_status(0), "preprocessor output", "")));
+        // The compiler invocation.
+        const COMPILER_STDOUT : &'static [u8] = b"compiler stdout";
+        const COMPILER_STDERR : &'static [u8] = b"compiler stderr";
+        let obj = f.tempdir.path().join("foo.o");
+        let o = obj.clone();
+        next_command_calls(&creator, move |_| {
+            // Pretend to compile something.
+            match File::create(&o)
+                .and_then(|mut f| f.write_all(b"file contents")) {
+                    Ok(_) => Ok(MockChild::new(exit_status(0), COMPILER_STDOUT, COMPILER_STDERR)),
+                    Err(e) => Err(e),
+                }
+        });
+        let cwd = f.tempdir.path().to_str().unwrap().to_string();
+        let arguments = stringvec!["-c", "foo.c", "-o", "foo.o"];
+        let hasher = match c.parse_arguments(&arguments, ".".as_ref()) {
+            CompilerArguments::Ok(h) => h,
+            o @ _ => panic!("Bad result from parse_arguments: {:?}", o),
+        };
+        // The cache will return an error.
+        storage.next_get(f_err("Some Error"));
+        // Storing the result should be OK though.
+        storage.next_put(Ok(CacheWrite::new()));
+        let (cached, res) = hasher.get_cached_or_compile(creator.clone(),
+                                                         storage.clone(),
+                                                         arguments.clone(),
+                                                         cwd.clone(),
+                                                         vec![],
+                                                         CacheControl::Default,
+                                                         pool.clone(),
+                                                         handle.clone()).wait().unwrap();
+        // Ensure that the object file was created.
+        assert_eq!(true, fs::metadata(&obj).and_then(|m| Ok(m.len() > 0)).unwrap());
+        match cached {
+            CompileResult::CacheMiss(MissType::CacheReadError, _, f) => {
+                // wait on cache write future so we don't race with it!
+                f.wait().unwrap();
+            }
+            _ => assert!(false, "Unexpected compile result: {:?}", cached),
+        }
+
+        assert_eq!(exit_status(0), res.status);
+        assert_eq!(COMPILER_STDOUT, res.stdout.as_slice());
+        assert_eq!(COMPILER_STDERR, res.stderr.as_slice());
+    }
+
     #[test]
     fn test_compiler_get_cached_or_compile_force_recache() {
         use env_logger;

src/server.rs

@@ -548,18 +548,18 @@ impl<C> SccacheService<C>
                         },
                         CompileResult::CacheMiss(miss_type, duration, future) => {
                             match miss_type {
-                                MissType::Normal => {
-                                    stats.cache_misses += 1;
-                                }
+                                MissType::Normal => {}
                                 MissType::ForcedRecache => {
-                                    stats.cache_misses += 1;
                                     stats.forced_recaches += 1;
                                 }
                                 MissType::TimedOut => {
-                                    stats.cache_misses += 1;
                                     stats.cache_timeouts += 1;
                                 }
+                                MissType::CacheReadError => {
+                                    stats.cache_errors += 1;
+                                }
                             }
+                            stats.cache_misses += 1;
                             stats.cache_read_miss_duration += duration;
                             cache_write = Some(future);
                         }
@@ -656,6 +656,8 @@ pub struct ServerStats {
     pub cache_misses: u64,
     /// The count of cache misses because the cache took too long to respond.
     pub cache_timeouts: u64,
+    /// The count of errors reading cache entries.
+    pub cache_read_errors: u64,
     /// The count of compilations which were successful but couldn't be cached.
     pub non_cacheable_compilations: u64,
     /// The count of compilations which forcibly ignored the cache.
@@ -695,6 +697,7 @@ impl Default for ServerStats {
             cache_hits: u64::default(),
             cache_misses: u64::default(),
             cache_timeouts: u64::default(),
+            cache_read_errors: u64::default(),
             non_cacheable_compilations: u64::default(),
             forced_recaches: u64::default(),
             cache_write_errors: u64::default(),
@@ -738,6 +741,7 @@ impl ServerStats {
         set_stat!(stats_vec, self.cache_hits, "Cache hits");
         set_stat!(stats_vec, self.cache_misses, "Cache misses");
         set_stat!(stats_vec, self.cache_timeouts, "Cache timeouts");
+        set_stat!(stats_vec, self.cache_read_errors, "Cache read errors");
         set_stat!(stats_vec, self.forced_recaches, "Forced recaches");
         set_stat!(stats_vec, self.cache_write_errors, "Cache write errors");
         set_stat!(stats_vec, self.compile_fails, "Compilation failures");

src/simples3/s3.rs

@@ -77,20 +77,32 @@ impl Bucket {
     pub fn get(&self, key: &str) -> SFuture<Vec<u8>> {
         let url = format!("{}{}", self.base_url, key);
         debug!("GET {}", url);
+        let url2 = url.clone();
         Box::new(self.client.get(url.parse().unwrap()).chain_err(move || {
             format!("failed GET: {}", url)
         }).and_then(|res| {
             if res.status().class() == hyper::status::StatusClass::Success {
-                Ok(res.body())
+                let content_length = res.headers().get::<header::ContentLength>()
+                    .map(|&header::ContentLength(len)| len);
+                Ok((res.body(), content_length))
             } else {
                 Err(ErrorKind::BadHTTPStatus(res.status().clone()).into())
             }
-        }).and_then(|body| {
+        }).and_then(|(body, content_length)| {
             body.fold(Vec::new(), |mut body, chunk| {
                 body.extend_from_slice(&chunk);
                 Ok::<_, hyper::Error>(body)
             }).chain_err(|| {
                 "failed to read HTTP body"
+            }).and_then(move |bytes| {
+                if let Some(len) = content_length {
+                    if len != bytes.len() as u64 {
+                        bail!(format!("Bad HTTP body size read: {}, expected {}", bytes.len(), len));
+                    } else {
+                        info!("Read {} bytes from {}", bytes.len(), url2);
+                    }
+                }
+                Ok(bytes)
             })
         }))
     }

src/test/mock_storage.rs

@@ -0,0 +1,63 @@
+// Copyright 2017 Mozilla Foundation
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use cache::{Cache,CacheWrite,Storage};
+use errors::*;
+use std::cell::RefCell;
+use std::time::Duration;
+
+/// A mock `Storage` implementation.
+pub struct MockStorage {
+    gets: RefCell<Vec<SFuture<Cache>>>,
+    puts: RefCell<Vec<Result<CacheWrite>>>,
+}
+
+impl MockStorage {
+    /// Create a new `MockStorage`.
+    pub fn new() -> MockStorage {
+        MockStorage {
+            gets: RefCell::new(vec![]),
+            puts: RefCell::new(vec![]),
+        }
+    }
+
+    /// Queue up `res` to be returned as the next result from `Storage::get`.
+    pub fn next_get(&self, res: SFuture<Cache>) {
+        self.gets.borrow_mut().push(res)
+    }
+
+    /// Queue up `res` to be returned as the next result from `Storage::start_put`.
+    pub fn next_put(&self, res: Result<CacheWrite>) {
+        self.puts.borrow_mut().push(res)
+    }
+}
+
+impl Storage for MockStorage {
+    fn get(&self, _key: &str) -> SFuture<Cache> {
+        let mut g = self.gets.borrow_mut();
+        assert!(g.len() > 0, "MockStorage get called, but no get results available");
+        g.remove(0)
+    }
+    fn start_put(&self, _key: &str) -> Result<CacheWrite> {
+        let mut p = self.puts.borrow_mut();
+        assert!(p.len() > 0, "MockStorage start_put called, but no put results available");
+        p.remove(0)
+    }
+    fn finish_put(&self, _key: &str, _entry: CacheWrite) -> SFuture<Duration> {
+        f_ok(Duration::from_secs(0))
+    }
+    fn location(&self) -> String { "Mock Storage".to_string() }
+    fn current_size(&self) -> Option<usize> { None }
+    fn max_size(&self) -> Option<usize> { None }
+}

src/test/mod.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod mock_storage;
 #[macro_use]
 pub mod utils;
 mod tests;