Skip to content

binder: Fail fast on transaction sequence gaps #13008

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
NianJiuZst wants to merge 1 commit into
base: master
Choose a base branch
from
+109 −129
Open

binder: Fail fast on transaction sequence gaps #13008

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 26 additions & 33 deletions binder/src/main/java/io/grpc/binder/internal/Inbound.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,8 +39,6 @@
/**
* Handles incoming binder transactions for a single stream, turning those transactions into calls
* to the stream listener.
*
* <p>Out-of-order messages are reassembled into their correct order.
*/
abstract class Inbound<L extends StreamListener, T extends BinderTransport>
implements StreamListener.MessageProducer {
Expand Down Expand Up @@ -74,6 +72,9 @@ abstract class Inbound<L extends StreamListener, T extends BinderTransport>
@GuardedBy("this")
private int firstQueuedTransactionIndex;

@GuardedBy("this")
private int nextExpectedTransactionIndex;

@GuardedBy("this")
private int nextCompleteMessageEnd;

Expand Down Expand Up @@ -102,9 +103,7 @@ abstract class Inbound<L extends StreamListener, T extends BinderTransport>
* 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.
Expand Down Expand Up @@ -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 "

@jdcormie jdcormie Aug 28, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

gap isn't quite the right word. It could be a gap but it could also be a duplicate. How about "out-of-order transaction" ?

+ 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);
Expand All @@ -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);
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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")
Expand All @@ -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;
}
}
}
}
Expand Down
4 changes: 2 additions & 2 deletions binder/src/main/java/io/grpc/binder/internal/Outbound.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
167 changes: 74 additions & 93 deletions binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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()
Expand All @@ -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()
Expand All @@ -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);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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(
Expand All @@ -683,13 +671,6 @@ private static QueueingOneWayBinderProxy.Transaction takeNextTransaction(
return tx;
}

private static InputStream takeNextMessage(BlockingQueue<InputStream> 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');
Expand Down
Loading
Loading