Merge branch 'grpc:master' into Issue_fixed_12142

Author: Sangamesh1997

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

@@ -19,7 +19,6 @@
 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 @@ protected int writeSuffix(Parcel parcel) throws IOException {
     @GuardedBy("this")
     void setDeadline(Deadline deadline) {
       headers.discardAll(TIMEOUT_KEY);
-      long effectiveTimeoutNanos = max(0, deadline.timeRemaining(TimeUnit.NANOSECONDS));
-      headers.put(TIMEOUT_KEY, effectiveTimeoutNanos);
+      headers.put(TIMEOUT_KEY, deadline.timeRemaining(TimeUnit.NANOSECONDS));
     }
   }
 

census/src/main/java/io/grpc/census/GrpcCensus.java

@@ -0,0 +1,176 @@
+/*
+ * Copyright 2025 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.census;
+
+import com.google.common.base.Stopwatch;
+import com.google.common.base.Supplier;
+import io.grpc.ClientInterceptor;
+import io.grpc.ExperimentalApi;
+import io.grpc.ManagedChannelBuilder;
+import io.grpc.ServerBuilder;
+import io.grpc.ServerStreamTracer;
+import io.opencensus.trace.Tracing;
+
+/**
+ *  The entrypoint for OpenCensus instrumentation functionality in gRPC.
+ *
+ *  <p>GrpcCensus uses {@link io.opencensus.api.OpenCensus} APIs for instrumentation.
+ *
+ */
+@ExperimentalApi("https://github.com/grpc/grpc-java/issues/12178")
+public final class GrpcCensus {
+
+  private final boolean statsEnabled;
+  private final boolean tracingEnabled;
+
+  private GrpcCensus(Builder builder) {
+    this.statsEnabled = builder.statsEnabled;
+    this.tracingEnabled = builder.tracingEnabled;
+  }
+
+  /**
+   * Creates a new builder for {@link GrpcCensus}.
+   */
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  private static final Supplier<Stopwatch> STOPWATCH_SUPPLIER = new Supplier<Stopwatch>() {
+    @Override
+    public Stopwatch get() {
+      return Stopwatch.createUnstarted();
+    }
+  };
+
+  /**
+   * Configures a {@link ServerBuilder} to enable census stats and tracing.
+   *
+   * @param serverBuilder The server builder to configure.
+   * @return The configured server builder.
+   */
+  public <T extends ServerBuilder<T>> T configureServerBuilder(T serverBuilder) {
+    if (statsEnabled) {
+      serverBuilder.addStreamTracerFactory(newServerStatsStreamTracerFactory());
+    }
+    if (tracingEnabled) {
+      serverBuilder.addStreamTracerFactory(newServerTracingStreamTracerFactory());
+    }
+    return serverBuilder;
+  }
+
+  /**
+   * Configures a {@link ManagedChannelBuilder} to enable census stats and tracing.
+   *
+   * @param channelBuilder The channel builder to configure.
+   * @return The configured channel builder.
+   */
+  public <T extends ManagedChannelBuilder<T>> T configureChannelBuilder(T channelBuilder) {
+    if (statsEnabled) {
+      channelBuilder.intercept(newClientStatsInterceptor());
+    }
+    if (tracingEnabled) {
+      channelBuilder.intercept(newClientTracingInterceptor());
+    }
+    return channelBuilder;
+  }
+
+  /**
+   * Returns a {@link ClientInterceptor} with default stats implementation.
+   */
+  private static ClientInterceptor newClientStatsInterceptor() {
+    CensusStatsModule censusStats =
+        new CensusStatsModule(
+            STOPWATCH_SUPPLIER,
+            true,
+            true,
+            true,
+            false,
+            true);
+    return censusStats.getClientInterceptor();
+  }
+
+  /**
+   * Returns a {@link ClientInterceptor} with default tracing implementation.
+   */
+  private static ClientInterceptor newClientTracingInterceptor() {
+    CensusTracingModule censusTracing =
+        new CensusTracingModule(
+            Tracing.getTracer(),
+            Tracing.getPropagationComponent().getBinaryFormat());
+    return censusTracing.getClientInterceptor();
+  }
+
+  /**
+   * Returns a {@link ServerStreamTracer.Factory} with default stats implementation.
+   */
+  private static ServerStreamTracer.Factory newServerStatsStreamTracerFactory() {
+    CensusStatsModule censusStats =
+        new CensusStatsModule(
+            STOPWATCH_SUPPLIER,
+            true,
+            true,
+            true,
+            false,
+            true);
+    return censusStats.getServerTracerFactory();
+  }
+
+  /**
+   * Returns a {@link ServerStreamTracer.Factory} with default tracing implementation.
+   */
+  private static ServerStreamTracer.Factory newServerTracingStreamTracerFactory() {
+    CensusTracingModule censusTracing =
+        new CensusTracingModule(
+            Tracing.getTracer(),
+            Tracing.getPropagationComponent().getBinaryFormat());
+    return censusTracing.getServerTracerFactory();
+  }
+
+  /**
+   * Builder for {@link GrpcCensus}.
+   */
+  public static final class Builder {
+    private boolean statsEnabled = true;
+    private boolean tracingEnabled = true;
+
+    private Builder() {
+    }
+
+    /**
+     * Disables stats collection.
+     */
+    public Builder disableStats() {
+      this.statsEnabled = false;
+      return this;
+    }
+
+    /**
+     * Disables tracing.
+     */
+    public Builder disableTracing() {
+      this.tracingEnabled = false;
+      return this;
+    }
+
+    /**
+     * Builds a new {@link GrpcCensus}.
+     */
+    public GrpcCensus build() {
+      return new GrpcCensus(this);
+    }
+  }
+}

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

