add checksum check in ManagedDirectory

CHANGELOG.md

@@ -8,6 +8,7 @@ Tantivy 0.11.0
 - Closes #498 - add support for Elastic-style unbounded range queries for alphanumeric types eg. "title:>hello", "weight:>=70.5", "height:<200" (@petr-tik)
 - API change around `Box<BoxableTokenizer>`. See detail in #629
 - Avoid rebuilding Regex automaton whenever a regex query is reused. #630 (@brainlock)
+- Add footer with some metadata to index files. #605 (@fdb-hiroshima)
 
 ## How to update?
 

Cargo.toml

@@ -15,6 +15,7 @@ edition = "2018"
 [dependencies]
 base64 = "0.10.0"
 byteorder = "1.0"
+crc32fast = "1.2.0"
 once_cell = "1.0"
 regex = "1.0"
 tantivy-fst = "0.1"

src/common/composite_file.rs

@@ -2,7 +2,7 @@ use crate::common::BinarySerializable;
 use crate::common::CountingWriter;
 use crate::common::VInt;
 use crate::directory::ReadOnlySource;
-use crate::directory::WritePtr;
+use crate::directory::{TerminatingWrite, WritePtr};
 use crate::schema::Field;
 use crate::space_usage::FieldUsage;
 use crate::space_usage::PerFieldSpaceUsage;
@@ -42,7 +42,7 @@ pub struct CompositeWrite<W = WritePtr> {
     offsets: HashMap<FileAddr, u64>,
 }
 
