core: grpc-timeout should always be positive (#12201)

PROTOCOL-HTTP2.md specifies "TimeoutValue → {positive integer as ASCII
string of at most 8 digits}". Zero is not positive, so it should be
avoided. So make sure timeouts are at least 1 nanosecond instead of 0
nanoseconds.

grpc-go recently began disallowing zero timeouts in
https://github.com/grpc/grpc-go/pull/8290 which caused a regression as
grpc-java can generate such timeouts. Apparently no gRPC implementation
had previously been checking for zero timeouts.

Instead of changing the max(0) to max(1) everywhere, just move the max
handling into TimeoutMarshaller, since every caller of TIMEOUT_KEY was
doing the same max() handling.

Before fd8fd517d (in 2016!), grpc-java actually behaved correctly, as it
failed RPCs with timeouts "<= 0". The commit changed the handling to the
max(0) handling we see now.

b/427338711

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

@@ -19,7 +19,6 @@ package io.grpc.binder.internal;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 import static io.grpc.internal.GrpcUtil.TIMEOUT_KEY;
-import static java.lang.Math.max;
 
 import android.os.Parcel;
 import com.google.errorprone.annotations.concurrent.GuardedBy;
@@ -397,8 +396,7 @@ abstract class Outbound {
     @GuardedBy("this")
     void setDeadline(Deadline deadline) {
       headers.discardAll(TIMEOUT_KEY);
-      long effectiveTimeoutNanos = max(0, deadline.timeRemaining(TimeUnit.NANOSECONDS));
+      headers.put(TIMEOUT_KEY, deadline.timeRemaining(TimeUnit.NANOSECONDS));
-      headers.put(TIMEOUT_KEY, effectiveTimeoutNanos);
     }
   }
 

core/src/main/java/io/grpc/internal/AbstractClientStream.java

@@ -21,7 +21,6 @@ import static com.google.common.base.Preconditions.checkState;
 import static io.grpc.internal.GrpcUtil.CONTENT_ENCODING_KEY;
 import static io.grpc.internal.GrpcUtil.MESSAGE_ENCODING_KEY;
 import static io.grpc.internal.GrpcUtil.TIMEOUT_KEY;
-import static java.lang.Math.max;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
@@ -124,8 +123,7 @@ public abstract class AbstractClientStream extends AbstractStream
   @Override
   public void setDeadline(Deadline deadline) {
     headers.discardAll(TIMEOUT_KEY);
-    long effectiveTimeout = max(0, deadline.timeRemaining(TimeUnit.NANOSECONDS));
+    headers.put(TIMEOUT_KEY, deadline.timeRemaining(TimeUnit.NANOSECONDS));
-    headers.put(TIMEOUT_KEY, effectiveTimeout);
   }
 
   @Override

core/src/main/java/io/grpc/internal/GrpcUtil.java

