Fix a leak in `HttpEncodedResponse` #5858

ikhoon · 2024-08-05T11:33:04Z

Motivation:

An HttpData produced in HttpEncodedResponse.beforeComplete() is not collected by CollectingSubscriberAndSubscription but is leaked.

  Hint: {10B, pooled, <unknown>}
  com.linecorp.armeria.common.HttpData.wrap(HttpData.java:110)
  com.linecorp.armeria.server.encoding.HttpEncodedResponse.beforeComplete(HttpEncodedResponse.java:163)
  com.linecorp.armeria.common.stream.FilteredStreamMessage.lambda$collect$0(FilteredStreamMessage.java:201)
  java.base/java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:934)
  java.base/java.util.concurrent.CompletableFuture.uniHandleStage(CompletableFuture.java:950)
  java.base/java.util.concurrent.CompletableFuture.handle(CompletableFuture.java:2340)
  com.linecorp.armeria.common.stream.FilteredStreamMessage.collect(FilteredStreamMessage.java:142)

CollectingSubscriberAndSubscription was designed to only apply filter() to the upstream.collect(). I didn't consider that an object could be published via onNext() in beforeComplete(). The purpose of CollectingSubscriberAndSubscription was to provide an optimized code path for unary calls. it didn't seem the code provides a trivial performance improvement but the implementation was complex and error-prone.

I was able to fix the code not to leak the data but I didn't want to additional complexity to it. It might be cleaner to use the Reactive Streams API instead of keeping the custom collect() implementation. There will be no change in performance for normal message sizes.

Modifications:

Remove the custom collect() implemtation in FilteredStreamMessage.

Result:

Fix a potential leak when sending compressed responses.

Motivation: An `HttpData` produced in `HttpEncodedResponse.beforeComplete()` is not collected by `CollectingSubscriberAndSubscription` and leaked. ```java Hint: {10B, pooled, <unknown>} com.linecorp.armeria.common.HttpData.wrap(HttpData.java:110) com.linecorp.armeria.server.encoding.HttpEncodedResponse.beforeComplete(HttpEncodedResponse.java:163) com.linecorp.armeria.common.stream.FilteredStreamMessage.lambda$collect$0(FilteredStreamMessage.java:201) java.base/java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:934) java.base/java.util.concurrent.CompletableFuture.uniHandleStage(CompletableFuture.java:950) java.base/java.util.concurrent.CompletableFuture.handle(CompletableFuture.java:2340) com.linecorp.armeria.common.stream.FilteredStreamMessage.collect(FilteredStreamMessage.java:142) ``` `CollectingSubscriberAndSubscription` was designed to only apply `filter()` of the `upstream.collect()`. I didn't consider that an object could be published via `onNext()` in `beforeComplete()`. `CollectingSubscriberAndSubscription` was added to provide an optimized code path for unary calls. it doesn't seem the code provides a trival performance improvement but the implemetation is complex and error-prone. I was able to fix the code not to leak the data but I didn't want to additional complexity to it. It might be cleaner to use the Reactive Streams API instead of keeping the custom `collect()` implemetation. There will be no change in performance for normal message sizes. Modifications: - Remove the custom `collect()` implemtation in `FilteredStreamMessage`. Result: Fix a potential leak when sending compressed responses.

github-actions · 2024-08-05T13:38:42Z

🔍 Build Scan® (commit: `6fae642`)

Job name	Status	Build Scan®
build-self-hosted-unsafe-jdk-8	✅	https://ge.armeria.dev/s/klkoouagfihh6
build-self-hosted-unsafe-jdk-21-snapshot-blockhound	✅	https://ge.armeria.dev/s/ibcwhhzgn47og
build-self-hosted-unsafe-jdk-17-min-java-17-coverage	✅	https://ge.armeria.dev/s/n5n34okpsjmqk
build-self-hosted-unsafe-jdk-17-min-java-11	✅	https://ge.armeria.dev/s/a43wsrnaafmcm
build-self-hosted-unsafe-jdk-17-leak	✅	https://ge.armeria.dev/s/ohe3kkalkcirs
build-self-hosted-unsafe-jdk-11	✅	https://ge.armeria.dev/s/s6cas3mwdv2xu
build-macos-12-jdk-21	✅	https://ge.armeria.dev/s/ts3gjfryluan4

jrhee17 · 2024-08-07T01:31:24Z

core/src/main/java/com/linecorp/armeria/common/stream/StreamMessage.java

