Subscription results keep in upstream order #3574

bbakerman · 2024-04-21T09:21:02Z

This PR has a reproduction of the problem reported and also a series fo fixes

However after discussions with the Spring team, its not apparent that the old "stream what completes first" approach is wrong.

So I have added two modes - the older CompletionStageMappingPublisher which will emit objects as soon as they complete and a newer CompletionStageMappingOrderedPublisher that will emit elements in the original upstream Publisher order.

Both have bufferring (for elements that have not completed when they are mapped into graphql objects) but once keeps them in original presentation order while the old one does what it has always done.

I have split the inner Subscriber classes out into their own classes and hence can unit test them better

I have changed the tests considerable. I have tested more edge cases.

I have also made them finish up when onError is called and they cancel any futures in flight.

spring-projects/spring-graphql#949

bbakerman · 2024-04-21T09:59:16Z

Updated it to have a Mono property thats async

onSubscribe
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=0, name=name0, lastEpisode=episode-boundedElastic-1for0, isFavorite=true}}, dataPresent=true, extensions=null}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=1, name=name1, lastEpisode=episode-boundedElastic-1for1, isFavorite=false}}, dataPresent=true, extensions=null}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=2, name=name2, lastEpisode=episode-boundedElastic-1for2, isFavorite=true}}, dataPresent=true, extensions=null}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=3, name=name3, lastEpisode=episode-boundedElastic-1for3, isFavorite=false}}, dataPresent=true, extensions=null}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=4, name=name4, lastEpisode=episode-boundedElastic-1for4, isFavorite=true}}, dataPresent=true, extensions=null}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=5, name=name5, lastEpisode=episode-boundedElastic-1for5, isFavorite=false}}, dataPresent=true, extensions=null}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=6, name=name6, lastEpisode=episode-boundedElastic-2for6, isFavorite=true}}, dataPresent=true, extensions=null}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=7, name=name7, lastEpisode=episode-boundedElastic-2for7, isFavorite=false}}, dataPresent=true, extensions=null}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=8, name=name8, lastEpisode=episode-boundedElastic-2for8, isFavorite=true}}, dataPresent=true, extensions=null}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=9, name=name9, lastEpisode=episode-boundedElastic-1for9, isFavorite=false}}, dataPresent=true, extensions=null}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=10, name=name10, lastEpisode=episode-boundedElastic-1for10, isFavorite=true}}, dataPresent=true, extensions=null}
complete

madwind · 2024-04-21T11:55:47Z

when request one by one

    private CompletableFuture<Flux<Object>> mkFluxDF(DataFetchingEnvironment env) {
        // async deliver of the publisher with random snoozing between values
        return CompletableFuture.supplyAsync(() -> Flux.generate(() -> 0, (counter, sink) -> {
            sink.next(mkValue(counter));
            snooze(rand(10, 100));
            if (counter == 10) {
                sink.complete();
            }
            return counter + 1;
        }).doOnNext(System.out::println));
    }

onSubscribe
{id=0, counter=0}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=0, name=name0, lastEpisode=episode-boundedElastic-1for0, isFavorite=true}}, dataPresent=true, extensions=null}
{id=1, counter=1}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=1, name=name1, lastEpisode=episode-boundedElastic-2for1, isFavorite=false}}, dataPresent=true, extensions=null}
{id=2, counter=2}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=2, name=name2, lastEpisode=episode-boundedElastic-2for2, isFavorite=true}}, dataPresent=true, extensions=null}
{id=3, counter=3}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=3, name=name3, lastEpisode=episode-boundedElastic-2for3, isFavorite=false}}, dataPresent=true, extensions=null}
{id=4, counter=4}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=4, name=name4, lastEpisode=episode-boundedElastic-2for4, isFavorite=true}}, dataPresent=true, extensions=null}
{id=5, counter=5}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=5, name=name5, lastEpisode=episode-boundedElastic-1for5, isFavorite=false}}, dataPresent=true, extensions=null}
{id=6, counter=6}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=6, name=name6, lastEpisode=episode-boundedElastic-1for6, isFavorite=true}}, dataPresent=true, extensions=null}
{id=7, counter=7}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=7, name=name7, lastEpisode=episode-boundedElastic-1for7, isFavorite=false}}, dataPresent=true, extensions=null}
{id=8, counter=8}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=8, name=name8, lastEpisode=episode-boundedElastic-2for8, isFavorite=true}}, dataPresent=true, extensions=null}
{id=9, counter=9}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=9, name=name9, lastEpisode=episode-boundedElastic-2for9, isFavorite=false}}, dataPresent=true, extensions=null}
{id=10, counter=10}
	onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=10, name=name10, lastEpisode=episode-boundedElastic-2for10, isFavorite=true}}, dataPresent=true, extensions=null}
