Merge branch 'issue-5033' of github.com:my4-dev/armeria into pr-5228-…

…my4-dev-issue-5033

.github/workflows/actions_build.yml

@@ -25,47 +25,32 @@ env:
   DEVELOCITY_ACCESS_KEY: ${{ secrets.DEVELOCITY_ACCESS_KEY }}
 
 jobs:
-  choose-self-hosted:
-    runs-on: ubuntu-latest
-    outputs:
-      runner: ${{ steps.runner.outputs.runner }}
-
-    steps:
-      - id: runner
-        run: |
-          if [[ "${{ github.event_name }}" == "pull_request" ]]; then
-            echo "runner=self-hosted-unsafe" >> "$GITHUB_OUTPUT"
-          else
-            echo "runner=self-hosted-safe" >> "$GITHUB_OUTPUT"
-          fi
-
   build:
-    needs: [ choose-self-hosted ]
     if: github.repository == 'line/armeria'
     runs-on: ${{ matrix.on }}
     timeout-minutes: 120
     strategy:
       fail-fast: false
       matrix:
-        on: [ "${{ needs.choose-self-hosted.outputs.runner }}", macos-12, windows-latest ]
+        on: [ ubicloud-standard-8, macos-12, windows-latest ]
         java: [ 21 ]
         include:
           - java: 8
-            on: ${{ needs.choose-self-hosted.outputs.runner }}
+            on: ubicloud-standard-8
           - java: 11
-            on: ${{ needs.choose-self-hosted.outputs.runner }}
+            on: ubicloud-standard-8
           - java: 17
-            on: ${{ needs.choose-self-hosted.outputs.runner }}
+            on: ubicloud-standard-8
             leak: true
           - java: 17
-            on: ${{ needs.choose-self-hosted.outputs.runner }}
+            on: ubicloud-standard-8
             min-java: 11
           - java: 17
-            on: ${{ needs.choose-self-hosted.outputs.runner }}
+            on: ubicloud-standard-8
             min-java: 17
             coverage: true
           - java: 21
-            on: ${{ needs.choose-self-hosted.outputs.runner }}
+            on: ubicloud-standard-8
             snapshot: true
             # blockhound makes the build run about 10 minutes slower
             blockhound: true
@@ -111,7 +96,7 @@ jobs:
       - name: Build with Gradle (Shading only)
         run: |
           ./gradlew --no-daemon --stacktrace shadedJar shadedTestJar trimShadedJar \
-          ${{ startsWith(matrix.on, 'self-hosted') && '--max-workers=8' || '--max-workers=2' }} --parallel \
+          ${{ startsWith(matrix.on, 'ubicloud') && '--max-workers=8' || '--max-workers=2' }} --parallel \
           ${{ matrix.coverage && '-Pcoverage' || '' }} \
           -PnoLint \
           -PbuildJdkVersion=${{ env.BUILD_JDK_VERSION }} \
@@ -130,7 +115,7 @@ jobs:
       - name: Build with Gradle
         run: |
           ./gradlew --no-daemon --stacktrace build \
-          ${{ startsWith(matrix.on, 'self-hosted') && '--max-workers=8' || '--max-workers=2' }} --parallel \
+          ${{ startsWith(matrix.on, 'ubicloud') && '--max-workers=8' || '--max-workers=2' }} --parallel \
           ${{ matrix.coverage && '-Pcoverage' || '' }} \
           ${{ matrix.leak && '-Pleak' || '' }} \
           ${{ matrix.blockhound && '-Pblockhound' || '' }} \
@@ -255,8 +240,6 @@ jobs:
 
   site:
     if: github.repository == 'line/armeria'
-    # ubuntu-latest is preferred for site job.
-    # node_modules need complicated dependencies that are difficult to install on self-hosted runners.
     runs-on: ubuntu-latest
     timeout-minutes: 60
     steps:

.github/workflows/public-suffixes.yml

@@ -37,7 +37,7 @@ jobs:
           git_commit_gpgsign: true
 
       - name: Create pull request
-        uses: peter-evans/create-pull-request@v6
+        uses: peter-evans/create-pull-request@v7
         with:
           # The title of the pull request.
           title: Update public suffix list