@@ -651,12 +651,14 @@ public final class GrpcUtil {
   static class TimeoutMarshaller implements Metadata.AsciiMarshaller<Long> {
 
     @Override
-    public String toAsciiString(Long timeoutNanos) {
+    public String toAsciiString(Long timeoutNanosObject) {
       long cutoff = 100000000;
+      // Timeout checking is inherently racy. RPCs with timeouts in the past ideally don't even get
+      // here, but if the timeout is expired assume that happened recently and adjust it to the
+      // smallest allowed timeout
+      long timeoutNanos = Math.max(1, timeoutNanosObject);
       TimeUnit unit = TimeUnit.NANOSECONDS;
-      if (timeoutNanos < 0) {
+      if (timeoutNanos < cutoff) {
-        throw new IllegalArgumentException("Timeout too small");
-      } else if (timeoutNanos < cutoff) {
         return timeoutNanos + "n";
       } else if (timeoutNanos < cutoff * 1000L) {
         return unit.toMicros(timeoutNanos) + "u";

core/src/test/java/io/grpc/internal/AbstractClientStreamTest.java

@@ -465,6 +465,24 @@ public class AbstractClientStreamTest {
         .isGreaterThan(TimeUnit.MILLISECONDS.toNanos(600));
   }
 
+  @Test
+  public void setDeadline_thePastBecomesPositive() {
+    AbstractClientStream.Sink sink = mock(AbstractClientStream.Sink.class);
+    ClientStream stream = new BaseAbstractClientStream(
+        allocator, new BaseTransportState(statsTraceCtx, transportTracer), sink, statsTraceCtx,
+        transportTracer);
+
+    stream.setDeadline(Deadline.after(-1, TimeUnit.NANOSECONDS));
+    stream.start(mockListener);
+
+    ArgumentCaptor<Metadata> headersCaptor = ArgumentCaptor.forClass(Metadata.class);
+    verify(sink).writeHeaders(headersCaptor.capture(), ArgumentMatchers.<byte[]>any());
+
+    Metadata headers = headersCaptor.getValue();
+    assertThat(headers.get(Metadata.Key.of("grpc-timeout", Metadata.ASCII_STRING_MARSHALLER)))
+        .isEqualTo("1n");
+  }
+
   @Test
   public void appendTimeoutInsight() {
     InsightBuilder insight = new InsightBuilder();

core/src/test/java/io/grpc/internal/GrpcUtilTest.java

@@ -98,8 +98,8 @@ public class GrpcUtilTest {
     GrpcUtil.TimeoutMarshaller marshaller =
             new GrpcUtil.TimeoutMarshaller();
     // nanos
-    assertEquals("0n", marshaller.toAsciiString(0L));
+    assertEquals("1n", marshaller.toAsciiString(1L));
-    assertEquals(0L, (long) marshaller.parseAsciiString("0n"));
+    assertEquals(1L, (long) marshaller.parseAsciiString("1n"));
 
     assertEquals("99999999n", marshaller.toAsciiString(99999999L));
     assertEquals(99999999L, (long) marshaller.parseAsciiString("99999999n"));

core/src/test/java/io/grpc/internal/ServerImplTest.java

@@ -53,6 +53,7 @@ import io.grpc.BinaryLog;
 import io.grpc.Channel;
 import io.grpc.Compressor;
 import io.grpc.Context;
+import io.grpc.Deadline;
 import io.grpc.Grpc;
 import io.grpc.HandlerRegistry;
 import io.grpc.IntegerMarshaller;
@@ -1146,11 +1147,21 @@ public class ServerImplTest {
   @Test
   public void testContextExpiredBeforeStreamCreate_StreamCancelNotCalledBeforeSetListener()
       throws Exception {
+    builder.ticker = new Deadline.Ticker() {
+      private long time;
+
+      @Override
+      public long nanoTime() {
+        time += 1000;
+        return time;
+      }
+    };
+
     AtomicBoolean contextCancelled = new AtomicBoolean(false);
     AtomicReference<Context> context = new AtomicReference<>();
     AtomicReference<ServerCall<String, Integer>> callReference = new AtomicReference<>();
 
-    testStreamClose_setup(callReference, context, contextCancelled, 0L);
+    testStreamClose_setup(callReference, context, contextCancelled, 1L);
 
     // This assert that stream.setListener(jumpListener) is called before stream.cancel(), which
     // prevents extremely short deadlines causing NPEs.

inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java

@@ -18,7 +18,6 @@ package io.grpc.inprocess;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static io.grpc.internal.GrpcUtil.TIMEOUT_KEY;
-import static java.lang.Math.max;
 
 import com.google.common.base.MoreObjects;
 import com.google.common.io.ByteStreams;
@@ -939,8 +938,7 @@ final class InProcessTransport implements ServerTransport, ConnectionClientTrans
       @Override
       public void setDeadline(Deadline deadline) {
         headers.discardAll(TIMEOUT_KEY);
-        long effectiveTimeout = max(0, deadline.timeRemaining(TimeUnit.NANOSECONDS));
+        headers.put(TIMEOUT_KEY, deadline.timeRemaining(TimeUnit.NANOSECONDS));
-        headers.put(TIMEOUT_KEY, effectiveTimeout);
       }
 
       @Override