@@ -21,7 +21,6 @@
 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 @@ protected AbstractClientStream(
   @Override
   public void setDeadline(Deadline deadline) {
     headers.discardAll(TIMEOUT_KEY);
-    long effectiveTimeout = max(0, deadline.timeRemaining(TimeUnit.NANOSECONDS));
-    headers.put(TIMEOUT_KEY, effectiveTimeout);
+    headers.put(TIMEOUT_KEY, deadline.timeRemaining(TimeUnit.NANOSECONDS));
   }
 
   @Override

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

@@ -651,12 +651,14 @@ public Stopwatch get() {
   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) {
-        throw new IllegalArgumentException("Timeout too small");
-      } else if (timeoutNanos < cutoff) {
+      if (timeoutNanos < cutoff) {
         return timeoutNanos + "n";
       } else if (timeoutNanos < cutoff * 1000L) {
         return unit.toMicros(timeoutNanos) + "u";

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

@@ -99,7 +99,7 @@ public static ServerBuilder<?> forPort(int port) {
   ServerCallExecutorSupplier executorSupplier;
 
   /**
-   * An interface to provide to provide transport specific information for the server. This method
+   * An interface to provide transport specific information for the server. This method
    * is meant for Transport implementors and should not be used by normal users.
    */
   public interface ClientTransportServersBuilder {

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

@@ -465,6 +465,24 @@ allocator, new BaseTransportState(statsTraceCtx, transportTracer), sink, statsTr
         .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 void timeoutTest() {
     GrpcUtil.TimeoutMarshaller marshaller =
             new GrpcUtil.TimeoutMarshaller();
     // nanos
-    assertEquals("0n", marshaller.toAsciiString(0L));
-    assertEquals(0L, (long) marshaller.parseAsciiString("0n"));
+    assertEquals("1n", marshaller.toAsciiString(1L));
+    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.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 ServerCall.Listener<String> startCall(
   @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 @@
 
 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 @@ public void setMaxOutboundMessageSize(int maxSize) {}
       @Override
       public void setDeadline(Deadline deadline) {
         headers.discardAll(TIMEOUT_KEY);
-        long effectiveTimeout = max(0, deadline.timeRemaining(TimeUnit.NANOSECONDS));
-        headers.put(TIMEOUT_KEY, effectiveTimeout);
+        headers.put(TIMEOUT_KEY, deadline.timeRemaining(TimeUnit.NANOSECONDS));
       }
 
       @Override