Check footer checksum

src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java

@@ -29,7 +29,6 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
-import java.util.zip.CRC32;
 
 import javax.annotation.Nullable;
 

src/java/org/apache/cassandra/index/sai/utils/IndexFileUtils.java

@@ -45,6 +45,7 @@
 import org.apache.cassandra.io.util.RandomAccessReader;
 import org.apache.cassandra.io.util.SequentialWriter;
 import org.apache.cassandra.io.util.SequentialWriterOption;
+import org.apache.lucene.codecs.CodecUtil;
 
 public class IndexFileUtils
 {
@@ -142,10 +143,13 @@ class IncrementalChecksumSequentialWriter extends SequentialWriter implements Ch
                 var fileLength = file.length();
                 skipBytes(fileLength);
 
-                // It is possible we didn't get a good checksum, because the cache has limited size.
-                // We could have gotten a fresh zero checksum. In this case we need to recalculate:
-                if (checksum.fileLength != fileLength)
-                    recalculateChecksum();
+                // It is possible we didn't get a good checksum.
+                // We could have gotten a zero checksum because the cache has a limited size,
+                // or the file could have been changed in the meantime by another process, in which case the
+                // footer checksum would not match.
+                // However, we don't recalculate the checksum always, because it is very costly:
+                if (checksum.fileLength != fileLength || checksum.footerChecksum != calculateFooterChecksum())
+                    recalculateFileChecksum();
             }
             else
             {
@@ -165,7 +169,7 @@ class IncrementalChecksumSequentialWriter extends SequentialWriter implements Ch
          *
          * @throws IOException if file read failed.
          */
-        public void recalculateChecksum() throws IOException
+        public void recalculateFileChecksum() throws IOException
         {
             checksum.reset();
             if (!file.exists())
@@ -190,6 +194,32 @@ public void recalculateChecksum() throws IOException
             assert checksum.fileLength == position();
         }
 
+        /**
+         * Returns the checksum of the footer of the index file.
+         * Those bytes contain the checksum for the whole file, so
+         * this checksum can be used to verify the integrity of the whole file.
+         * Note that this is not the same as checksum written in the file footer;
+         * this is done this way so we don't have to decode the footer here.
+         */
+        public long calculateFooterChecksum() throws IOException
+        {
+            CRC32 footerChecksum = new CRC32();
+            try (FileChannel ch = StorageProvider.instance.writeTimeReadFileChannelFor(file))
+            {
+                ch.position(Math.max(0, file.length() - CodecUtil.footerLength()));
+                final ByteBuffer buf = ByteBuffer.allocate(CodecUtil.footerLength());
+                int b = ch.read(buf);
+                while (b > 0)
+                {
+                    buf.flip();
+                    footerChecksum.update(buf);
+                    buf.clear();
+                    b = ch.read(buf);
+                }
+            }
+            return footerChecksum.getValue();
+        }
+
         @Override
         public void write(ByteBuffer src) throws IOException
         {
@@ -278,11 +308,24 @@ public void close()
             }
             finally
             {
-                // Copy the checksum value to a field in order to make the checksum value available past close().
-                // Release the FileChecksum object so it can be used by another writer.
-                finalChecksum = checksum.getValue();
-                checksum = null;
-                checksumGuard.release();
+                try
+                {
+                    // Copy the checksum value to a field in order to make the checksum value available past close().
+                    // Release the FileChecksum object so it can be used by another writer.
+                    finalChecksum = checksum.getValue();
+                    checksum.footerChecksum = calculateFooterChecksum();
+                }
+                catch (IOException e)
+                {
+                    // mark the checksum as unusable
+                    // even though it stays in the cache, it won't ever match on the file length
+                    checksum.fileLength = -1;
+                }
+                finally
+                {
+                    checksumGuard.release();
+                    checksum = null;
+                }
             }
         }
     }
@@ -327,41 +370,36 @@ public void release()
     static class FileChecksum
     {
         long fileLength = 0;
-        CRC32 checksum = new CRC32();
+        long footerChecksum = 0;
+        CRC32 fileChecksum = new CRC32();
 
         public void reset()
         {
             fileLength = 0;
-            checksum.reset();
+            fileChecksum.reset();
         }
 
         public void update(int b)
         {
             fileLength += 1;
-            checksum.update(b);
-        }
-
-        public void update(byte[] b)
-        {
-            fileLength += b.length;
-            checksum.update(b);
+            fileChecksum.update(b);
         }
 
         public void update(byte[] b, int off, int len)
         {
             fileLength += len;
-            checksum.update(b, off, len);
+            fileChecksum.update(b, off, len);
         }
 
         public void update(ByteBuffer b)
         {
             fileLength += b.remaining();
-            checksum.update(b);
+            fileChecksum.update(b);
         }
 
         public long getValue()
         {
-            return checksum.getValue();
+            return fileChecksum.getValue();
         }
     }