@@ -752,6 +754,11 @@ default CompletableFuture<List<T>> collect(EventExecutor executor, SubscriptionO
        requireNonNull(executor, "executor");
        requireNonNull(options, "options");
        final StreamMessageCollector<T> collector = new StreamMessageCollector<>(options);
+        if (!containsNotifyCancellation(options)) {


Before setting this PR as ready for review, it would be easier to review if you explain why this change is needed in the PR description

StreamMessageCollector collects the published elements until onComplete() or onError() is called. It generally worked but I found a corner case where Subscription.cancel() made collector.collect() incomplete forever.

You can reproduce it by disabling this block and then running it below.

armeria/core/src/test/java/com/linecorp/armeria/common/stream/StreamMessageCollectingTest.java

Line 168 in a870edf

void filteredStreamMessage_cancel() {

I believe it would make more sense to specify NOTIFY_CANCELLATION for collect() method because it is used to fully consume the upstream data instead of partial ones.
CancelledSubscriptionException will make the collecting future completed exceptionally.

For context, previously, the custom collect() implementation in FilteredStreamMessage hooked a Subscription.cancel() event and completed the collecting future with the accumulated items so far.

I see, I understood the issue now. Agree with the approach 👍

jrhee17 · 2024-08-07T01:31:47Z

core/src/main/java/com/linecorp/armeria/common/stream/FuseableStreamMessage.java

                onError(ex);
+                upstream.cancel();


So in general, do we always cancel the upstream before propagating the error downstream now?

If we cancel the upstream first, cancel() may recursively call downstream.onError(CancelledSubscriptionException) which could prevent the error from being propagated to downstream.

So I think onError(ex) should be invoked first to propagate the cause to the downstream that will be eventually passed to ServerErrorHandler and RequestLog.*Cause()

So this change is unrelated to the leak, am I understanding correctly?

Right. I found the bug while fixing broken tests after removing CollectingSubscriberAndSubscription and related code.

jrhee17

Other changes look good, left a question about the early return in onComplete

jrhee17 · 2024-08-08T02:08:52Z

core/src/main/java/com/linecorp/armeria/common/stream/StreamMessage.java

@@ -752,6 +754,11 @@ default CompletableFuture<List<T>> collect(EventExecutor executor, SubscriptionO
        requireNonNull(executor, "executor");
        requireNonNull(options, "options");
        final StreamMessageCollector<T> collector = new StreamMessageCollector<>(options);
+        if (!containsNotifyCancellation(options)) {


I see, I understood the issue now. Agree with the approach 👍

jrhee17 · 2024-08-08T02:11:33Z

core/src/main/java/com/linecorp/armeria/common/stream/FuseableStreamMessage.java

                onError(ex);
+                upstream.cancel();


So this change is unrelated to the leak, am I understanding correctly?

jrhee17 · 2024-08-08T02:12:48Z

core/src/main/java/com/linecorp/armeria/common/stream/FilteredStreamMessage.java

+            if (completed) {
+                // onError(Throwable) or onComplete() has been called in filter().
+                StreamMessageUtil.closeOrAbort(filtered);
+                return;


I didn't understånd this change 😅 Shouldn't the last filtered object be also passed downstream?

We should not publish items if after calling onError() or onComplete()`.
https://github.com/reactive-streams/reactive-streams-jvm#1.7

Otherwise, did you mean something else?

I imagined FilteredStreamMessage#beforeComplete publishing an item via:

armeria/core/src/main/java/com/linecorp/armeria/common/stream/FilteredStreamMessage.java

Lines 268 to 270 in 6fae642

completed = true;

try {

beforeComplete(delegate);

where completed=true is set but onNext is still called in the following scenario:

armeria/core/src/main/java/com/linecorp/armeria/server/encoding/HttpEncodedResponse.java

Line 163 in 6fae642

subscriber.onNext(HttpData.wrap(buf));

I thought this was the scenario this PR was trying to address

I fixed this to pass StreamMessageCollectingTest.filteredStreamMessage_cancel().

armeria/core/src/test/java/com/linecorp/armeria/common/stream/StreamMessageCollectingTest.java

Lines 185 to 190 in c7aca10

protected HttpData filter(HttpData obj) {

count++;

if (count < 3) {

return obj;

} else {

subscription.cancel();

subscription.cancel() made upstream signal subscriber.onError(CancelledSubscriptionException). Afterward, subscribe.onNext() was called.

It's impossible to know what all subclasses do. So we may need to prevent onNext() from being called by beforeComplete() or beforeOnError().

@Override public void onNext(T o) { if (complete) { return; } U filtered; try { filtered = filter(o); }

By the way, only the Subscription API of FilteringSubscriber is exposed to the subclasses. It seems FilteringSubscriber.onNext() couldn't be called by beforeCompete()

By the way, only the Subscription API of FilteringSubscriber is exposed to the subclasses.

I just realized that the downstream Subscriber#onNext is called instead of FilteringSubscriber#onNext. I guess the current implementation should be fine then 👍

It's not nice that a Subscriber is exposed through beforeComplete(). We may need to redesign the API later.

minwoox

👍 👍 👍

minwoox · 2024-08-08T02:37:05Z

core/src/main/java/com/linecorp/armeria/common/stream/FilteredStreamMessage.java

@@ -298,17 +219,21 @@ public void onNext(T o) {
            try {
                filtered = filter(o);
            } catch (Throwable ex) {
-                StreamMessageUtil.closeOrAbort(o);


We don't need to call StreamMessageUtil.closeOrAbort(filtered);?

Isn't filtered null when we reach here?

For context, I found a case where o was double-released when a maximum length was exceeded. Calling StreamMessageUtil.closeOrAbort(o) didn't make sense since the ownership has been transferred to filter() method.

That is correct. 👍 The previous logic was wrong.

jrhee17

👍 👍 👍

ikhoon added the defect label Aug 5, 2024

ikhoon added this to the 1.30.0 milestone Aug 5, 2024

ikhoon requested review from jrhee17, minwoox and trustin as code owners August 5, 2024 11:33

ikhoon mentioned this pull request Aug 5, 2024

Update dependencies #5857

Merged

fix a broken test

86ea2cc

ikhoon marked this pull request as draft August 6, 2024 04:20

Handle corner cases

6921034

jrhee17 reviewed Aug 7, 2024

View reviewed changes

Fix flakyness

5d78935

ikhoon marked this pull request as ready for review August 7, 2024 04:45

Merge branch 'main' into fix-leak-in-HttpEncodedResponse

6fae642

jrhee17 reviewed Aug 8, 2024

View reviewed changes

minwoox approved these changes Aug 8, 2024

View reviewed changes

jrhee17 approved these changes Aug 8, 2024

View reviewed changes

ikhoon merged commit c1d5475 into line:main Aug 8, 2024
15 checks passed

ikhoon deleted the fix-leak-in-HttpEncodedResponse branch August 8, 2024 06:18

ikhoon mentioned this pull request Aug 8, 2024

Fix blocking annotation not working on gRPC calls properly #5399

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix a leak in `HttpEncodedResponse` #5858

Fix a leak in `HttpEncodedResponse` #5858

ikhoon commented Aug 5, 2024

github-actions bot commented Aug 5, 2024 •

edited

Loading

jrhee17 Aug 7, 2024

ikhoon Aug 7, 2024

ikhoon Aug 7, 2024

jrhee17 Aug 8, 2024

jrhee17 Aug 7, 2024

ikhoon Aug 7, 2024

jrhee17 Aug 8, 2024

ikhoon Aug 8, 2024

jrhee17 left a comment

jrhee17 Aug 8, 2024

jrhee17 Aug 8, 2024

jrhee17 Aug 8, 2024

ikhoon Aug 8, 2024

jrhee17 Aug 8, 2024 •

edited

Loading

ikhoon Aug 8, 2024 •

edited

Loading

ikhoon Aug 8, 2024

jrhee17 Aug 8, 2024

ikhoon Aug 8, 2024

minwoox left a comment

minwoox Aug 8, 2024

ikhoon Aug 8, 2024

ikhoon Aug 8, 2024

minwoox Aug 8, 2024

jrhee17 left a comment

	protected HttpData filter(HttpData obj) {
	count++;
	if (count < 3) {
	return obj;
	} else {
	subscription.cancel();

Fix a leak in HttpEncodedResponse #5858

Fix a leak in HttpEncodedResponse #5858

Conversation

ikhoon commented Aug 5, 2024

github-actions bot commented Aug 5, 2024 • edited Loading

🔍 Build Scan® (commit: 6fae642)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jrhee17 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jrhee17 Aug 8, 2024 • edited Loading

Choose a reason for hiding this comment

ikhoon Aug 8, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

minwoox left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jrhee17 left a comment

Choose a reason for hiding this comment

Fix a leak in `HttpEncodedResponse` #5858

Fix a leak in `HttpEncodedResponse` #5858

github-actions bot commented Aug 5, 2024 •

edited

Loading

🔍 Build Scan® (commit: `6fae642`)

jrhee17 Aug 8, 2024 •

edited

Loading

ikhoon Aug 8, 2024 •

edited

Loading