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

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

@@ -104,8 +104,10 @@ public void fill(long position)
     {
         Block block = getBlock();
         ByteBuffer blockBuffer = block.buffer;
-        long realPosition = Math.min(channelSize, position);
-        int blockNo = (int) (realPosition / bufferSize);
+        if (position >= channelSize)
+            throw new CorruptSSTableException(null, channel.filePath());
+
+        int blockNo = (int) (position / bufferSize);
         long blockPosition = blockNo * (long) bufferSize;
 
         long remaining = channelSize - blockPosition;
@@ -119,7 +121,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,58 @@ 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);
+        }
+
+        // Sanity: read-ahead must actually engage for this test to exercise the scan path
+        Assert.assertTrue("read-ahead must be larger than chunk length for scanReader to be non-null",
+                          DatabaseDescriptor.getCompressedReadAheadBufferSize() > metadata.chunkLength());
+
+        // 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);
+        }
+    }
 }