.github/workflows/publish-release.yml

@@ -13,7 +13,7 @@ jobs:
   publish:
     name: Publish final artifacts
     if: github.repository == 'line/armeria'
-    runs-on: self-hosted-safe
+    runs-on: ubicloud-standard-8
     steps:
       - uses: actions/checkout@v4
 

core/src/main/java/com/linecorp/armeria/client/AbstractClientOptionsBuilder.java

@@ -508,6 +508,18 @@ public AbstractClientOptionsBuilder contextCustomizer(
         return this;
     }
 
+    /**
+     * Sets the {@link ResponseTimeoutMode} which determines when a {@link #responseTimeout(Duration)}}
+     * will start to be scheduled.
+     *
+     * @see ResponseTimeoutMode
+     */
+    @UnstableApi
+    public AbstractClientOptionsBuilder responseTimeoutMode(ResponseTimeoutMode responseTimeoutMode) {
+        return option(ClientOptions.RESPONSE_TIMEOUT_MODE,
+                      requireNonNull(responseTimeoutMode, "responseTimeoutMode"));
+    }
+
     /**
      * Builds {@link ClientOptions} with the given options and the
      * {@linkplain ClientOptions#of() default options}.

core/src/main/java/com/linecorp/armeria/client/AbstractHttpRequestHandler.java

@@ -45,6 +45,8 @@
 import com.linecorp.armeria.internal.client.ClientRequestContextExtension;
 import com.linecorp.armeria.internal.client.DecodedHttpResponse;
 import com.linecorp.armeria.internal.client.HttpSession;
+import com.linecorp.armeria.internal.common.CancellationScheduler;
+import com.linecorp.armeria.internal.common.CancellationScheduler.CancellationTask;
 import com.linecorp.armeria.internal.common.RequestContextUtil;
 import com.linecorp.armeria.unsafe.PooledObjects;
 
@@ -89,6 +91,7 @@ enum State {
     private ScheduledFuture<?> timeoutFuture;
     private State state = State.NEEDS_TO_WRITE_FIRST_HEADER;
     private boolean loggedRequestFirstBytesTransferred;
+    private boolean failed;
 
     AbstractHttpRequestHandler(Channel ch, ClientHttpObjectEncoder encoder, HttpResponseDecoder responseDecoder,
                                DecodedHttpResponse originalRes,
@@ -192,9 +195,33 @@ final boolean tryInitialize() {
                     () -> failAndReset(WriteTimeoutException.get()),
                     timeoutMillis, TimeUnit.MILLISECONDS);
         }
+        final CancellationScheduler scheduler = cancellationScheduler();
+        if (scheduler != null) {
+            scheduler.updateTask(newCancellationTask());
+            if (ctx.responseTimeoutMode() == ResponseTimeoutMode.CONNECTION_ACQUIRED) {
+                scheduler.start();
+            }
+        }
+        if (ctx.isCancelled()) {
+            // The previous cancellation task wraps the cause with an UnprocessedRequestException
+            // so we return early
+            return false;
+        }
         return true;
     }
 
+    private CancellationTask newCancellationTask() {
+        return cause -> {
+            if (ch.eventLoop().inEventLoop()) {
+                try (SafeCloseable ignored = RequestContextUtil.pop()) {
+                    failAndReset(cause);
+                }
+            } else {
+                ch.eventLoop().execute(() -> failAndReset(cause));
+            }
+        };
+    }
+
     RequestHeaders mergedRequestHeaders(RequestHeaders headers) {
         final HttpHeaders internalHeaders;
         final ClientRequestContextExtension ctxExtension = ctx.as(ClientRequestContextExtension.class);
@@ -354,6 +381,10 @@ final void failRequest(Throwable cause) {
     }
 
     private void fail(Throwable cause) {
+        if (failed) {
+            return;
+        }
+        failed = true;
         state = State.DONE;
         cancel();
         logBuilder.endRequest(cause);
@@ -368,9 +399,20 @@ private void fail(Throwable cause) {
             logBuilder.endResponse(cause);
             originalRes.close(cause);
         }
+
+        final CancellationScheduler scheduler = cancellationScheduler();
+        if (scheduler != null) {
+            // best-effort attempt to cancel the scheduled timeout task so that RequestContext#cause
+            // isn't set unnecessarily
+            scheduler.cancelScheduled();
+        }
     }
 
     final void failAndReset(Throwable cause) {
+        if (failed) {
+            return;
+        }
+
         if (cause instanceof WriteTimeoutException) {
             final HttpSession session = HttpSession.get(ch);
             // Mark the session as unhealthy so that subsequent requests do not use it.
@@ -394,7 +436,7 @@ final void failAndReset(Throwable cause) {
             error = Http2Error.INTERNAL_ERROR;
         }
 
-        if (error.code() != Http2Error.CANCEL.code()) {
+        if (error.code() != Http2Error.CANCEL.code() && cause != ctx.cancellationCause()) {
             Exceptions.logIfUnexpected(logger, ch,
                                        HttpSession.get(ch).protocol(),
                                        "a request publisher raised an exception", cause);
@@ -415,4 +457,13 @@ final boolean cancelTimeout() {
         this.timeoutFuture = null;
         return timeoutFuture.cancel(false);
     }
+
+    @Nullable
+    private CancellationScheduler cancellationScheduler() {
+        final ClientRequestContextExtension ctxExt = ctx.as(ClientRequestContextExtension.class);
+        if (ctxExt != null) {
+            return ctxExt.responseCancellationScheduler();
+        }
+        return null;
+    }
 }

core/src/main/java/com/linecorp/armeria/client/BlockingWebClientRequestPreparation.java

@@ -223,6 +223,12 @@ public BlockingWebClientRequestPreparation exchangeType(ExchangeType exchangeTyp
         return this;
     }
 
+    @Override
+    public BlockingWebClientRequestPreparation responseTimeoutMode(ResponseTimeoutMode responseTimeoutMode) {
+        delegate.responseTimeoutMode(responseTimeoutMode);
+        return this;
+    }
+
     @Override
     public BlockingWebClientRequestPreparation requestOptions(RequestOptions requestOptions) {
         delegate.requestOptions(requestOptions);

core/src/main/java/com/linecorp/armeria/client/ClientBuilder.java

@@ -295,4 +295,9 @@ public ClientBuilder contextCustomizer(
     public ClientBuilder contextHook(Supplier<? extends AutoCloseable> contextHook) {
         return (ClientBuilder) super.contextHook(contextHook);
     }
+
+    @Override
+    public ClientBuilder responseTimeoutMode(ResponseTimeoutMode responseTimeoutMode) {
+        return (ClientBuilder) super.responseTimeoutMode(responseTimeoutMode);
+    }
 }

core/src/main/java/com/linecorp/armeria/client/ClientFactoryBuilder.java

@@ -64,6 +64,7 @@
 import com.linecorp.armeria.common.TlsSetters;
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.annotation.UnstableApi;
+import com.linecorp.armeria.common.outlier.OutlierDetection;
 import com.linecorp.armeria.common.util.EventLoopGroups;
 import com.linecorp.armeria.common.util.TlsEngineType;
 import com.linecorp.armeria.internal.common.IgnoreHostsTrustManager;
@@ -864,6 +865,18 @@ public ClientFactoryBuilder connectionPoolListener(
         return this;
     }
 
+    /**
+     * Sets the {@link OutlierDetection} which is used to detect unhealthy connections.
+     * If an unhealthy connection is detected, it is disabled and a new connection will be created.
+     * This option is disabled by default.
+     */
+    @UnstableApi
+    public ClientFactoryBuilder connectionOutlierDetection(OutlierDetection outlierDetection) {
+        option(ClientFactoryOptions.CONNECTION_OUTLIER_DETECTION,
+               requireNonNull(outlierDetection, "outlierDetection"));
+        return this;
+    }
+
     /**
      * Sets the graceful connection shutdown timeout in milliseconds.
      * {@code 0} disables the timeout and closes the connection immediately after sending a GOAWAY frame.

core/src/main/java/com/linecorp/armeria/client/ClientFactoryOptions.java

@@ -38,6 +38,7 @@
 import com.linecorp.armeria.common.TlsKeyPair;
 import com.linecorp.armeria.common.TlsProvider;
 import com.linecorp.armeria.common.annotation.UnstableApi;
+import com.linecorp.armeria.common.outlier.OutlierDetection;
 import com.linecorp.armeria.common.util.AbstractOptions;
 import com.linecorp.armeria.common.util.TlsEngineType;
 import com.linecorp.armeria.internal.common.util.ChannelUtil;
@@ -215,6 +216,15 @@ public final class ClientFactoryOptions
     public static final ClientFactoryOption<Long> MAX_CONNECTION_AGE_MILLIS =
             ClientFactoryOption.define("MAX_CONNECTION_AGE_MILLIS", clampedDefaultMaxClientConnectionAge());
 
+    /**
+     * The {@link OutlierDetection} which is used to detect unhealthy connections.
+     * If an unhealthy connection is detected, it is disabled and a new connection will be created.
+     * This option is disabled by default.
+     */
+    @UnstableApi
+    public static final ClientFactoryOption<OutlierDetection> CONNECTION_OUTLIER_DETECTION =
+            ClientFactoryOption.define("CONNECTION_OUTLIER_DETECTION", OutlierDetection.disabled());
+
     private static long clampedDefaultMaxClientConnectionAge() {
         final long connectionAgeMillis = Flags.defaultMaxClientConnectionAgeMillis();
         if (connectionAgeMillis > 0 && connectionAgeMillis < MIN_MAX_CONNECTION_AGE_MILLIS) {
@@ -602,6 +612,14 @@ public ConnectionPoolListener connectionPoolListener() {
         return get(CONNECTION_POOL_LISTENER);
     }
 
+    /**
+     * Returns the {@link OutlierDetection} which is used to detect unhealthy connections.
+     */
+    @UnstableApi
+    public OutlierDetection connectionOutlierDetection() {
+        return get(CONNECTION_OUTLIER_DETECTION);
+    }
+
     /**
      * Returns the graceful connection shutdown timeout in milliseconds.
      */

core/src/main/java/com/linecorp/armeria/client/ClientOptions.java

@@ -156,6 +156,10 @@ public final class ClientOptions
     public static final ClientOption<Supplier<? extends AutoCloseable>> CONTEXT_HOOK =
             ClientOption.define("CONTEXT_HOOK", NOOP_CONTEXT_HOOK);
 
+    @UnstableApi
+    public static final ClientOption<ResponseTimeoutMode> RESPONSE_TIMEOUT_MODE =
+            ClientOption.define("RESPONSE_TIMEOUT_MODE", Flags.responseTimeoutMode());
+
     private static final List<AsciiString> PROHIBITED_HEADER_NAMES = ImmutableList.of(
             HttpHeaderNames.HTTP2_SETTINGS,
             HttpHeaderNames.METHOD,
@@ -395,6 +399,17 @@ public Supplier<AutoCloseable> contextHook() {
         return (Supplier<AutoCloseable>) get(CONTEXT_HOOK);
     }
 
+    /**
+     * Returns the {@link ResponseTimeoutMode} which determines when a {@link #responseTimeoutMillis()}
+     * will start to be scheduled.
+     *
+     * @see ResponseTimeoutMode
+     */
+    @UnstableApi
+    public ResponseTimeoutMode responseTimeoutMode() {
+        return get(RESPONSE_TIMEOUT_MODE);
+    }
+
     /**
      * Returns a new {@link ClientOptionsBuilder} created from this {@link ClientOptions}.
      */

core/src/main/java/com/linecorp/armeria/client/ClientOptionsBuilder.java

@@ -222,4 +222,9 @@ public ClientOptionsBuilder contextCustomizer(
     public ClientOptionsBuilder contextHook(Supplier<? extends AutoCloseable> contextHook) {
         return (ClientOptionsBuilder) super.contextHook(contextHook);
     }
+
+    @Override
+    public ClientOptionsBuilder responseTimeoutMode(ResponseTimeoutMode responseTimeoutMode) {
+        return (ClientOptionsBuilder) super.responseTimeoutMode(responseTimeoutMode);
+    }
 }