complete

when request 10

        @Override
        public void onSubscribe(Subscription subscription) {
            this.subscription = subscription;
            System.out.println("onSubscribe");
            subscription.request(10);
        }

        @Override
        public void onNext(Object o) {
            System.out.println("\tonNext : " + o);
//            subscription.request(1);
        }

{id=0, counter=0}
{id=1, counter=1}
{id=2, counter=2}
{id=3, counter=3}
{id=4, counter=4}
{id=5, counter=5}
{id=6, counter=6}
onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=0, name=name0, lastEpisode=episode-boundedElastic-1for0, isFavorite=true}}, dataPresent=true, extensions=null}
onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=2, name=name2, lastEpisode=episode-boundedElastic-3for2, isFavorite=true}}, dataPresent=true, extensions=null}
onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=3, name=name3, lastEpisode=episode-boundedElastic-4for3, isFavorite=false}}, dataPresent=true, extensions=null}
{id=7, counter=7}
{id=8, counter=8}
{id=9, counter=9}
onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=1, name=name1, lastEpisode=episode-boundedElastic-2for1, isFavorite=false}}, dataPresent=true, extensions=null}
onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=4, name=name4, lastEpisode=episode-boundedElastic-5for4, isFavorite=true}}, dataPresent=true, extensions=null}
onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=5, name=name5, lastEpisode=episode-boundedElastic-1for5, isFavorite=false}}, dataPresent=true, extensions=null}
onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=6, name=name6, lastEpisode=episode-boundedElastic-2for6, isFavorite=true}}, dataPresent=true, extensions=null}
onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=8, name=name8, lastEpisode=episode-boundedElastic-4for8, isFavorite=true}}, dataPresent=true, extensions=null}
onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=7, name=name7, lastEpisode=episode-boundedElastic-3for7, isFavorite=false}}, dataPresent=true, extensions=null}
onNext : ExecutionResultImpl{errors=[], data={searchVideo={id=9, name=name9, lastEpisode=episode-boundedElastic-5for9, isFavorite=false}}, dataPresent=true, extensions=null}

…aks on tests and how we handle things

…it out the different publishers and also took out the subscribers

…e tests and more code tweaks

… subscribers now inhrent from each other and outstanding futures are now cancelled if the Publisher fails

bbakerman · 2024-04-24T11:12:40Z

src/main/java/graphql/execution/SubscriptionExecutionStrategy.java

