Request content preview isn't produced if decorate a service with ThrottlingService

[0x1306e6d]

I found that RequestLog#requestContentPreview returns null when I decorate a service using ContentPreviewingService and ThrottlingService.

The request content preview is not produced when aggregate a service which is decorated with following structure:

.decorator(((delegate, ctx, req) -> HttpResponse.from(
     completedFuture(null).handleAsync((ignored, cause) -> {
         try {
             return delegate.serve(ctx, req);
         } catch (Exception e) {
             return Exceptions.throwUnsafely(e);
         }
     }, ctx.eventLoop()))))

Following test can reproduce:

class RequestContentPreviewTest {

    private static final AtomicReference<ServiceRequestContext> ctxCaptor = new AtomicReference<>();

    @RegisterExtension
    static ServerExtension server = new ServerExtension() {
        @Override
        protected void configure(ServerBuilder builder) {
            builder.service("/", (ctx, req) -> {
                       ctxCaptor.set(ctx);
                       return HttpResponse.from(req.aggregate().thenApply(
                               aggregated -> HttpResponse.of("Hello, " + aggregated.contentUtf8() + "!")));
                   })
                   .decorator(LoggingService.builder()
                                            .requestLogLevel(LogLevel.INFO)
                                            .successfulResponseLogLevel(LogLevel.INFO)
                                            .failureResponseLogLevel(LogLevel.WARN)
                                            .newDecorator())
                   .decorator(ContentPreviewingService.newDecorator(Integer.MAX_VALUE))
                   .decorator(((delegate, ctx, req) -> HttpResponse.from(
                           completedFuture(null).handleAsync((ignored, cause) -> {
                               try {
                                   return delegate.serve(ctx, req);
                               } catch (Exception e) {
                                   return Exceptions.throwUnsafely(e);
                               }
                           }, ctx.eventLoop()))));
        }
    };

    @Test
    void requestContentPreviewH1C() {
        final WebClient client = WebClient.of(SessionProtocol.H1C,
                                              Endpoint.of("127.0.0.1", server.httpPort()));
        final RequestHeaders headers = RequestHeaders.of(HttpMethod.POST, "/",
                                                         HttpHeaderNames.CONTENT_TYPE, "text/plain");
        final AggregatedHttpResponse res = client.execute(headers, HttpData.ofUtf8("Armeria"))
                                                 .aggregate()
                                                 .join();
        assertThat(res.contentUtf8()).isEqualTo("Hello, Armeria!");

        final ServiceRequestContext ctx = ctxCaptor.get();
        final RequestLog log = ctx.log().whenComplete().join();
        assertThat(log.responseContentPreview()).isEqualTo("Hello, Armeria!");
        // expected: "Armeria"
        // but was: null
        // org.opentest4j.AssertionFailedError:
        // expected: "Hello, World!"
        assertThat(log.requestContentPreview()).isEqualTo("Armeria");
    }

    @Test
    void requestContentPreviewHTTP() {
        final WebClient client = WebClient.of(SessionProtocol.HTTP,
                                              Endpoint.of("127.0.0.1", server.httpPort()));
        final RequestHeaders headers = RequestHeaders.of(HttpMethod.POST, "/",
                                                         HttpHeaderNames.CONTENT_TYPE, "text/plain");
        final AggregatedHttpResponse res = client.execute(headers, HttpData.ofUtf8("Armeria"))
                                                 .aggregate()
                                                 .join();
        assertThat(res.contentUtf8()).isEqualTo("Hello, Armeria!");

        final ServiceRequestContext ctx = ctxCaptor.get();
        final RequestLog log = ctx.log().whenComplete().join();
        assertThat(log.responseContentPreview()).isEqualTo("Hello, Armeria!");
        assertThat(log.requestContentPreview()).isEqualTo("Armeria");
    }
}

logging H1C:

