From 4bb80039eb9d69384adf24fbc5186e5d9bd9b082 Mon Sep 17 00:00:00 2001 From: Bikram Sharma Date: Thu, 24 Sep 2026 12:54:11 -0700 Subject: [PATCH] fix: don't signal onComplete when a chunk is fully consumed by the range skip AdjustedRangeSubscriber called onComplete (and fell through to an NPE) when numBytesToSkip exceeded a chunk, so a small first chunk ended ranged GETs with an empty stream. Consume the chunk toward the skip and wait for the next one instead. Fixes #517. --- .../internal/AdjustedRangeSubscriber.java | 14 +- .../internal/AdjustedRangeSubscriberTest.java | 124 ++++++++++++++++++ 2 files changed, 134 insertions(+), 4 deletions(-) create mode 100644 src/test/java/software/amazon/encryption/s3/legacy/internal/AdjustedRangeSubscriberTest.java diff --git a/src/main/java/software/amazon/encryption/s3/legacy/internal/AdjustedRangeSubscriber.java b/src/main/java/software/amazon/encryption/s3/legacy/internal/AdjustedRangeSubscriber.java index 73de3b214..011497d64 100644 --- a/src/main/java/software/amazon/encryption/s3/legacy/internal/AdjustedRangeSubscriber.java +++ b/src/main/java/software/amazon/encryption/s3/legacy/internal/AdjustedRangeSubscriber.java @@ -61,11 +61,17 @@ public void onNext(ByteBuffer byteBuffer) { if (numBytesToSkip != 0) { byte[] buf = byteBuffer.array(); - if (numBytesToSkip > buf.length) { - // If we need to skip past the available data, - // we are returning nothing, so signal completion + if (numBytesToSkip >= buf.length) { + // This chunk is entirely consumed by the leading-byte skip, so there is no + // in-range data to deliver yet. We must still signal the wrapped subscriber to + // keep the reactive-streams demand flowing: downstream drives the stream one + // element at a time (request(1)), and it only requests the next element after it + // receives an onNext. Returning without signaling would leave it waiting forever. + // Forward an empty buffer, mirroring CipherSubscriber's "avoid blocking" idiom, + // and wait for the next chunk instead of completing the stream. numBytesToSkip -= buf.length; - wrappedSubscriber.onComplete(); + wrappedSubscriber.onNext(ByteBuffer.wrap(new byte[0])); + return; } else { outputBuffer = Arrays.copyOfRange(buf, numBytesToSkip, buf.length); numBytesToSkip = 0; diff --git a/src/test/java/software/amazon/encryption/s3/legacy/internal/AdjustedRangeSubscriberTest.java b/src/test/java/software/amazon/encryption/s3/legacy/internal/AdjustedRangeSubscriberTest.java new file mode 100644 index 000000000..1173a472a --- /dev/null +++ b/src/test/java/software/amazon/encryption/s3/legacy/internal/AdjustedRangeSubscriberTest.java @@ -0,0 +1,124 @@ +// Copyright Amazon.com Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.encryption.s3.legacy.internal; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; + +import java.io.ByteArrayOutputStream; +import java.nio.ByteBuffer; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; + +public class AdjustedRangeSubscriberTest { + + /** + * Records everything delivered downstream so tests can assert on bytes and + * terminal signals. + */ + private static class RecordingSubscriber implements Subscriber { + final ByteArrayOutputStream data = new ByteArrayOutputStream(); + final AtomicInteger completeCount = new AtomicInteger(); + final AtomicInteger onNextCount = new AtomicInteger(); + Throwable error; + + @Override + public void onSubscribe(Subscription s) { + } + + @Override + public void onNext(ByteBuffer byteBuffer) { + onNextCount.incrementAndGet(); + byte[] b = new byte[byteBuffer.remaining()]; + byteBuffer.get(b); + data.write(b, 0, b.length); + } + + @Override + public void onError(Throwable t) { + error = t; + } + + @Override + public void onComplete() { + completeCount.incrementAndGet(); + } + } + + private static ByteBuffer bytes(int start, int length) { + byte[] b = new byte[length]; + for (int i = 0; i < length; i++) { + b[i] = (byte) (start + i); + } + return ByteBuffer.wrap(b); + } + + @Test + public void testFirstChunkSmallerThanSkipDoesNotCompleteOrThrow() throws Exception { + // rangeBeginning=20 => numBytesToSkip=20, virtualAvailable=100 + RecordingSubscriber downstream = new RecordingSubscriber(); + AdjustedRangeSubscriber subscriber = new AdjustedRangeSubscriber(downstream, 20L, 119L); + + // First chunk (10 bytes) is smaller than the 20-byte skip. + subscriber.onNext(bytes(0, 10)); + assertEquals(0, downstream.completeCount.get()); + assertNull(downstream.error); + assertEquals(0, downstream.data.size()); + + // Second chunk (10 bytes) exactly finishes the skip; still no data delivered. + subscriber.onNext(bytes(10, 10)); + assertEquals(0, downstream.completeCount.get()); + assertEquals(0, downstream.data.size()); + + // Third chunk carries the actual payload, which must now be delivered. + subscriber.onNext(bytes(100, 100)); + assertArrayEquals(bytes(100, 100).array(), downstream.data.toByteArray()); + } + + @Test + public void testEmptyFirstChunkIsSkippedNotTreatedAsCompletion() throws Exception { + RecordingSubscriber downstream = new RecordingSubscriber(); + AdjustedRangeSubscriber subscriber = new AdjustedRangeSubscriber(downstream, 20L, 119L); + + // CipherSubscriber can emit an empty buffer; it must not complete the stream. + subscriber.onNext(ByteBuffer.allocate(0)); + assertEquals(0, downstream.completeCount.get()); + assertNull(downstream.error); + + subscriber.onNext(bytes(0, 20)); // finish the skip + subscriber.onNext(bytes(50, 100)); + assertArrayEquals(bytes(50, 100).array(), downstream.data.toByteArray()); + } + + @Test + public void testSkipOnlyChunkSignalsEmptyOnNextToKeepDemandFlowing() throws Exception { + // Under one-at-a-time (request(1)) demand, a chunk fully consumed by the skip must still + // signal the wrapped subscriber, or downstream waits forever for the next element. The + // subscriber forwards an empty buffer (no in-range data, but the demand chain advances). + RecordingSubscriber downstream = new RecordingSubscriber(); + AdjustedRangeSubscriber subscriber = new AdjustedRangeSubscriber(downstream, 20L, 119L); + + subscriber.onNext(bytes(0, 10)); // 10 bytes < 20-byte skip: entirely skipped + + assertEquals(1, downstream.onNextCount.get(), "skip-only chunk must forward exactly one onNext"); + assertEquals(0, downstream.data.size(), "the forwarded onNext must carry no in-range bytes"); + assertEquals(0, downstream.completeCount.get(), "skip-only chunk must not complete the stream"); + assertNull(downstream.error); + } + + @Test + public void testChunkLargerThanSkipDeliversRemainder() throws Exception { + RecordingSubscriber downstream = new RecordingSubscriber(); + AdjustedRangeSubscriber subscriber = new AdjustedRangeSubscriber(downstream, 20L, 119L); + + // A single 120-byte chunk: 20 skipped, 100 delivered. + subscriber.onNext(bytes(0, 120)); + assertArrayEquals(bytes(20, 100).array(), downstream.data.toByteArray()); + assertFalse(downstream.completeCount.get() == 0); + } +}