Add exception when tryWrite returns false

[pushrsp]

Motivation:

tryWrite in GraphqlWSSubProtocol.writeNext(...) can return false when WebSocketWriter is closed which makes not to cancel the publisher and cleanup the resource.

Modifications:

• When tryWrite returns false, it throws ClosedStreamException.
• If ClosedStreamException is occurred, the exception handler in ExecutionResultSubscriber.onNext(...) can lead to cancel the publisher through subscription.cancel().

Result:

• Closes Do not ignore the result of StreamMessage.tryWrite() in GraphqlWSSubProtocol #5542

[CLAassistant]

All committers have signed the CLA.

[codecov]

Codecov Report

Attention: Patch coverage is 20.00000% with 4 lines in your changes are missing coverage. Please review.

Project coverage is 74.09%. Comparing base (b1a1044) to head (12451c9).
Report is 82 commits behind head on main.

❗ Current head 12451c9 differs from pull request most recent head 73f4765

Please upload reports for the commit 73f4765 to get more accurate results.

Files	Patch %	Lines
...p/armeria/server/graphql/GraphqlWSSubProtocol.java	25.00%	3 Missing ⚠️
...eria/server/graphql/ExecutionResultSubscriber.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #5557      +/-   ##
============================================
- Coverage     74.10%   74.09%   -0.02%     
+ Complexity    21012    21007       -5     
============================================
  Files          1819     1819              
  Lines         77394    77399       +5     
  Branches       9889     9889              
============================================
- Hits          57353    57347       -6     
- Misses        15386    15396      +10     
- Partials       4655     4656       +1     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[github-actions]

🔍 Build Scan® (commit: 73f4765)

Job name	Status	Build Scan®
build-windows-latest-jdk-21	✅	https://ge.armeria.dev/s/3qhuhlcahtjg2
build-self-hosted-unsafe-jdk-8	✅	https://ge.armeria.dev/s/7tzxflt2mq5ra
build-self-hosted-unsafe-jdk-21-snapshot-blockhound	✅	https://ge.armeria.dev/s/sex5b7kgaysdw
build-self-hosted-unsafe-jdk-17-min-java-17-coverage	✅	https://ge.armeria.dev/s/gzbnmfueyzmmc
build-self-hosted-unsafe-jdk-17-min-java-11	✅	https://ge.armeria.dev/s/vg6kcdgmnrrci
build-self-hosted-unsafe-jdk-17-leak	❌ (failure)	https://ge.armeria.dev/s/sfribxizbtnm6
build-self-hosted-unsafe-jdk-11	✅	https://ge.armeria.dev/s/v36fsh3kchutq
build-macos-12-jdk-21	✅	https://ge.armeria.dev/s/z6qyvu36jikqe

[ikhoon]

Regardless of the exception type, it would be safe to cancel.

Suggested change

	} catch (JsonProcessingException | ClosedStreamException e) {
	} catch (Throwable e) {

[ikhoon]

.write() will raise ClosedStreamException if it is closed.

Suggested change

	if (!out.tryWrite(event)) {
	throw ClosedStreamException.get();
	out.write(event);

Should we also use write() in other places to cancel the publisher?
A considering point is that writeError() in completeWithError() may cause an error again because out was closed already. If writeNext() fails, we may skip writeError().

[pushrsp]

Should we also use write() in other places to cancel the publisher?

I think writeError should be considered to use write instead of tryWrite.

armeria/graphql/src/main/java/com/linecorp/armeria/server/graphql/GraphqlWSSubProtocol.java

Line 358 in 89c7cf4

private static void writeError(WebSocketWriter out, String operationId, List<GraphQLError> errors)

A considering point is that writeError() in completeWithError() may cause an error again because out was closed already. If writeNext() fails, we may skip writeError().

I think we can fix like this below

private static void writeError(WebSocketWriter out, String operationId, Throwable t) {
        ...
        try {
            final String event = serializeToJson(errorResponse);
            logger.trace("ERROR: {}", event);
            out.write(event);
        } catch (Throwable e) {
            if (out.isOpen()) {
                logger.warn("Error serializing error event", e);
                out.close(e);
            }
        }
    }

[ikhoon]

We may simply return in the begging of the method if t is an instance of ClosedStreamException

[trustin]

Would you mind adding a test case for the problem you're trying to solve?

[pushrsp]

I'm trying to solve the problem to clean up the resources and cancel the publisher when WebSocketWriter was closed for some reasons. But I'm not sure how to test this problem. To test it, I need to close the WebSocketWriter and check if the publisher has been canceled.

[trustin]

@minwoox Any suggestions on testing strategy?
@pushrsp Could you fix build errors meanwhile? (checkstyle violations)

[minwoox]

Several tryWrite are still used in this class. Could you also fix it?

[pushrsp]

btw, do i have to handle for ClosedStreamException in other place?

[minwoox]

What do you think of merging ExecutionResultSubscriber and GraphqlSubProtocol to make it more readable? They don't need to be separate in the first place.

armeria/graphql/src/main/java/com/linecorp/armeria/server/graphql/GraphqlWSSubProtocol.java

Line 239 in 73f4765

new ExecutionResultSubscriber(id, new GraphqlSubProtocol() {

[minwoox]

Any suggestions on testing strategy?

I think we have to

• Create the GraphqlService using builder
• Create a mock request and call WebSocketHandler.handle()
• Close the returned WebSocket
• See if the StreamMessage is cancelled or not.

final StreamWriter<Object> streaming = StreamMessage.streaming();
final GraphqlService graphqlService =
        GraphqlService.builder()
                      .enableWebSocket(true)
                      .runtimeWiring(c -> {
                          c.type("Subscription",
                                 typeWiring -> typeWiring.dataFetcher("hello", e -> streaming));
                      })
                      .build();
assert graphqlService instanceof WebSocketHandler;
final WebSocket out = ((WebSocketHandler) graphqlService).handle(ctx, in);