02:08:33.724 [armeria-common-worker-nio-2-2] INFO  c.l.a.server.logging.LoggingService - [sreqId=64b5968f, chanId=ec5eedd2, raddr=127.0.0.1:54446, laddr=127.0.0.1:54445][h1c://al01540922.local/#POST] Request: {startTime=2021-09-08T17:08:33.696Z(1631120913696000), length=7B, duration=24472µs(24472956ns), scheme=none+h1c, name=POST, headers=[:method=POST, :path=/, :scheme=http, host=127.0.0.1:54445, content-type=text/plain, user-agent=armeria/1.11.1-SNAPSHOT, content-length=7]}
02:08:33.741 [armeria-common-worker-nio-2-2] INFO  c.l.a.server.logging.LoggingService - [sreqId=64b5968f, chanId=ec5eedd2, raddr=127.0.0.1:54446, laddr=127.0.0.1:54445][h1c://al01540922.local/#POST] Response: {startTime=2021-09-08T17:08:33.737Z(1631120913737000), length=15B, duration=3305µs(3305059ns), totalDuration=43624µs(43624769ns), headers=[:status=200, content-type=text/plain; charset=utf-8, content-length=15], contentPreview=Hello, Armeria!}

logging H2C:

02:08:33.884 [armeria-common-worker-nio-2-4] INFO  c.l.a.server.logging.LoggingService - [sreqId=d4e30197, chanId=1d1d7144, raddr=127.0.0.1:54447, laddr=127.0.0.1:54445][h2c://al01540922.local/#POST] Request: {startTime=2021-09-08T17:08:33.882Z(1631120913882000), length=7B, duration=1976µs(1976048ns), scheme=none+h2c, name=POST, headers=[:method=POST, :path=/, :scheme=http, :authority=127.0.0.1:54445, content-type=text/plain, user-agent=armeria/1.11.1-SNAPSHOT, content-length=7], contentPreview=Armeria}
02:08:33.886 [armeria-common-worker-nio-2-4] INFO  c.l.a.server.logging.LoggingService - [sreqId=d4e30197, chanId=1d1d7144, raddr=127.0.0.1:54447, laddr=127.0.0.1:54445][h2c://al01540922.local/#POST] Response: {startTime=2021-09-08T17:08:33.884Z(1631120913884000), length=15B, duration=1648µs(1648011ns), totalDuration=3309µs(3309007ns), headers=[:status=200, content-type=text/plain; charset=utf-8, content-length=15], contentPreview=Hello, Armeria!}

It works well if not decorated by the last one.

What I found is that

armeria/core/src/main/java/com/linecorp/armeria/internal/logging/ContentPreviewingUtil.java

Line 62 in dbc7d4d

logBuilder.requestContentPreview(null);

is called before

armeria/core/src/main/java/com/linecorp/armeria/internal/logging/ContentPreviewingUtil.java

Line 82 in dbc7d4d

logBuilder.requestContentPreview(produced);

And this is because a CloseEvent from Http1RequestDecoder is already in queue before collect.

[ikhoon]

It seems like the req.whenComplete() was completed before filteredHttpRequest.whenComplete().
The following callback is used to detect the abortion of the upstream.

armeria/core/src/main/java/com/linecorp/armeria/internal/logging/ContentPreviewingUtil.java

Line 59 in 5b384fb

req.whenComplete().handle((unused, unused1) -> {

I think we can safely change req.whenComplete().handle() to req.whenComplete().exceptionally() so that the logBuilder.requestContentPreview(null) is set only when the request is completed exceptionally.
Otherwise, FilteredHttpRequest will set logBuilder.requestContentPreview().

Are you sending a PR? 😆

[ikhoon]

On second thought, it might be better to fix FilterStreamMessage to be completed when the upstream is exceptionally completed.

public abstract class FilteredStreamMessage<T, U> implements StreamMessage<U> {
    // Set true when being collected or subscribed
    private volatile boolean subscribed;

    protected FilteredStreamMessage(StreamMessage<T> upstream, boolean withPooledObjects) {
        ...
        upstream.whenComplete().exceptionally(cause -> {
            if (!subscribed) {
                // complete `completionFuture` with the cause
            }
        }
    }

, and could remove req.whenComplete().handle() block in ContentPreviewingUtil.

[0x1306e6d]

Thanks @ikhoon!

Second idea looks good. Let me prepare PR.

I have a question. I found that when a HttpResponse is deferred, handleCloseEvent can be called before delegated service is served in HTTP/1. Maybe is it ok?

[ikhoon]

I found that when a HttpResponse is deferred, handleCloseEvent can be called before delegated service is served in HTTP/1. Maybe is it ok?

When I locally tested it, handleCloseEvent seems to be called by req.aggregate() in service.
It should be a bug if handleCloseEvent is called before subscription.

[0x1306e6d]

I think I was confused. Please ignore 😅

[0x1306e6d]

@ikhoon, could you check following test?

@Test
void shouldCompleteWhenUpstreamCompletedExceptionally() {
    final EventLoop eventLoop = eventLoopExtension.get();

    final ByteBuf buf = newPooledBuffer();
    final HttpData data = HttpData.wrap(buf).withEndOfStream();
    final DefaultStreamMessage<HttpData> stream = new DefaultStreamMessage<>();
    stream.abort(new RuntimeException());

    final AtomicBoolean completed = new AtomicBoolean();

    final FilteredStreamMessage<HttpData, HttpData> filtered =
            new FilteredStreamMessage<HttpData, HttpData>(stream) {
                @Override
                protected HttpData filter(HttpData obj) {
                    return obj;
                }
            };
    filtered.whenComplete().handle((ignored, cause) -> {
        completed.set(true);
        return null;
    });

    eventLoop.execute(() -> filtered.collect(eventLoop));

    await().untilAtomic(completed, Matchers.is(true));
}

it might be better to fix FilterStreamMessage to be completed when the upstream is exceptionally completed.

I expected that FilteredStreamMessage isn't completed when upstream completed exceptionally, but it is.

So I guess we can simply

remove req.whenComplete().handle() block in ContentPreviewingUtil.

[ikhoon]

I expected that FilteredStreamMessage isn't completed when upstream completed exceptionally, but it is.

Good analysis. If upstream is aborted, it is propagated to FilteredStreamMessage.
As like your suggestion, we can safely remove req.whenComplete().handle(). 👍