From 1a72be8cd0dbf08568d07a8ffa9044ac3b82e11f Mon Sep 17 00:00:00 2001 From: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> Date: Thu, 27 Aug 2026 01:03:26 +0800 Subject: [PATCH] binder: Fail fast on transaction sequence gaps Binder one-way transactions are delivered in order, so sparse buffering cannot recover a silently dropped transaction. Detect unexpected sequence numbers and close the stream with UNAVAILABLE instead of waiting indefinitely. Simplify the now-sequential message queue and cover dropped messages and fragments in both directions. Fixes #12747 --- .../java/io/grpc/binder/internal/Inbound.java | 59 +++---- .../io/grpc/binder/internal/Outbound.java | 4 +- .../RobolectricBinderTransportTest.java | 167 ++++++++---------- .../binder/internal/OneWayBinderProxies.java | 8 +- 4 files changed, 109 insertions(+), 129 deletions(-) diff --git a/binder/src/main/java/io/grpc/binder/internal/Inbound.java b/binder/src/main/java/io/grpc/binder/internal/Inbound.java index 83fc8273d6f..983e55dcbf8 100644 --- a/binder/src/main/java/io/grpc/binder/internal/Inbound.java +++ b/binder/src/main/java/io/grpc/binder/internal/Inbound.java @@ -39,8 +39,6 @@ /** * Handles incoming binder transactions for a single stream, turning those transactions into calls * to the stream listener. - * - *

