Skip to content

Commit 7ddc19d

Browse files
jlouvelclaude
andcommitted
fix: harden the GraphQL SSE subscription lifecycle
On the server, release the subscription slot when the first write fails, answer document errors of event stream requests with 400 and keep adapters without subscription transports restartable. On the client, signal non-positive demand without sending or waiting for a request, unregister subscriptions cancelled from onSubscribe() and abort requests cancelled before their exchange started. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1 parent 7b6b8c7 commit 7ddc19d

5 files changed

Lines changed: 228 additions & 20 deletions

File tree

‎org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -441,16 +441,17 @@ void unregister(GraphqlSsePublisher.EventStreamSubscription subscription) {
441441
}
442442

443443
/**
444-
* Creates the HTTP request of an operation sent for an event stream.
444+
* Fills the HTTP request of an operation sent for an event stream.
445445
*
446+
* @param request The new request to fill.
446447
* @param document The document.
447448
* @param operationName The name of the operation to execute, or null.
448449
* @param variables The variables, or null.
449450
* @return The request, customized.
450451
*/
451452
Request createEventStreamRequest(
452-
String document, String operationName, Map<String, Object> variables) {
453-
Request request = createRequest(document, operationName, variables);
453+
Request request, String document, String operationName, Map<String, Object> variables) {
454+
createRequest(request, document, operationName, variables);
454455
request.getClientInfo()
455456
.getAcceptedMediaTypes()
456457
.add(new Preference<>(MediaType.TEXT_EVENT_STREAM, 1F));
@@ -491,6 +492,11 @@ private static ExecutionResult failure(List<GraphQLError> errors) {
491492
* @return The request, customized.
492493
*/
493494
Request createRequest(String document, String operationName, Map<String, Object> variables) {
495+
return createRequest(new Request(), document, operationName, variables);
496+
}
497+
498+
private Request createRequest(
499+
Request request, String document, String operationName, Map<String, Object> variables) {
494500
Map<String, Object> body = new LinkedHashMap<>();
495501
body.put("query", document);
496502
if (operationName != null) {
@@ -500,7 +506,8 @@ Request createRequest(String document, String operationName, Map<String, Object>
500506
body.put("variables", variables);
501507
}
502508

503-
Request request = new Request(Method.POST, endpoint);
509+
request.setMethod(Method.POST);
510+
request.setResourceRef(endpoint);
504511
request.setEntity(
505512
new StringRepresentation(
506513
GraphqlJson.writeString(body),

‎org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -353,8 +353,10 @@ public synchronized void start() throws Exception {
353353
if (parserOptions == null) {
354354
parserOptions = ParserOptions.getDefaultOperationParserOptions();
355355
}
356-
subscriptionTransports =
357-
Collections.unmodifiableSet(EnumSet.copyOf(subscriptionTransports));
356+
// A typed copy, as EnumSet.copyOf() fails on an empty set once it is wrapped.
357+
EnumSet<GraphqlTransport> transports = EnumSet.noneOf(GraphqlTransport.class);
358+
transports.addAll(subscriptionTransports);
359+
subscriptionTransports = Collections.unmodifiableSet(transports);
358360
super.start();
359361
}
360362
}
@@ -437,10 +439,10 @@ public void handle(Request request, Response response) {
437439
operation =
438440
selectOperation(executionInput.getQuery(), executionInput.getOperationName());
439441
} catch (InvalidSyntaxException e) {
440-
write(response, responseType, failure(e.toInvalidSyntaxError()));
442+
write(response, responseType, failure(e.toInvalidSyntaxError()), sse);
441443
return;
442444
} catch (RequestException e) {
443-
write(response, responseType, failure(error(e.getMessage())));
445+
write(response, responseType, failure(error(e.getMessage())), sse);
444446
return;
445447
}
446448

‎org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSsePublisher.java‎

Lines changed: 57 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import java.io.InputStream;
1515
import java.util.Map;
1616
import java.util.Objects;
17+
import java.util.function.BooleanSupplier;
1718
import org.reactivestreams.Publisher;
1819
import org.reactivestreams.Subscriber;
1920
import org.reactivestreams.Subscription;
@@ -58,6 +59,9 @@ public void subscribe(Subscriber<? super ExecutionResult> subscriber) {
5859
subscriber.onSubscribe(subscription);
5960
if (!adapter.register(subscription)) {
6061
subscription.stop();
62+
} else if (subscription.isDone()) {
63+
// Cancelled or finished from onSubscribe(), before being registered.
64+
adapter.unregister(subscription);
6165
}
6266
}
6367

@@ -70,10 +74,12 @@ final class EventStreamSubscription implements Subscription {
7074

7175
private boolean started;
7276

73-
private boolean invalidDemand;
77+
private volatile boolean invalidDemand;
7478

7579
private volatile boolean cancelled;
7680

81+
private volatile boolean finished;
82+
7783
private volatile boolean stopped;
7884

7985
private volatile Request request;
@@ -84,22 +90,43 @@ final class EventStreamSubscription implements Subscription {
8490

8591
@Override
8692
public void request(long n) {
93+
boolean signal = false;
94+
boolean start = false;
8795
synchronized (this) {
8896
if (cancelled) {
8997
return;
9098
}
9199
if (n <= 0) {
92100
invalidDemand = true;
101+
// Without a reader thread, signal from here and prevent one from starting.
102+
signal = !started;
103+
if (signal) {
104+
cancelled = true;
105+
}
93106
} else {
94107
demand = demand + n < 0 ? Long.MAX_VALUE : demand + n;
108+
start = !started;
95109
}
110+
started = true;
96111
notifyAll();
97-
if (started) {
98-
return;
112+
}
113+
if (signal) {
114+
subscriber.onError(invalidDemandError());
115+
adapter.unregister(this);
116+
} else if (start) {
117+
Thread.ofVirtual().name("restlet-graphql-sse").start(this::run);
118+
} else if (n <= 0) {
119+
// Unblocks a pending read, then the reader thread signals the error.
120+
Request current = request;
121+
if (current != null) {
122+
current.abort();
99123
}
100-
started = true;
101124
}
102-
Thread.ofVirtual().name("restlet-graphql-sse").start(this::run);
125+
}
126+
127+
/** Indicates if the subscription is cancelled or its reader thread is done. */
128+
boolean isDone() {
129+
return cancelled || finished;
103130
}
104131

105132
@Override
@@ -136,6 +163,10 @@ void stop() {
136163
cancel();
137164
}
138165

166+
private IllegalArgumentException invalidDemandError() {
167+
return new IllegalArgumentException("The requested number of results must be positive");
168+
}
169+
139170
private GraphqlTransportException stoppedError() {
140171
return new GraphqlTransportException(
141172
"The GraphQL client was stopped", null, null, null);
@@ -145,7 +176,8 @@ private void run() {
145176
Response response = null;
146177
try {
147178
Request current =
148-
adapter.createEventStreamRequest(document, operationName, variables);
179+
adapter.createEventStreamRequest(
180+
new EventStreamRequest(), document, operationName, variables);
149181
request = current;
150182
if (cancelled) {
151183
terminateIfStopped();
@@ -184,6 +216,7 @@ private void run() {
184216
if (response != null && response.getEntity() != null) {
185217
response.getEntity().release();
186218
}
219+
finished = true;
187220
adapter.unregister(this);
188221
}
189222
}
@@ -243,9 +276,7 @@ private boolean awaitDemand() {
243276
}
244277
if (invalidDemand) {
245278
cancelled = true;
246-
subscriber.onError(
247-
new IllegalArgumentException(
248-
"The requested number of results must be positive"));
279+
subscriber.onError(invalidDemandError());
249280
return false;
250281
}
251282
if (demand != Long.MAX_VALUE) {
@@ -266,7 +297,23 @@ private void fail(Throwable error) {
266297
subscriber.onError(stoppedError());
267298
} else if (!cancelled) {
268299
cancelled = true;
269-
subscriber.onError(error);
300+
// A read aborted by invalid demand reports that demand.
301+
subscriber.onError(invalidDemand ? invalidDemandError() : error);
302+
}
303+
}
304+
305+
/**
306+
* A request aborted as soon as its connector can abort it, when cancelled before the
307+
* exchange started.
308+
*/
309+
private final class EventStreamRequest extends Request {
310+
311+
@Override
312+
public void setAbortAction(BooleanSupplier action) {
313+
super.setAbortAction(action);
314+
if (action != null && (cancelled || invalidDemand)) {
315+
action.getAsBoolean();
316+
}
270317
}
271318
}
272319
}

‎org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlEventStream.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -159,12 +159,12 @@ public void write(EventSink sink) throws IOException {
159159
sink.send(COMPLETE);
160160
return;
161161
}
162-
// An empty comment sends the response head at once, before the first result.
163-
sink.send(OPEN);
164162
QueueingSubscriber subscriber = new QueueingSubscriber(maxQueuedBytes);
165163
this.subscriber = subscriber;
166164
openStreams.add(this);
167165
try {
166+
// An empty comment sends the response head at once, before the first result.
167+
sink.send(OPEN);
168168
// Registered before checking, so that a concurrent stop of the adapter sees it.
169169
if (stopped || endpointStopped.getAsBoolean()) {
170170
subscriber.fail(new Failure("The GraphQL endpoint is stopping", null));
Lines changed: 152 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,152 @@
1+
/**
2+
* Copyright 2005-2026 Qlik
3+
*<p>
4+
* The content of this file is subject to the terms of the Apache 2.0 open
5+
* source license available at https://www.opensource.org/licenses/apache-2.0
6+
*<p>
7+
* Restlet is a registered trademark of QlikTech International AB.
8+
*/
9+
package org.restlet.ext.graphql;
10+
11+
import static org.junit.jupiter.api.Assertions.assertEquals;
12+
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
13+
import static org.junit.jupiter.api.Assertions.assertTrue;
14+
15+
import graphql.ExecutionResult;
16+
import java.util.Map;
17+
import java.util.concurrent.CountDownLatch;
18+
import java.util.concurrent.TimeUnit;
19+
import java.util.concurrent.atomic.AtomicInteger;
20+
import java.util.concurrent.atomic.AtomicReference;
21+
import org.junit.jupiter.api.AfterEach;
22+
import org.junit.jupiter.api.BeforeEach;
23+
import org.junit.jupiter.api.Test;
24+
import org.reactivestreams.Subscriber;
25+
import org.reactivestreams.Subscription;
26+
import org.restlet.Client;
27+
import org.restlet.Component;
28+
import org.restlet.Request;
29+
import org.restlet.Response;
30+
import org.restlet.Restlet;
31+
import org.restlet.data.MediaType;
32+
import org.restlet.data.Method;
33+
import org.restlet.data.Preference;
34+
import org.restlet.data.Protocol;
35+
import org.restlet.data.Status;
36+
import org.restlet.ext.graphql.internal.GraphqlJson;
37+
import org.restlet.representation.StringRepresentation;
38+
39+
/** Edge cases of the {@code graphql-sse} binding: request errors, restarts and demand. */
40+
class GraphqlSseEdgeCasesTestCase {
41+
42+
private final AtomicInteger requests = new AtomicInteger();
43+
44+
private Component component;
45+
46+
private Client client;
47+
48+
private String base;
49+
50+
@BeforeEach
51+
void setUp() throws Exception {
52+
component = new Component();
53+
component.getServers().add(Protocol.HTTP, 0);
54+
GraphqlServerAdapter sse = new GraphqlServerAdapter();
55+
sse.setGraphQL(GraphqlTestSchemas.graphQL(new AtomicInteger()));
56+
sse.getSubscriptionTransports().add(GraphqlTransport.SSE);
57+
component.getDefaultHost().attach("/graphql", sse);
58+
component
59+
.getDefaultHost()
60+
.attach(
61+
"/counted",
62+
new Restlet() {
63+
@Override
64+
public void handle(Request request, Response response) {
65+
requests.incrementAndGet();
66+
response.setStatus(Status.CLIENT_ERROR_NOT_FOUND);
67+
}
68+
});
69+
component.start();
70+
base = "http://localhost:" + component.getServers().get(0).getActualPort();
71+
client = new Client(Protocol.HTTP);
72+
client.start();
73+
}
74+
75+
@AfterEach
76+
void tearDown() throws Exception {
77+
client.stop();
78+
component.stop();
79+
}
80+
81+
private Response stream(Map<String, Object> body) {
82+
Request request = new Request(Method.POST, base + "/graphql");
83+
request.setEntity(
84+
new StringRepresentation(
85+
GraphqlJson.writeString(body), MediaType.APPLICATION_JSON));
86+
request.getClientInfo()
87+
.getAcceptedMediaTypes()
88+
.add(new Preference<>(MediaType.TEXT_EVENT_STREAM));
89+
return client.handle(request);
90+
}
91+
92+
@Test
93+
void documentErrorsAreBadRequests() {
94+
assertEquals(
95+
Status.CLIENT_ERROR_BAD_REQUEST,
96+
stream(Map.of("query", "subscription {")).getStatus());
97+
assertEquals(
98+
Status.CLIENT_ERROR_BAD_REQUEST,
99+
stream(Map.of("query", "{ hello }", "operationName", "Unknown")).getStatus());
100+
assertEquals(
101+
Status.CLIENT_ERROR_BAD_REQUEST,
102+
stream(Map.of("query", "query A { hello } query B { hello }")).getStatus());
103+
}
104+
105+
@Test
106+
void adapterWithoutTransportsRestarts() throws Exception {
107+
GraphqlServerAdapter adapter = new GraphqlServerAdapter();
108+
adapter.setGraphQL(GraphqlTestSchemas.graphQL(new AtomicInteger()));
109+
adapter.start();
110+
adapter.stop();
111+
adapter.start();
112+
assertTrue(adapter.isStarted());
113+
assertTrue(adapter.getSubscriptionTransports().isEmpty());
114+
adapter.stop();
115+
}
116+
117+
@Test
118+
void nonPositiveDemandFailsWithoutRequest() throws Exception {
119+
GraphqlClientAdapter graphql = new GraphqlClientAdapter();
120+
graphql.setClient(client);
121+
graphql.setEndpoint(base + "/counted");
122+
graphql.start();
123+
CountDownLatch done = new CountDownLatch(1);
124+
AtomicReference<Throwable> error = new AtomicReference<>();
125+
graphql.subscribe("subscription { ticks(count: 1) }", null, null)
126+
.subscribe(
127+
new Subscriber<ExecutionResult>() {
128+
@Override
129+
public void onSubscribe(Subscription subscription) {
130+
subscription.request(0);
131+
}
132+
133+
@Override
134+
public void onNext(ExecutionResult item) {}
135+
136+
@Override
137+
public void onError(Throwable throwable) {
138+
error.set(throwable);
139+
done.countDown();
140+
}
141+
142+
@Override
143+
public void onComplete() {
144+
done.countDown();
145+
}
146+
});
147+
assertTrue(done.await(5, TimeUnit.SECONDS));
148+
assertInstanceOf(IllegalArgumentException.class, error.get());
149+
assertEquals(0, requests.get());
150+
graphql.stop();
151+
}
152+
}

0 commit comments

Comments
 (0)