binder: Add unit tests for ServerInbound message reassembly

Author: jdcormie

binder/build.gradle

@@ -54,6 +54,7 @@ dependencies {
     testImplementation project(':grpc-testing')
     testImplementation project(':grpc-inprocess')
     testImplementation testFixtures(project(':grpc-core'))
+    testImplementation testFixtures(project(':grpc-api'))
 
     androidTestAnnotationProcessor libraries.auto.value
     androidTestImplementation project(':grpc-testing')

binder/src/main/java/io/grpc/binder/internal/BlockPool.java

@@ -40,7 +40,7 @@ final class BlockPool {
    * The size of each standard block. (Currently 16k) The block size must be at least as large as
    * the maximum header list size.
    */
-  private static final int BLOCK_SIZE = Math.max(16 * 1024, GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE);
+  static final int BLOCK_SIZE = Math.max(16 * 1024, GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE);
 
   /**
    * Maximum number of blocks to keep around. (Max 128k). This limit is a judgement call. 128k is

binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java

@@ -18,12 +18,14 @@
 
 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.binder.internal.BinderTransport.REMOTE_UID;
 import static io.grpc.binder.internal.BinderTransport.SETUP_TRANSPORT;
 import static io.grpc.binder.internal.BinderTransport.SHUTDOWN_TRANSPORT;
 import static io.grpc.binder.internal.BinderTransport.WIRE_FORMAT_VERSION;
+import static io.grpc.testing.StatusSubject.status;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.junit.Assume.assumeTrue;
 import static org.mockito.ArgumentMatchers.any;
@@ -47,15 +49,20 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.truth.TruthJUnit;
 import io.grpc.Attributes;
+import io.grpc.CallOptions;
 import io.grpc.InternalChannelz.SocketStats;
+import io.grpc.Metadata;
 import io.grpc.ServerStreamTracer;
 import io.grpc.Status;
 import io.grpc.binder.AndroidComponentAddress;
 import io.grpc.binder.ApiConstants;
 import io.grpc.binder.AsyncSecurityPolicy;
 import io.grpc.binder.SecurityPolicies;
+import io.grpc.binder.internal.OneWayBinderProxies.*;
 import io.grpc.binder.internal.SettableAsyncSecurityPolicy.AuthRequest;
 import io.grpc.internal.AbstractTransportTest;
+import io.grpc.internal.ClientStream;
+import io.grpc.internal.ClientStreamListenerBase;
 import io.grpc.internal.ClientTransport;
 import io.grpc.internal.ClientTransportFactory.ClientTransportOptions;
 import io.grpc.internal.ConnectionClientTransport;
@@ -66,7 +73,9 @@
 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;
@@ -124,6 +133,8 @@ public final class RobolectricBinderTransportTest extends AbstractTransportTest
   ServiceInfo serviceInfo;
 
   private int nextServerAddress;
+  private BlockingBinderDecorator<OneWayBinderProxy> blockingDecorator =
+      new BlockingBinderDecorator<>();
 
   @Parameter(value = 0)
   public boolean preAuthServersParam;
@@ -433,4 +444,148 @@ public void flowControlPushBack() {}
   @Ignore("See BinderTransportTest#serverAlreadyListening")
   @Override
   public void serverAlreadyListening() {}
+
+  @Test
+  public void singleTxnMsgsDeliveredToServerOutOfOrder() throws Exception {
+    server.start(serverListener);
+    client =
+        newClientTransportBuilder()
+            .setFactory(
+                newClientTransportFactoryBuilder()
+                    .setBinderDecorator(blockingDecorator)
+                    .buildClientTransportFactory())
+            .build();
+    runIfNotNull(client.start(mockClientTransportListener));
+    blockingDecorator.putNextResult(takeNextBinder(blockingDecorator)); // Endpoint binder.
+    QueueingOneWayBinderProxy queueingServerProxy =
+        new QueueingOneWayBinderProxy(takeNextBinder(blockingDecorator)); // Server binder.
+    blockingDecorator.putNextResult(queueingServerProxy);
+
+    verify(mockClientTransportListener, timeout(TIMEOUT_MS)).transportReady();
+
+    ClientStream stream =
+        client.newStream(methodDescriptor, new Metadata(), CallOptions.DEFAULT, noopTracers);
+    ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
+    stream.start(clientStreamListener);
+    stream.writeMessage(methodDescriptor.streamRequest("one"));
+    stream.writeMessage(methodDescriptor.streamRequest("two"));
+    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);
+
+    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();
+  }
+
+  @Test
+  public void msgFragmentsDeliveredToServerOutOfOrder() throws Exception {
+    server.start(serverListener);
+    client =
+        newClientTransportBuilder()
+            .setFactory(
+                newClientTransportFactoryBuilder()
+                    .setBinderDecorator(blockingDecorator)
+                    .buildClientTransportFactory())
+            .build();
+    runIfNotNull(client.start(mockClientTransportListener));
+    blockingDecorator.putNextResult(takeNextBinder(blockingDecorator)); // Endpoint binder.
+    QueueingOneWayBinderProxy queueingServerProxy =
+        new QueueingOneWayBinderProxy(takeNextBinder(blockingDecorator)); // Server binder.
+    blockingDecorator.putNextResult(queueingServerProxy);
+
+    verify(mockClientTransportListener, timeout(TIMEOUT_MS)).transportReady();
+
+    ClientStream stream =
+        client.newStream(methodDescriptor, new Metadata(), CallOptions.DEFAULT, noopTracers);
+    ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
+    stream.start(clientStreamListener);
+
+    String largeMessage = newStringOfLength(BlockPool.BLOCK_SIZE + 1);
+    stream.writeMessage(methodDescriptor.streamRequest(largeMessage));
+    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.
+    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();
+  }
+
+  private static OneWayBinderProxy takeNextBinder(
+      BlockingBinderDecorator<OneWayBinderProxy> decorator) throws InterruptedException {
+    OneWayBinderProxy proxy = decorator.takeNextRequest(TIMEOUT_MS, MILLISECONDS);
+    assertThat(proxy).isNotNull();
+    return proxy;
+  }
+
+  private static QueueingOneWayBinderProxy.Transaction takeNextTransaction(
+      QueueingOneWayBinderProxy proxy) throws InterruptedException {
+    QueueingOneWayBinderProxy.Transaction tx = proxy.pollNextTransaction(TIMEOUT_MS, MILLISECONDS);
+    assertThat(tx).isNotNull();
+    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');
+    return new String(chars);
+  }
 }

binder/src/testFixtures/java/io/grpc/binder/internal/OneWayBinderProxies.java

@@ -18,6 +18,7 @@
 import android.os.RemoteException;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
 
 /** A collection of {@link OneWayBinderProxy}-related test helpers. */
@@ -42,6 +43,18 @@ public OneWayBinderProxy takeNextRequest() throws InterruptedException {
       return requests.take();
     }
 
+    /**
+     * Returns the next {@link OneWayBinderProxy} that needs decorating, blocking for up to the
+     * specified timeout if it hasn't yet been provided to {@link #decorate}.
+     *
+     * <p>Follow this with a call to {@link #putNextResult(OneWayBinderProxy)} to provide the result
+     * of {@link #decorate} and unblock the waiting caller.
+     */
+    public OneWayBinderProxy takeNextRequest(long timeout, TimeUnit unit)
+        throws InterruptedException {
+      return requests.poll(timeout, unit);
+    }
+
     /** Provides the next value to return from {@link #decorate}. */
     public void putNextResult(T next) throws InterruptedException {
       results.put(next);

core/src/testFixtures/java/io/grpc/internal/AbstractTransportTest.java

@@ -185,7 +185,7 @@ public void log(ChannelLogLevel level, String messageFormat, Object... args) {}
   protected final ClientStreamTracer[] tracers = new ClientStreamTracer[] {
       clientStreamTracer1, clientStreamTracer2
   };
-  private final ClientStreamTracer[] noopTracers = new ClientStreamTracer[] {
+  protected final ClientStreamTracer[] noopTracers = new ClientStreamTracer[] {
     new ClientStreamTracer() {}
   };
 
@@ -2195,7 +2195,7 @@ public void streamCreated(Attributes transportAttrs, Metadata metadata) {
     }
   }
 
-  private static class StringMarshaller implements MethodDescriptor.Marshaller<String> {
+  protected static class StringMarshaller implements MethodDescriptor.Marshaller<String> {
     public static final StringMarshaller INSTANCE = new StringMarshaller();
 
     @Override