CASSANDRA-21519: Return CorruptSSTableException if chunk metadata and file size are out of sync

CHANGES.txt

@@ -69,6 +69,7 @@
  * Fix a removed TTLed row re-appearance in a materialized view after a cursor compaction (CASSANDRA-21152)
  * Rework ZSTD dictionary compression logic to create a trainer per training (CASSANDRA-21209)
 Merged from 5.0:
+ * Allow setCompressedReadAheadBufferSizeInKb(0) to disable read-ahead buffer (CASSANDRA-21522)
  * SAI Component Checksum Validation Should be Segment-Aware (CASSANDRA-21516)
  * Support Python 3.12 and 3.13 in cqlsh (CASSANDRA-20997)
  * Make synchronization on VectorMemoryIndex inserts more granular (CASSANDRA-21160)

src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java

@@ -24,6 +24,7 @@
 import java.util.function.Supplier;
 
 import org.apache.cassandra.io.compress.BufferType;
+import org.apache.cassandra.io.compress.CorruptBlockException;
 import org.apache.cassandra.io.sstable.CorruptSSTableException;
 import org.apache.cassandra.utils.Closeable;
 import org.apache.cassandra.utils.memory.MemoryUtil;
@@ -100,12 +101,14 @@ private Block block()
         return blockMap.get().computeIfAbsent(channel.filePath(), k -> new Block());
     }
 
-    public void fill(long position)
+    public void fill(long position) throws CorruptBlockException
     {
         Block block = getBlock();
         ByteBuffer blockBuffer = block.buffer;
-        long realPosition = Math.min(channelSize, position);
-        int blockNo = (int) (realPosition / bufferSize);
+        if (position >= channelSize)
+            throw new CorruptBlockException(channel.filePath(), position, bufferSize);
+
+        int blockNo = (int) (position / bufferSize);
         long blockPosition = blockNo * (long) bufferSize;
 
         long remaining = channelSize - blockPosition;
@@ -119,7 +122,7 @@ public void fill(long position)
 
         blockBuffer.flip();
         blockBuffer.limit(sizeToRead);
-        blockBuffer.position((int) (realPosition - blockPosition));
+        blockBuffer.position((int) (position - blockPosition));
     }
 
     protected void loadBlock(ByteBuffer blockBuffer, long blockPosition, int sizeToRead)

test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java

@@ -19,7 +19,9 @@
 package org.apache.cassandra.io.util;
 
 import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
 import java.nio.file.Files;
+import java.nio.file.StandardOpenOption;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.assertj.core.api.Assertions;
@@ -33,6 +35,7 @@
 import org.apache.cassandra.io.compress.CompressedSequentialWriter;
 import org.apache.cassandra.io.compress.CompressionMetadata;
 import org.apache.cassandra.io.filesystem.ListenableFileSystem;
+import org.apache.cassandra.io.sstable.CorruptSSTableException;
 import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
 import org.apache.cassandra.schema.CompressionParams;
 import org.apache.cassandra.utils.memory.MemoryUtil;
@@ -112,4 +115,55 @@ protected void doReads(File f, CompressionMetadata metadata, long length, boolea
             MemoryUtil.clean(buffer);
         }
     }
+
+    @Test(timeout = 10_000)
+    public void scanReaderShouldNotHangOnTruncatedFile() throws Exception
+    {
+        SequentialWriterOption writerOption = SequentialWriterOption.newBuilder().finishOnClose(false).bufferSize(1 << 10).build();
+        CompressionParams params = CompressionParams.snappy(4096, 1.1);
+
+        FileSystems.newGlobalInMemoryFileSystem();
+        File f = new File("/truncated_hang_repro.db");
+        File offsets = new File("/truncated_hang_repro.offset");
+        File digest = new File("/truncated_hang_repro.digest");
+
+        long longsToWrite = 600; // 4800 uncompressed bytes -> 2 compressed chunks (second one partial)
+        CompressionMetadata metadata;
+        try (CompressedSequentialWriter writer = new CompressedSequentialWriter(f, offsets, digest, writerOption, params, new MetadataCollector(new ClusteringComparator())))
+        {
+            for (long i = 0; i < longsToWrite; i++)
+                writer.writeLong(i);
+
+            writer.sync();
+            metadata = writer.open(0);
+        }
+
+        DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(256);
+
+        // Truncate file so that chunk metadata expects a chunk to extend further than the actual file size
+        long originalSize = Files.size(f.toPath());
+        long truncatedSize = originalSize - (params.chunkLength() / 2);
+        try (FileChannel fc = FileChannel.open(f.toPath(), StandardOpenOption.WRITE))
+        {
+            fc.truncate(truncatedSize);
+        }
+        long uncompressedTotal = longsToWrite * Long.BYTES;
+        long lastChunkUncompressedStart = ((uncompressedTotal - 1) / metadata.chunkLength()) * metadata.chunkLength();
+
+        ByteBuffer buffer = ByteBuffer.allocateDirect(metadata.chunkLength());
+        try (ChannelProxy channel = new ChannelProxy(f);
+             CompressedChunkReader reader = new CompressedChunkReader.Standard(channel, metadata, () -> 1.1);
+             metadata)
+        {
+            reader.forScan();
+
+            Assertions.assertThatThrownBy(() -> reader.readChunk(lastChunkUncompressedStart, buffer))
+                      .as("readChunk() reading past truncated EOF via the scan path")
+                      .isInstanceOf(CorruptSSTableException.class);
+        }
+        finally
+        {
+            MemoryUtil.clean(buffer);
+        }
+    }
 }

test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java

@@ -37,6 +37,7 @@
 
 import org.apache.cassandra.config.DataStorageSpec;
 import org.apache.cassandra.io.compress.BufferType;
+import org.apache.cassandra.io.compress.CorruptBlockException;
 import org.apache.cassandra.io.sstable.CorruptSSTableException;
 import org.apache.cassandra.utils.Pair;
 
@@ -125,7 +126,7 @@ protected static void testRead(Pair<Long, Integer> read, ChannelProxy bufferedCh
                     copied += tlrab.read(buf2, tlrab.remaining());
             }
         }
-        catch (CorruptSSTableException e)
+        catch (CorruptSSTableException | CorruptBlockException e)
         {
             throw new RuntimeException(e);
         }