-impl<W: Write> CompositeWrite<W> {
+impl<W: TerminatingWrite + Write> CompositeWrite<W> {
     /// Crate a new API writer that writes a composite file
     /// in a given write.
     pub fn wrap(w: W) -> CompositeWrite<W> {
@@ -91,8 +91,7 @@ impl<W: Write> CompositeWrite<W> {
 
         let footer_len = (self.write.written_bytes() - footer_offset) as u32;
         footer_len.serialize(&mut self.write)?;
-        self.write.flush()?;
-        Ok(())
+        self.write.terminate()
     }
 }
 

src/common/counting_writer.rs

@@ -1,3 +1,5 @@
+use crate::directory::AntiCallToken;
+use crate::directory::TerminatingWrite;
 use std::io;
 use std::io::Write;
 
@@ -42,6 +44,13 @@ impl<W: Write> Write for CountingWriter<W> {
     }
 }
 
+impl<W: TerminatingWrite> TerminatingWrite for CountingWriter<W> {
+    fn terminate_ref(&mut self, token: AntiCallToken) -> io::Result<()> {
+        self.flush()?;
+        self.underlying.terminate_ref(token)
+    }
+}
+
 #[cfg(test)]
 mod test {
 

src/core/index.rs

@@ -26,9 +26,10 @@ use crate::IndexWriter;
 use crate::Result;
 use num_cpus;
 use std::borrow::BorrowMut;
+use std::collections::HashSet;
 use std::fmt;
 #[cfg(feature = "mmap")]
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
 fn load_metas(directory: &dyn Directory, inventory: &SegmentMetaInventory) -> Result<IndexMeta> {
@@ -368,6 +369,11 @@ impl Index {
             .map(SegmentMeta::id)
             .collect())
     }
+
+    /// Returns the set of corrupted files
+    pub fn validate_checksum(&self) -> Result<HashSet<PathBuf>> {
+        self.directory.list_damaged().map_err(Into::into)
+    }
 }
 
 impl fmt::Debug for Index {

src/directory/directory.rs

@@ -118,6 +118,8 @@ pub trait Directory: DirectoryClone + fmt::Debug + Send + Sync + 'static {
     ///
     /// Specifically, subsequent writes or flushes should
     /// have no effect on the returned `ReadOnlySource` object.
+    ///
+    /// You should only use this to read files create with [`open_write`]
     fn open_read(&self, path: &Path) -> result::Result<ReadOnlySource, OpenReadError>;
 
     /// Removes a file
@@ -157,6 +159,8 @@ pub trait Directory: DirectoryClone + fmt::Debug + Send + Sync + 'static {
     /// atomic_write.
     ///
     /// This should only be used for small files.
+    ///
+    /// You should only use this to read files create with [`atomic_write`]
     fn atomic_read(&self, path: &Path) -> Result<Vec<u8>, OpenReadError>;
 
     /// Atomically replace the content of a file with data.

src/directory/footer.rs

@@ -0,0 +1,213 @@
+use crate::directory::read_only_source::ReadOnlySource;
+use crate::directory::{AntiCallToken, TerminatingWrite};
+use byteorder::{ByteOrder, LittleEndian};
+use crc32fast::Hasher;
+use std::io;
+use std::io::Write;
+
+const COMMON_FOOTER_SIZE: usize = 4 * 5;
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct Footer {
+    pub tantivy_version: (u32, u32, u32),
+    pub meta: String,
+    pub versioned_footer: VersionedFooter,
+}
+
+impl Footer {
+    pub fn new(versioned_footer: VersionedFooter) -> Self {
+        let tantivy_version = (
+            env!("CARGO_PKG_VERSION_MAJOR").parse().unwrap(),
+            env!("CARGO_PKG_VERSION_MINOR").parse().unwrap(),
+            env!("CARGO_PKG_VERSION_PATCH").parse().unwrap(),
+        );
+        Footer {
+            tantivy_version,
+            meta: format!(
+                "tantivy {}.{}.{}, index v{}",
+                tantivy_version.0,
+                tantivy_version.1,
+                tantivy_version.2,
+                versioned_footer.version()
+            ),
+            versioned_footer,
+        }
+    }
+
+    pub fn to_bytes(&self) -> Vec<u8> {
+        let mut res = self.versioned_footer.to_bytes();
+        res.extend_from_slice(self.meta.as_bytes());
+        let len = res.len();
+        res.resize(len + COMMON_FOOTER_SIZE, 0);
+        let mut common_footer = &mut res[len..];
+        LittleEndian::write_u32(&mut common_footer, self.meta.len() as u32);
+        LittleEndian::write_u32(&mut common_footer[4..], self.tantivy_version.0);
+        LittleEndian::write_u32(&mut common_footer[8..], self.tantivy_version.1);
+        LittleEndian::write_u32(&mut common_footer[12..], self.tantivy_version.2);
+        LittleEndian::write_u32(&mut common_footer[16..], (len + COMMON_FOOTER_SIZE) as u32);
+        res
+    }
+
+    pub fn from_bytes(data: &[u8]) -> Result<Self, io::Error> {
+        let len = data.len();
+        if len < COMMON_FOOTER_SIZE + 4 {
+            // 4 bytes for index version, stored in versioned footer
+            return Err(io::Error::new(
+                    io::ErrorKind::UnexpectedEof,
+                    format!("File corrupted. The footer len must be over 24, while the entire file len is {}", len)
+                    )
+            );
+        }
+
+        let size = LittleEndian::read_u32(&data[len - 4..]) as usize;
+        if len < size as usize {
+            return Err(io::Error::new(
+                io::ErrorKind::UnexpectedEof,
+                format!(
+                    "File corrupted. The footer len is {}, while the entire file len is {}",
+                    size, len
+                ),
+            ));
+        }
+        let footer = &data[len - size as usize..];
+        let meta_len = LittleEndian::read_u32(&footer[size - 20..]) as usize;
+        let tantivy_major = LittleEndian::read_u32(&footer[size - 16..]);
+        let tantivy_minor = LittleEndian::read_u32(&footer[size - 12..]);
+        let tantivy_patch = LittleEndian::read_u32(&footer[size - 8..]);
+        Ok(Footer {
+            tantivy_version: (tantivy_major, tantivy_minor, tantivy_patch),
+            meta: String::from_utf8_lossy(&footer[size - meta_len - 20..size - 20]).into_owned(),
+            versioned_footer: VersionedFooter::from_bytes(&footer[..size - meta_len - 20])?,
+        })
+    }
+
+    pub fn extract_footer(source: ReadOnlySource) -> Result<(Footer, ReadOnlySource), io::Error> {
+        let footer = Footer::from_bytes(source.as_slice())?;
+        let reader = source.slice_to(source.as_slice().len() - footer.size());
+        Ok((footer, reader))
+    }
+
+    pub fn size(&self) -> usize {
+        self.versioned_footer.size() as usize + self.meta.len() + 20
+    }
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub enum VersionedFooter {
+    UnknownVersion { version: u32, size: u32 },
+    V0(u32), // crc
+}
+
+impl VersionedFooter {
+    pub fn to_bytes(&self) -> Vec<u8> {
+        match self {
+            Self::V0(crc) => {
+                let mut res = vec![0; 8];
+                LittleEndian::write_u32(&mut res, 0);
+                LittleEndian::write_u32(&mut res[4..], *crc);
+                res
+            }
+            Self::UnknownVersion { .. } => {
+                panic!("Unsupported index should never get serialized");
+            }
+        }
+    }
+
+    pub fn from_bytes(footer: &[u8]) -> Result<Self, io::Error> {
+        assert!(footer.len() >= 4);
+        let version = LittleEndian::read_u32(footer);
+        match version {
+            0 => {
+                if footer.len() == 8 {
+                    Ok(Self::V0(LittleEndian::read_u32(&footer[4..])))
+                } else {
+                    Err(io::Error::new(
+                        io::ErrorKind::UnexpectedEof,
+                        format!(
+                            "File corrupted. The versioned footer len is {}, while it should be 8",
+                            footer.len()
+                        ),
+                    ))
+                }
+            }
+            version => Ok(Self::UnknownVersion {
+                version,
+                size: footer.len() as u32,
+            }),
+        }
+    }
+
+    pub fn size(&self) -> u32 {
+        match self {
+            Self::V0(_) => 8,
+            Self::UnknownVersion { size, .. } => *size,
+        }
+    }
+
+    pub fn version(&self) -> u32 {
+        match self {
+            Self::V0(_) => 0,
+            Self::UnknownVersion { version, .. } => *version,
+        }
+    }
+
+    pub fn crc(&self) -> Option<u32> {
+        match self {
+            Self::V0(crc) => Some(*crc),
+            Self::UnknownVersion { .. } => None,
+        }
+    }
+}
+
+pub(crate) struct FooterProxy<W: TerminatingWrite> {
+    /// always Some except after terminate call
+    hasher: Option<Hasher>,
+    /// always Some except after terminate call
+    writer: Option<W>,
+}
+
+impl<W: TerminatingWrite> FooterProxy<W> {
+    pub fn new(writer: W) -> Self {
+        FooterProxy {
+            hasher: Some(Hasher::new()),
+            writer: Some(writer),
+        }
+    }
+}
+
+impl<W: TerminatingWrite> Write for FooterProxy<W> {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        let count = self.writer.as_mut().unwrap().write(buf)?;
+        self.hasher.as_mut().unwrap().update(&buf[..count]);
+        Ok(count)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.writer.as_mut().unwrap().flush()
+    }
+}
+
+impl<W: TerminatingWrite> TerminatingWrite for FooterProxy<W> {
+    fn terminate_ref(&mut self, _: AntiCallToken) -> io::Result<()> {
+        let crc = self.hasher.take().unwrap().finalize();
+
+        let footer = Footer::new(VersionedFooter::V0(crc)).to_bytes();
+        let mut writer = self.writer.take().unwrap();
+        writer.write_all(&footer)?;
+        writer.terminate()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::directory::footer::{Footer, VersionedFooter};
+
+    #[test]
+    fn test_serialize_deserialize_footer() {
+        let crc = 123456;
+        let footer = Footer::new(VersionedFooter::V0(crc));
+        let footer_bytes = footer.to_bytes();
+
+        assert_eq!(Footer::from_bytes(&footer_bytes).unwrap(), footer);
+    }
+}