+     * source publisher.  But this can be changed to {@link Boolean#TRUE} to keep them in order.
+     */
+    public static final String KEEP_SUBSCRIPTION_EVENTS_ORDERED = "KEEP_SUBSCRIPTION_EVENTS_ORDERED";
+


By default this is off (old behavior) but it can be opted into

bbakerman · 2024-04-24T11:13:23Z

src/main/java/graphql/execution/reactive/CompletionStageMappingOrderedPublisher.java

+ * @param <U> the upstream type to be mapped to
+ */
+@Internal
+public class CompletionStageMappingOrderedPublisher<D, U> extends CompletionStageMappingPublisher<D, U> {


it extends CompletionStageMappingPublisher because they have a LOT in common - just the ordering on inflight objects is different

Note also this has been split out of the publisher class

bbakerman · 2024-04-24T11:14:19Z

src/main/java/graphql/execution/reactive/CompletionStageMappingPublisher.java

-    private final Publisher<U> upstreamPublisher;
-    private final Function<U, CompletionStage<D>> mapper;
+    protected final Publisher<U> upstreamPublisher;
+    protected final Function<U, CompletionStage<D>> mapper;


now designed to be inherited from its ordered peer

bbakerman · 2024-04-24T11:14:37Z

src/main/java/graphql/execution/reactive/CompletionStageMappingPublisher.java

@@ -52,126 +54,4 @@ public Publisher<U> getUpstreamPublisher() {
        return upstreamPublisher;
    }

-    @SuppressWarnings("ReactiveStreamsSubscriberImplementation")


This was moved to its own class up in the same package

bbakerman · 2024-04-24T11:15:11Z

src/main/java/graphql/execution/reactive/CompletionStageOrderedSubscriber.java

+ * @param <D> mapped downstream values
+ */
+@Internal
+public class CompletionStageOrderedSubscriber<U, D> extends CompletionStageSubscriber<U, D> implements Subscriber<U> {


It inherents from CompletionStageSubscriber because again they are very much the same

except for ordering of source events

…ked method name on review

bbakerman · 2024-04-24T11:17:09Z

src/main/java/graphql/execution/reactive/CompletionStageOrderedSubscriber.java

+        if (throwable instanceof CompletionException & throwable.getCause() != null) {
+            return throwable.getCause();
+        }
+        return throwable;


the onError will be the underling exception not the wrapper CompletionException that comes from CF usage

bbakerman · 2024-04-24T11:17:40Z

src/main/java/graphql/execution/reactive/CompletionStageSubscriber.java

+ * @param <D> mapped downstream values
+ */
+@Internal
+public class CompletionStageSubscriber<U, D> implements Subscriber<U> {


This was split out from the peer Publisher code as its own file. Its complicated enough to deserve that.

bbakerman · 2024-04-24T11:18:04Z

src/main/java/graphql/execution/reactive/CompletionStageSubscriber.java

+    protected final Queue<CompletionStage<?>> inFlightDataQ;
+    protected final LockKit.ReentrantLock lock = new LockKit.ReentrantLock();
+    protected final AtomicReference<Runnable> onCompleteRun;
+    protected final AtomicBoolean isTerminal;


Designed to be inheret from by CompletionStageOrderedSubscriber

bbakerman · 2024-04-24T11:19:50Z

src/test/groovy/graphql/execution/SubscriptionExecutionStrategyTest.groovy

+                text
+              }
+            }
+        """).graphQLContext([(SubscriptionExecutionStrategy.KEEP_SUBSCRIPTION_EVENTS_ORDERED): true]).build()


our hint to make them ordered. Here is an example

bbakerman · 2024-04-24T11:22:09Z

src/test/groovy/graphql/execution/reactive/CompletionStageMappingOrderedPublisherTest.groovy

+import java.util.concurrent.CompletionStage
+import java.util.function.Function
+
+class CompletionStageMappingOrderedPublisherTest extends Specification {


The tests for the two classes kinda repeat themselves - but I figuered that even though they inherent from each other - the tests should stress them the same and make no assumptions on how they are implemented

With even more work maybe the tests become common ones use where: but I didnt go that far - copy pasta rather

And - no I ended up refactoring so that the same tests are run for both implementations - one in order and one not in order

bbakerman · 2024-04-24T11:23:19Z

src/test/groovy/graphql/execution/reactive/CompletionStageOrderedSubscriberTest.groovy

+
+import static graphql.execution.reactive.CompletionStageSubscriberTest.mapperThatDoesNotComplete
+
+class CompletionStageOrderedSubscriberTest extends Specification {


by breaking out the Subscriber classes into their own class - we can unit test them better

bbakerman · 2024-04-24T11:24:11Z

...on/reactive/tck/CompletionStageMappingOrderedPublisherRandomCompleteTckVerificationTest.java

@@ -0,0 +1,71 @@
+package graphql.execution.reactive.tck;


I have made a series of new TestNG compliance tests that use the TCK to ensure they are valid publishers

bbakerman · 2024-04-24T11:25:14Z

src/test/java/reproductions/SubscriptionReproduction.java

+ * This reproduction is to see what's happening with Subscriptions and whether they keep their
+ * order when values are async.
+ */
+public class SubscriptionReproduction {


This is a main() reproduction class - not a test - I wanted to find out what was happening and I thought this was a good way to find out. Other code now tests that the fix is in place

…bined the tests so there is less repeated test code

bbakerman · 2024-04-25T06:01:41Z

build.gradle

@@ -117,6 +117,7 @@ dependencies {

    testImplementation 'org.reactivestreams:reactive-streams-tck:' + reactiveStreamsVersion
    testImplementation "io.reactivex.rxjava2:rxjava:2.2.21"
+    testImplementation "io.projectreactor:reactor-core:3.6.5"


Added so we can do Reactor specific tests if we choose

bbakerman · 2024-04-25T06:02:41Z

src/main/java/graphql/execution/reactive/CompletionStageMappingPublisher.java

@@ -40,9 +35,16 @@ public CompletionStageMappingPublisher(Publisher<U> upstreamPublisher, Function<

    @Override
    public void subscribe(Subscriber<? super D> downstreamSubscriber) {
-        upstreamPublisher.subscribe(new CompletionStageSubscriber(downstreamSubscriber));
+        assertNotNullWithNPE(downstreamSubscriber, () -> "Subscriber passed to subscribe must not be null");


spec says it MUST be non null

bbakerman · 2024-04-25T06:03:01Z

src/main/java/graphql/execution/reactive/CompletionStageMappingPublisher.java

+
+    @NotNull
+    protected Subscriber<? super U> createSubscriber(Subscriber<? super D> downstreamSubscriber) {
+        return new CompletionStageSubscriber<>(mapper, downstreamSubscriber);
    }


The ordering comes from the Subscriber impl

bbakerman · 2024-04-25T06:03:57Z

src/main/java/graphql/execution/reactive/CompletionStageMappingPublisher.java

-            onCompleteOrError(() -> {
-                onCompleteOrErrorRunCalled.set(true);
-                downstreamSubscriber.onError(t);
-            });


See in the old impl we waited until ALL inflights finished before we called onError - well now we dont.

We fail fast after discussing it with the Spring folks

bbakerman · 2024-04-25T06:05:45Z

src/main/java/graphql/execution/reactive/CompletionStageOrderedSubscriber.java

+                    CompletableFuture<?> cf = cs.toCompletableFuture();
+                    if (cf.isDone()) {
+                        // take it off the queue
+                        inFlightDataQ.poll();


in a lock so we can rely on it not changing here on the second mutative read

bbakerman · 2024-04-25T06:06:29Z

src/main/java/graphql/execution/reactive/CompletionStageSubscriber.java

+            downstreamSubscriber.onError(t);
+            cancelInFlightFutures();
+        }
+    }


now fails fast

bbakerman · 2024-04-25T06:07:11Z

src/main/java/graphql/execution/reactive/CompletionStageSubscriber.java

+            if (isTerminal.compareAndSet(false, true)) {
+                downstreamSubscriber.onComplete();
+            }
+        });


waits for all inflight CFs to complete

dondonz · 2024-06-10T08:42:38Z

src/main/java/graphql/execution/reactive/CompletionStageSubscriber.java

+    public CompletionStageSubscriber(Function<U, CompletionStage<D>> mapper, Subscriber<? super D> downstreamSubscriber) {
+        this.mapper = mapper;
+        this.downstreamSubscriber = downstreamSubscriber;
+        inFlightDataQ = new ArrayDeque<>();


Nitpick: to a non-native English speaker it's going to be confusing what the "Q" is (it being a homophone for "queue"). I can see it's not introduced in this PR, it's based off naming from previous code.

Could rename these methods/variables to "Queue" or mention in a comment that the "Q" means queue.

Reproduction showing async DFs keeping field order

5dd3583

spring-projects/spring-graphql#949

bbakerman mentioned this pull request Apr 21, 2024

@SchemaMapping may cause the order to be disrupted. How to solve it spring-projects/spring-graphql#949

Closed

Updated to have a Mono property

ed39788

bbakerman added 3 commits April 22, 2024 23:03

This has a fix to buffer published events to keep them in order

5520a2d

This has a fix to buffer published events to keep them in order - twe…

23c3689

…aks on tests and how we handle things

This has a fix to buffer published events to keep them in order - twe…

916232b

…aks on tests and how we handle things

bbakerman changed the title ~~Reproduction showing async DFs keeping field order~~ Subscription results keep in upstream order Apr 24, 2024

bbakerman added 3 commits April 24, 2024 11:57

This has a fix to buffer published events to keep them in order - spl…

f7098df

…it out the different publishers and also took out the subscribers

This has a fix to buffer published events to keep them in order - mor…

db46b1d

…e tests and more code tweaks

This has a fix to buffer published events to keep them in order - the…

92713a5

… subscribers now inhrent from each other and outstanding futures are now cancelled if the Publisher fails

bbakerman commented Apr 24, 2024

View reviewed changes

This has a fix to buffer published events to keep them in order -twea…

f15fbfb

…ked method name on review

bbakerman commented Apr 24, 2024

View reviewed changes

This has a fix to buffer published events to keep them in order - com…

5943ab0

…bined the tests so there is less repeated test code

bbakerman commented Apr 25, 2024

View reviewed changes

dondonz added this to the 2024 July: non-breaking changes milestone Apr 29, 2024

dondonz reviewed Jun 10, 2024

View reviewed changes

bbakerman added this pull request to the merge queue Jun 18, 2024

Merged via the queue into master with commit 285d2c0 Jun 18, 2024
1 check passed

rstoyanchev mentioned this pull request Jul 3, 2024

@BatchMapping is broken for subscriptions spring-projects/spring-graphql#1019

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Subscription results keep in upstream order #3574

Subscription results keep in upstream order #3574

bbakerman commented Apr 21, 2024 •

edited

Loading

bbakerman commented Apr 21, 2024

madwind commented Apr 21, 2024

bbakerman Apr 24, 2024

bbakerman Apr 24, 2024

bbakerman Apr 24, 2024

bbakerman Apr 24, 2024 •

edited

Loading

bbakerman Apr 24, 2024

bbakerman Apr 24, 2024

bbakerman Apr 25, 2024

bbakerman Apr 24, 2024

bbakerman Apr 24, 2024

bbakerman Apr 24, 2024

bbakerman Apr 24, 2024

bbakerman Apr 24, 2024

bbakerman Apr 25, 2024

bbakerman Apr 24, 2024

bbakerman Apr 24, 2024

bbakerman Apr 24, 2024

bbakerman Apr 25, 2024

bbakerman Apr 25, 2024

bbakerman Apr 25, 2024

bbakerman Apr 25, 2024

bbakerman Apr 25, 2024

bbakerman Apr 25, 2024

bbakerman Apr 25, 2024

dondonz Jun 10, 2024


		import static graphql.execution.reactive.CompletionStageSubscriberTest.mapperThatDoesNotComplete

		class CompletionStageOrderedSubscriberTest extends Specification {

Subscription results keep in upstream order #3574

Subscription results keep in upstream order #3574

Conversation

bbakerman commented Apr 21, 2024 • edited Loading

bbakerman commented Apr 21, 2024

madwind commented Apr 21, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbakerman Apr 24, 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

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

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

Choose a reason for hiding this comment

bbakerman commented Apr 21, 2024 •

edited

Loading

bbakerman Apr 24, 2024 •

edited

Loading