Out-of-order messages are reassembled into their correct order. */ abstract class Inbound implements StreamListener.MessageProducer { @@ -74,6 +72,9 @@ abstract class Inbound @GuardedBy("this") private int firstQueuedTransactionIndex; + @GuardedBy("this") + private int nextExpectedTransactionIndex; + @GuardedBy("this") private int nextCompleteMessageEnd; @@ -102,9 +103,7 @@ abstract class Inbound * delivery what we've sent. */ enum State { - // We aren't yet connected to a BinderStream instance and listener. Due to potentially - // out-of-order messages, a server-side instance can remain in this state for multiple - // transactions. + // We aren't yet connected to a BinderStream instance and listener. UNINITIALIZED, // We're attached to a BinderStream instance and we have a listener we can report to. @@ -344,6 +343,16 @@ final synchronized void handleTransaction(Parcel parcel) { return; } int index = parcel.readInt(); + if (index != nextExpectedTransactionIndex) { + throw Status.UNAVAILABLE + .withDescription( + "Binder transaction sequence gap: expected " + + nextExpectedTransactionIndex + + ", received " + + index) + .asException(); + } + nextExpectedTransactionIndex += 1; boolean hasPrefix = TransactionUtils.hasFlag(flags, TransactionUtils.FLAG_PREFIX); boolean hasMessageData = TransactionUtils.hasFlag(flags, TransactionUtils.FLAG_MESSAGE_DATA); boolean hasSuffix = TransactionUtils.hasFlag(flags, TransactionUtils.FLAG_SUFFIX); @@ -352,7 +361,7 @@ final synchronized void handleTransaction(Parcel parcel) { onDeliveryState(State.PREFIX_DELIVERED); } if (hasMessageData) { - handleMessageData(flags, index, parcel); + handleMessageData(flags, parcel); } if (hasSuffix) { handleSuffix(flags, parcel); @@ -383,7 +392,7 @@ final synchronized void handleTransaction(Parcel parcel) { abstract void handleSuffix(int flags, Parcel parcel) throws StatusException; @GuardedBy("this") - private void handleMessageData(int flags, int index, Parcel parcel) throws StatusException { + private void handleMessageData(int flags, Parcel parcel) throws StatusException { InputStream stream = null; byte[] block = null; boolean lastBlockOfMessage = true; @@ -417,7 +426,7 @@ private void handleMessageData(int flags, int index, Parcel parcel) throws Statu } } if (queuedTransactionData == null) { - if (numReceivedMessages == 0 && lastBlockOfMessage && index == firstQueuedTransactionIndex) { + if (numReceivedMessages == 0 && lastBlockOfMessage) { // Shortcut for when we receive a single message in one transaction. checkState(firstMessage == null); firstMessage = (stream != null) ? stream : new BlockInputStream(block); @@ -426,24 +435,13 @@ private void handleMessageData(int flags, int index, Parcel parcel) throws Statu } queuedTransactionData = new ArrayList<>(16); } - enqueueTransactionData(index, new TransactionData(stream, block, numBytes, lastBlockOfMessage)); + enqueueTransactionData(new TransactionData(stream, block, numBytes, lastBlockOfMessage)); } @GuardedBy("this") - private void enqueueTransactionData(int index, TransactionData data) { - int offset = index - firstQueuedTransactionIndex; - if (offset < queuedTransactionData.size()) { - queuedTransactionData.set(offset, data); - lookForCompleteMessage(); - } else if (offset > queuedTransactionData.size()) { - do { - queuedTransactionData.add(null); - } while (offset > queuedTransactionData.size()); - queuedTransactionData.add(data); - } else { - queuedTransactionData.add(data); - lookForCompleteMessage(); - } + private void enqueueTransactionData(TransactionData data) { + queuedTransactionData.add(data); + lookForCompleteMessage(); } @GuardedBy("this") @@ -452,17 +450,12 @@ private void lookForCompleteMessage() { if (nextCompleteMessageEnd == 0) { for (int i = 0; i < queuedTransactionData.size(); i++) { TransactionData data = queuedTransactionData.get(i); - if (data == null) { - // Missing block. + numBytes += data.numBytes; + if (data.lastBlockOfMessage) { + // Found a complete message. + nextCompleteMessageEnd = i + 1; + reportInboundMessage(numBytes); return; - } else { - numBytes += data.numBytes; - if (data.lastBlockOfMessage) { - // Found a complete message. - nextCompleteMessageEnd = i + 1; - reportInboundMessage(numBytes); - return; - } } } } diff --git a/binder/src/main/java/io/grpc/binder/internal/Outbound.java b/binder/src/main/java/io/grpc/binder/internal/Outbound.java index 7db5bf0fbe4..ff5c677cd6a 100644 --- a/binder/src/main/java/io/grpc/binder/internal/Outbound.java +++ b/binder/src/main/java/io/grpc/binder/internal/Outbound.java @@ -90,8 +90,8 @@ enum State { private boolean suffixReady; /** - * The index of the next transaction we'll send, allowing the receiver to re-assemble out-of-order - * messages. + * The index of the next transaction we'll send, allowing the receiver to detect dropped + * transactions. */ @GuardedBy("this") private int transactionIndex; diff --git a/binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java b/binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java index 63c47bf4f19..a5a10b63fd0 100644 --- a/binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java +++ b/binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java @@ -18,10 +18,8 @@ import static android.os.IBinder.FLAG_ONEWAY; import static android.os.Process.myUid; -import static com.google.common.truth.Truth.assertAbout; import static com.google.common.truth.Truth.assertThat; import static com.google.common.util.concurrent.MoreExecutors.directExecutor; -import static io.grpc.StatusSubject.status; import static io.grpc.binder.internal.BinderTransport.REMOTE_UID; import static io.grpc.binder.internal.BinderTransport.SETUP_TRANSPORT; import static io.grpc.binder.internal.BinderTransport.SHUTDOWN_TRANSPORT; @@ -73,9 +71,7 @@ import io.grpc.internal.MockServerTransportListener; import io.grpc.internal.ObjectPool; import io.grpc.internal.SharedResourcePool; -import java.io.InputStream; import java.util.List; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; import org.junit.Before; @@ -453,7 +449,7 @@ public void flowControlPushBack() {} public void serverAlreadyListening() {} @Test - public void singleTxnMsgsDeliveredToServerOutOfOrder() throws Exception { + public void serverFailsFastOnGapBetweenMessages() throws Exception { server.start(serverListener); client = newClientTransportBuilder() @@ -479,41 +475,30 @@ public void singleTxnMsgsDeliveredToServerOutOfOrder() throws Exception { stream.halfClose(); // Expect one transaction for headers, one for each message, and one for half-close. - QueueingOneWayBinderProxy.Transaction txHeaders = takeNextTransaction(queueingServerProxy); - QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(queueingServerProxy); - QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(queueingServerProxy); - QueueingOneWayBinderProxy.Transaction txHalfClose = takeNextTransaction(queueingServerProxy); - - // Deliver messages out of order! - queueingServerProxy.deliver(txHeaders); - queueingServerProxy.deliver(tx2); - queueingServerProxy.deliver(tx1); - queueingServerProxy.deliver(txHalfClose); + try (QueueingOneWayBinderProxy.Transaction txHeaders = + takeNextTransaction(queueingServerProxy); + QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(queueingServerProxy); + QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(queueingServerProxy); + QueueingOneWayBinderProxy.Transaction txHalfClose = + takeNextTransaction(queueingServerProxy)) { + queueingServerProxy.deliver(txHeaders); + // Simulate tx1 being silently dropped. Receiving tx2 exposes the gap. + queueingServerProxy.deliver(tx2); + } MockServerTransportListener serverTransportListener = serverListener.takeListenerOrFail(TIMEOUT_MS, MILLISECONDS); MockServerTransportListener.StreamCreation serverStreamCreation = serverTransportListener.takeStreamOrFail(TIMEOUT_MS, MILLISECONDS); - serverStreamCreation.stream.request(2); - - // Expect the server to deliver the messages in the order they were originally sent. - InputStream msg1 = takeNextMessage(serverStreamCreation.listener.messageQueue); - assertThat(methodDescriptor.parseResponse(msg1)).isEqualTo("one"); - - InputStream msg2 = takeNextMessage(serverStreamCreation.listener.messageQueue); - assertThat(methodDescriptor.parseResponse(msg2)).isEqualTo("two"); - assertThat(serverStreamCreation.listener.awaitHalfClosed(TIMEOUT_MS, MILLISECONDS)).isTrue(); - serverStreamCreation.stream.close(Status.OK, new Metadata()); - - assertAbout(status()).that(clientStreamListener.awaitClose(TIMEOUT_MS, MILLISECONDS)).isOk(); - assertAbout(status()) - .that(serverStreamCreation.listener.awaitClose(TIMEOUT_MS, MILLISECONDS)) - .isOk(); + assertTransactionSequenceGap( + clientStreamListener.awaitClose(TIMEOUT_MS, MILLISECONDS), 1, 2); + assertTransactionSequenceGap( + serverStreamCreation.listener.awaitClose(TIMEOUT_MS, MILLISECONDS), 1, 2); } @Test - public void msgFragmentsDeliveredToServerOutOfOrder() throws Exception { + public void serverFailsFastOnGapBetweenMessageFragments() throws Exception { server.start(serverListener); client = newClientTransportBuilder() @@ -540,37 +525,30 @@ public void msgFragmentsDeliveredToServerOutOfOrder() throws Exception { stream.halfClose(); // Expect the client to split largeMessage into two transactions, plus headers and half-close. - QueueingOneWayBinderProxy.Transaction txHeaders = takeNextTransaction(queueingServerProxy); - QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(queueingServerProxy); - QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(queueingServerProxy); - QueueingOneWayBinderProxy.Transaction txHalfClose = takeNextTransaction(queueingServerProxy); - - // Deliver fragments out of order! - queueingServerProxy.deliver(txHeaders); - queueingServerProxy.deliver(tx2); - queueingServerProxy.deliver(tx1); - queueingServerProxy.deliver(txHalfClose); - - // Verify that the server reassembles the transactions correctly. + try (QueueingOneWayBinderProxy.Transaction txHeaders = + takeNextTransaction(queueingServerProxy); + QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(queueingServerProxy); + QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(queueingServerProxy); + QueueingOneWayBinderProxy.Transaction txHalfClose = + takeNextTransaction(queueingServerProxy)) { + queueingServerProxy.deliver(txHeaders); + // Simulate the first fragment being silently dropped. + queueingServerProxy.deliver(tx2); + } + MockServerTransportListener serverTransportListener = serverListener.takeListenerOrFail(TIMEOUT_MS, MILLISECONDS); MockServerTransportListener.StreamCreation serverStreamCreation = serverTransportListener.takeStreamOrFail(TIMEOUT_MS, MILLISECONDS); - serverStreamCreation.stream.request(1); - InputStream msg = takeNextMessage(serverStreamCreation.listener.messageQueue); - assertThat(methodDescriptor.parseResponse(msg)).isEqualTo(largeMessage); - - assertThat(serverStreamCreation.listener.awaitHalfClosed(TIMEOUT_MS, MILLISECONDS)).isTrue(); - serverStreamCreation.stream.close(Status.OK, new Metadata()); - assertAbout(status()).that(clientStreamListener.awaitClose(TIMEOUT_MS, MILLISECONDS)).isOk(); - assertAbout(status()) - .that(serverStreamCreation.listener.awaitClose(TIMEOUT_MS, MILLISECONDS)) - .isOk(); + assertTransactionSequenceGap( + clientStreamListener.awaitClose(TIMEOUT_MS, MILLISECONDS), 1, 2); + assertTransactionSequenceGap( + serverStreamCreation.listener.awaitClose(TIMEOUT_MS, MILLISECONDS), 1, 2); } @Test - public void singleTxnMsgsDeliveredToClientOutOfOrder() throws Exception { + public void clientFailsFastOnGapBetweenMessages() throws Exception { server = newServerBuilder().setClientBinderDecorator(blockingDecorator).build(); registerServerWithRobolectric((BinderServer) server); server.start(serverListener); @@ -598,34 +576,29 @@ public void singleTxnMsgsDeliveredToClientOutOfOrder() throws Exception { MockServerTransportListener.StreamCreation serverStreamCreation = serverTransportListener.takeStreamOrFail(TIMEOUT_MS, MILLISECONDS); + serverStreamCreation.stream.writeHeaders(new Metadata(), true); serverStreamCreation.stream.writeMessage(methodDescriptor.streamResponse("one")); serverStreamCreation.stream.writeMessage(methodDescriptor.streamResponse("two")); - serverStreamCreation.stream.close(Status.OK, new Metadata()); - - // Expect one transaction from the server for each message. - QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(queueingClientProxy); - QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(queueingClientProxy); - QueueingOneWayBinderProxy.Transaction txClose = takeNextTransaction(queueingClientProxy); - - // Deliver messages to the client out of order! - queueingClientProxy.deliver(tx2); - queueingClientProxy.deliver(tx1); - queueingClientProxy.deliver(txClose); - - // Client should deliver messages to the application in the order sent. - InputStream msg1 = takeNextMessage(clientStreamListener.messageQueue); - assertThat(methodDescriptor.parseResponse(msg1)).isEqualTo("one"); - InputStream msg2 = takeNextMessage(clientStreamListener.messageQueue); - assertThat(methodDescriptor.parseResponse(msg2)).isEqualTo("two"); - - assertAbout(status()).that(clientStreamListener.awaitClose(TIMEOUT_MS, MILLISECONDS)).isOk(); - assertAbout(status()) - .that(serverStreamCreation.listener.awaitClose(TIMEOUT_MS, MILLISECONDS)) - .isOk(); + serverStreamCreation.stream.flush(); + + // Expect one transaction for headers and one for each message. + try (QueueingOneWayBinderProxy.Transaction txHeaders = + takeNextTransaction(queueingClientProxy); + QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(queueingClientProxy); + QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(queueingClientProxy)) { + queueingClientProxy.deliver(txHeaders); + // Simulate tx1 being silently dropped. Receiving tx2 exposes the gap. + queueingClientProxy.deliver(tx2); + } + + assertTransactionSequenceGap( + clientStreamListener.awaitClose(TIMEOUT_MS, MILLISECONDS), 1, 2); + assertTransactionSequenceGap( + serverStreamCreation.listener.awaitClose(TIMEOUT_MS, MILLISECONDS), 1, 2); } @Test - public void msgFragmentsDeliveredToClientOutOfOrder() throws Exception { + public void clientFailsFastOnGapBetweenMessageFragments() throws Exception { server = newServerBuilder().setClientBinderDecorator(blockingDecorator).build(); registerServerWithRobolectric((BinderServer) server); server.start(serverListener); @@ -653,20 +626,35 @@ public void msgFragmentsDeliveredToClientOutOfOrder() throws Exception { serverTransportListener.takeStreamOrFail(TIMEOUT_MS, MILLISECONDS); String largeMessage = newStringOfLength(BlockPool.BLOCK_SIZE + 1); + serverStreamCreation.stream.writeHeaders(new Metadata(), true); serverStreamCreation.stream.writeMessage(methodDescriptor.streamResponse(largeMessage)); serverStreamCreation.stream.flush(); - // Expect the client to split largeMessage into two transactions. - QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(queueingClientProxy); - QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(queueingClientProxy); + // Expect one transaction for headers and two for largeMessage. + try (QueueingOneWayBinderProxy.Transaction txHeaders = + takeNextTransaction(queueingClientProxy); + QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(queueingClientProxy); + QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(queueingClientProxy)) { + queueingClientProxy.deliver(txHeaders); + // Simulate the first fragment being silently dropped. + queueingClientProxy.deliver(tx2); + } - // Deliver them to the client out of order! - queueingClientProxy.deliver(tx2); - queueingClientProxy.deliver(tx1); + assertTransactionSequenceGap( + clientStreamListener.awaitClose(TIMEOUT_MS, MILLISECONDS), 1, 2); + assertTransactionSequenceGap( + serverStreamCreation.listener.awaitClose(TIMEOUT_MS, MILLISECONDS), 1, 2); + } - // Client should reassemble the message correctly. - InputStream msg = takeNextMessage(clientStreamListener.messageQueue); - assertThat(methodDescriptor.parseResponse(msg)).isEqualTo(largeMessage); + private static void assertTransactionSequenceGap( + Status status, int expectedIndex, int receivedIndex) { + assertThat(status.getCode()).isEqualTo(Status.Code.UNAVAILABLE); + assertThat(status.getDescription()) + .isEqualTo( + "Binder transaction sequence gap: expected " + + expectedIndex + + ", received " + + receivedIndex); } private static OneWayBinderProxy takeNextBinder( @@ -683,13 +671,6 @@ private static QueueingOneWayBinderProxy.Transaction takeNextTransaction( return tx; } - private static InputStream takeNextMessage(BlockingQueue messageQueue) - throws InterruptedException { - InputStream msg = messageQueue.poll(TIMEOUT_MS, MILLISECONDS); - assertThat(msg).isNotNull(); - return msg; - } - private static String newStringOfLength(int numChars) { char[] chars = new char[numChars]; java.util.Arrays.fill(chars, 'x'); diff --git a/binder/src/testFixtures/java/io/grpc/binder/internal/OneWayBinderProxies.java b/binder/src/testFixtures/java/io/grpc/binder/internal/OneWayBinderProxies.java index c7eee06e73a..d2d41fe73d0 100644 --- a/binder/src/testFixtures/java/io/grpc/binder/internal/OneWayBinderProxies.java +++ b/binder/src/testFixtures/java/io/grpc/binder/internal/OneWayBinderProxies.java @@ -134,7 +134,7 @@ public void transact(int code, ParcelHolder data) throws RemoteException { /** A {@link OneWayBinderProxy} that queues transactions for a test to deliver manually later. */ public static final class QueueingOneWayBinderProxy extends OneWayBinderProxy { - public static final class Transaction { + public static final class Transaction implements AutoCloseable { public final int code; private final ParcelHolder parcel; @@ -142,6 +142,12 @@ public Transaction(int code, ParcelHolder parcel) { this.code = code; this.parcel = parcel; } + + /** Discards this transaction and recycles its parcel if it hasn't already been delivered. */ + @Override + public void close() { + parcel.close(); + } } private final BlockingQueue queue = new LinkedBlockingQueue<>();