From fa93639eb84ad4384996d9fb5eafe6e2bf0ea66c Mon Sep 17 00:00:00 2001 From: nivy Date: Mon, 13 Jul 2026 23:03:01 -0700 Subject: [PATCH] Fix fill() to return CorruptSSTableException if chunk metadata and file size are out of sync --- .../io/util/ThreadLocalReadAheadBuffer.java | 8 ++- .../StandardCompressedChunkReaderTest.java | 57 +++++++++++++++++++ 2 files changed, 62 insertions(+), 3 deletions(-) diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java index c8ef6cf1e3bf..57490f404a65 100644 --- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java +++ b/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) diff --git a/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java b/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java index afad1cd11758..f38b2bf66ce3 100644 --- a/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java +++ b/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); + } + } } \ No newline at end of file