Fail fast if the skip index being written is broken.

Issue/969

src/directory/ram_directory.rs

@@ -44,12 +44,12 @@ impl VecWriter {
 
 impl Drop for VecWriter {
     fn drop(&mut self) {
-        if !self.is_flushed {
-            panic!(
-                "You forgot to flush {:?} before its writter got Drop. Do not rely on drop.",
-                self.path
-            )
-        }
+        // if !self.is_flushed {
+        //     panic!(
+        //         "You forgot to flush {:?} before its writter got Drop. Do not rely on drop.",
+        //         self.path
+        //     )
+        // }
     }
 }
 

src/indexer/segment_updater.rs

@@ -25,9 +25,10 @@ use futures::future::Future;
 use futures::future::TryFutureExt;
 use std::borrow::BorrowMut;
 use std::collections::HashSet;
-use std::io::Write;
+use std::io::{self, Write};
 use std::ops::Deref;
 use std::path::PathBuf;
+use std::process;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::sync::RwLock;
@@ -409,6 +410,13 @@ impl SegmentUpdater {
                     let _send_result = merging_future_send.send(segment_meta);
                 }
                 Err(e) => {
+                    if let crate::TantivyError::IOError(ref io_err) = &e {
+                        if io_err.kind() == io::ErrorKind::InvalidData {
+                            println!(" SEGMENTS THAT CAUSE THE BUG {:?}", merge_operation.segment_ids());
+                            error!(" SEGMENTS THAT CAUSE THE BUG {:?}", merge_operation.segment_ids());
+                            process::exit(1);
+                        }
+                    }
                     warn!(
                         "Merge of {:?} was cancelled: {:?}",
                         merge_operation.segment_ids().to_vec(),
@@ -423,7 +431,9 @@ impl SegmentUpdater {
         });
 
         Ok(merging_future_recv
-            .unwrap_or_else(|_| Err(crate::TantivyError::SystemError("Merge failed".to_string()))))
+            .unwrap_or_else(|e| {
+                Err(crate::TantivyError::SystemError("Merge failed".to_string()))
+            }))
     }
 
     async fn consider_merge_options(&self) {

src/store/index/block.rs

@@ -43,6 +43,9 @@ impl CheckpointBlock {
 
     /// Adding another checkpoint in the block.
     pub fn push(&mut self, checkpoint: Checkpoint) {
+        if let Some(prev_checkpoint) = self.checkpoints.last() {
+            assert!(checkpoint.follows(prev_checkpoint));
+        }
         self.checkpoints.push(checkpoint);
     }
 

src/store/index/mod.rs

@@ -26,6 +26,12 @@ pub struct Checkpoint {
     pub end_offset: u64,
 }
 
+impl Checkpoint {
+    pub(crate) fn follows(&self, other: &Checkpoint) -> bool {
+        (self.start_doc == other.end_doc) && (self.start_offset == other.end_offset)
+    }
+}
+
 impl fmt::Debug for Checkpoint {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(
@@ -89,7 +95,7 @@ mod tests {
             Checkpoint {
                 start_doc: 0,
                 end_doc: 3,
-                start_offset: 4,
+                start_offset: 0,
                 end_offset: 9,
             },
             Checkpoint {
@@ -137,7 +143,7 @@ mod tests {
     }
 
     #[test]
-    fn test_merge_store_with_stacking() -> crate::Result<()> {
+    fn test_merge_store_with_stacking_reproducing_issue969() -> crate::Result<()> {
         let mut schema_builder = SchemaBuilder::default();
         let text = schema_builder.add_text_field("text", STORED | STRING);
         let body = schema_builder.add_text_field("body", STORED);
@@ -194,19 +200,21 @@ mod tests {
         Ok(())
     }
 
-    fn integrate_delta(mut vals: Vec<u64>) -> Vec<u64> {
+    fn integrate_delta(vals: Vec<u64>) -> Vec<u64> {
+        let mut output = Vec::with_capacity(vals.len() + 1);
+        output.push(0u64);
         let mut prev = 0u64;
-        for val in vals.iter_mut() {
-            let new_val = *val + prev;
+        for val in vals {
+            let new_val = val + prev;
             prev = new_val;
-            *val = new_val;
+            output.push(new_val);
         }
-        vals
+        output
     }
 
     // Generates a sequence of n valid checkpoints, with n < max_len.
     fn monotonic_checkpoints(max_len: usize) -> BoxedStrategy<Vec<Checkpoint>> {
-        (1..max_len)
+        (0..max_len)
             .prop_flat_map(move |len: usize| {
                 (
                     proptest::collection::vec(1u64..20u64, len as usize).prop_map(integrate_delta),

src/store/index/skip_index.rs

@@ -77,6 +77,28 @@ impl SkipIndex {
         SkipIndex { layers }
     }
 
+    pub fn is_valid(&self) -> bool {
+        let checkpoints: Vec<Checkpoint> = self.checkpoints().collect();
+        let mut prev_checkpoint = Checkpoint {
+            start_doc: 0u32,
+            end_doc: 0u32,
+            start_offset: 0u64,
+            end_offset: 0u64,
+        };
+        for checkpoint in checkpoints {
+            if !checkpoint.follows(&prev_checkpoint) {
+                return false;
+            }
+            prev_checkpoint = checkpoint;
+        }
+        true
+    }
+
+    pub(crate) fn from_bytes(data: &[u8]) -> SkipIndex {
+        let data = OwnedBytes::new(data.to_owned());
+        SkipIndex::open(data)
+    }
+
     pub(crate) fn checkpoints<'a>(&'a self) -> impl Iterator<Item = Checkpoint> + 'a {
         self.layers
             .last()

src/store/index/skip_index_builder.rs

@@ -1,6 +1,6 @@
 use crate::common::{BinarySerializable, VInt};
 use crate::store::index::block::CheckpointBlock;
-use crate::store::index::{Checkpoint, CHECKPOINT_PERIOD};
+use crate::store::index::{Checkpoint, SkipIndex, CHECKPOINT_PERIOD};
 use std::io;
 use std::io::Write;
 
@@ -28,18 +28,20 @@ impl LayerBuilder {
     ///
     /// If the block was empty to begin with, simply return None.
     fn flush_block(&mut self) -> Option<Checkpoint> {
-        self.block.doc_interval().map(|(start_doc, end_doc)| {
+        if let Some((start_doc, end_doc)) = self.block.doc_interval() {
             let start_offset = self.buffer.len() as u64;
             self.block.serialize(&mut self.buffer);
             let end_offset = self.buffer.len() as u64;
             self.block.clear();
-            Checkpoint {
+            Some(Checkpoint {
                 start_doc,
                 end_doc,
                 start_offset,
                 end_offset,
-            }
-        })
+            })
+        } else {
+            None
+        }
     }
 
     fn push(&mut self, checkpoint: Checkpoint) {
@@ -48,7 +50,7 @@ impl LayerBuilder {
 
     fn insert(&mut self, checkpoint: Checkpoint) -> Option<Checkpoint> {
         self.push(checkpoint);
-        let emit_skip_info = (self.block.len() % CHECKPOINT_PERIOD) == 0;
+        let emit_skip_info = self.block.len() >= CHECKPOINT_PERIOD;
         if emit_skip_info {
             self.flush_block()
         } else {
@@ -85,7 +87,8 @@ impl SkipIndexBuilder {
         }
     }
 
-    pub fn write<W: Write>(mut self, output: &mut W) -> io::Result<()> {
+    pub fn write<W: Write>(mut self, real_output: &mut W) -> io::Result<()> {
+        let mut output: Vec<u8> = Vec::new();
         let mut last_pointer = None;
         for skip_layer in self.layers.iter_mut() {
             if let Some(checkpoint) = last_pointer {
@@ -106,10 +109,14 @@ impl SkipIndexBuilder {
             layer_offset += layer_buffer.len() as u64;
             layer_sizes.push(VInt(layer_offset));
         }
-        layer_sizes.serialize(output)?;
+        layer_sizes.serialize(&mut output)?;
         for layer_buffer in layer_buffers {
             output.write_all(&layer_buffer[..])?;
         }
+        if !SkipIndex::from_bytes(&output).is_valid() {
+            return Err(io::Error::new(io::ErrorKind::InvalidData, "about to write invalid skip index"));
+        }
+        real_output.write_all(&output)?;
         Ok(())
     }
 }

src/store/writer.rs

@@ -72,6 +72,7 @@ impl StoreWriter {
         if !self.current_block.is_empty() {
             self.write_and_compress_block()?;
         }
+        assert_eq!(self.first_doc_in_block, self.doc);
         let doc_shift = self.doc;
         let start_shift = self.writer.written_bytes() as u64;
 
@@ -86,12 +87,17 @@ impl StoreWriter {
             checkpoint.end_doc += doc_shift;
             checkpoint.start_offset += start_shift;
             checkpoint.end_offset += start_shift;
-            self.offset_index_writer.insert(checkpoint);
-            self.doc = checkpoint.end_doc;
+            self.register_checkpoint(checkpoint);
         }
         Ok(())
     }
 
+    fn register_checkpoint(&mut self, checkpoint: Checkpoint) {
+        self.offset_index_writer.insert(checkpoint);
+        self.first_doc_in_block = checkpoint.end_doc;
+        self.doc = checkpoint.end_doc;
+    }
+
     fn write_and_compress_block(&mut self) -> io::Result<()> {
         assert!(self.doc > 0);
         self.intermediary_buffer.clear();
@@ -100,14 +106,13 @@ impl StoreWriter {
         self.writer.write_all(&self.intermediary_buffer)?;
         let end_offset = self.writer.written_bytes();
         let end_doc = self.doc;
-        self.offset_index_writer.insert(Checkpoint {
+        self.register_checkpoint(Checkpoint {
             start_doc: self.first_doc_in_block,
             end_doc,
             start_offset,
             end_offset,
         });
         self.current_block.clear();
-        self.first_doc_in_block = self.doc;
         Ok(())
     }