Non deterministic crash caused by race conditions when accessing `contents` in GraphQL Query/Subscription

[AngeloFilaseta]

I'm tagging @Tbaile because we found the error together. Feel free to add whatever I forget.
I'm opening this Issue so I can add information as we discover them.

Description

Using the contents field in a GraphQL Query or Subscription can lead to a runtime error. In the case of subscriptions, it also causes the entire channel to be cancelled.
We are not sure that contents is the only API that causes this issue. Additional details in a bit.

Minimal Reproduction

This simulation can cause the problem: dodgeball.zip

The following subscription should trigger the issue in about 20 seconds or so (it really depends):

subscription crashReproduction {
  environment {
    nodes {
      contents {
        size
      }
    }
  }
}

Resulting Error

The GraphQL response contains a non-nullable violation error:

{
  "errors": [
    {
      "message": "The field at path '/environment' was declared as a non null type, but the code involved in retrieving data has wrongly returned a null value. [...] The non-nullable type is 'EnvironmentSurrogate' within parent type 'Subscription'"
    }
  ]
}

Weirdest part now, the error is not logged in any way server side, I doubt this is caused by a logger misconfiguration.
However, by injecting custom logic into the Ktor server engine, we were able to capture this exception:

12:12:12.302 [eventLoopGroupProxy-3-3] ERROR it.unibo.alchemist.boundary.graphql.server.modules.CustomFunctionFetcher -- Exception caught while executing function contents with parameters {instance parameter of fun it.unibo.alchemist.boundary.graphql.schema.model.surrogates.NodeSurrogate<T>.contents(): it.unibo.alchemist.boundary.graphql.schema.util.MoleculeToConcentrationMap=NodeSurrogate(origin=1587 contains: [<hit, 23>, <launching>], id=1587)}
java.lang.reflect.InvocationTargetException: null
        at jdk.internal.reflect.GeneratedMethodAccessor44.invoke(Unknown Source)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:569)
        at kotlin.reflect.jvm.internal.calls.CallerImpl$Method.callMethod(CallerImpl.kt:97)
        at kotlin.reflect.jvm.internal.calls.CallerImpl$Method$Instance.call(CallerImpl.kt:113)
        at kotlin.reflect.jvm.internal.KCallableImpl.callDefaultMethod$kotlin_reflection(KCallableImpl.kt:207)
        at kotlin.reflect.jvm.internal.KCallableImpl.callBy(KCallableImpl.kt:112)
        at it.unibo.alchemist.boundary.graphql.server.modules.CustomFunctionFetcher.retryTillNoException(GraphQLModule.kt:126)
        at it.unibo.alchemist.boundary.graphql.server.modules.CustomFunctionFetcher.runBlockingFunction(GraphQLModule.kt:117)
        at com.expediagroup.graphql.generator.execution.FunctionDataFetcher.get(FunctionDataFetcher.kt:63)
        at graphql.execution.instrumentation.dataloader.FallbackDataLoaderDispatchStrategy.lambda$modifyDataFetcher$0(FallbackDataLoaderDispatchStrategy.java:25)
        at graphql.execution.ExecutionStrategy.invokeDataFetcher(ExecutionStrategy.java:533)
        at graphql.execution.ExecutionStrategy.fetchField(ExecutionStrategy.java:497)
        at graphql.execution.ExecutionStrategy.fetchField(ExecutionStrategy.java:438)
        at graphql.execution.ExecutionStrategy.resolveFieldWithInfo(ExecutionStrategy.java:397)
        at graphql.execution.ExecutionStrategy.getAsyncFieldValueInfo(ExecutionStrategy.java:335)
        at graphql.execution.ExecutionStrategy.executeObject(ExecutionStrategy.java:214)
        at graphql.execution.ExecutionStrategy.completeValueForObject(ExecutionStrategy.java:963)
        at graphql.execution.ExecutionStrategy.completeValue(ExecutionStrategy.java:701)
        at graphql.execution.ExecutionStrategy.completeValueForList(ExecutionStrategy.java:816)
        at graphql.execution.ExecutionStrategy.completeValueForList(ExecutionStrategy.java:769)
        at graphql.execution.ExecutionStrategy.completeValue(ExecutionStrategy.java:686)
        at graphql.execution.ExecutionStrategy.completeField(ExecutionStrategy.java:653)
        at graphql.execution.ExecutionStrategy.resolveFieldWithInfo(ExecutionStrategy.java:409)
        at graphql.execution.ExecutionStrategy.getAsyncFieldValueInfo(ExecutionStrategy.java:335)
        at graphql.execution.ExecutionStrategy.executeObject(ExecutionStrategy.java:214)
        at graphql.execution.ExecutionStrategy.completeValueForObject(ExecutionStrategy.java:963)
        at graphql.execution.ExecutionStrategy.completeValue(ExecutionStrategy.java:701)
        at graphql.execution.ExecutionStrategy.completeField(ExecutionStrategy.java:653)
        at graphql.execution.ExecutionStrategy.completeField(ExecutionStrategy.java:631)
        at com.expediagroup.graphql.generator.execution.FlowSubscriptionExecutionStrategy.executeSubscriptionEvent(FlowSubscriptionExecutionStrategy.kt:168)
        at com.expediagroup.graphql.generator.execution.FlowSubscriptionExecutionStrategy.access$executeSubscriptionEvent(FlowSubscriptionExecutionStrategy.kt:51)
        at com.expediagroup.graphql.generator.execution.FlowSubscriptionExecutionStrategy$execute$lambda$1$$inlined$map$1$2.emit(Emitters.kt:51)
        at kotlinx.coroutines.flow.SharedFlowImpl.collect$suspendImpl(SharedFlow.kt:397)
        at kotlinx.coroutines.flow.SharedFlowImpl$collect$1.invokeSuspend(SharedFlow.kt)
        at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
        at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:100)
        at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: java.util.ConcurrentModificationException: null
        at java.base/java.util.ArrayList$Itr.checkForComodification(ArrayList.java:1013)
        at java.base/java.util.ArrayList$Itr.next(ArrayList.java:967)
        at it.unibo.alchemist.model.sapere.nodes.LsaNode.getContents(LsaNode.java:87)
        at it.unibo.alchemist.boundary.graphql.schema.model.surrogates.NodeSurrogate.contents(NodeSurrogate.kt:48)
        ... 45 common frames omitted

• Because GraphQL expects non-null fields, this unhandled error causes GraphQL to propagate null up the field chain and ultimately fails the subscription.
• This behavior might affect other fields, such as any dynamically changing property (nodes itself?), but for now we did not execute in-depth tests.
• The chosen Incarnation may influence how frequently this occurs, though the issue is inherently not strictly tied to one incarnation.

What happens behind the scene

When we declare a GraphQLMonitor, a Ktor server is attached to the simualation.
Upon every stepDone call, the environment is mapped to a EnvironmentSurrogate, and emitted on the Hot Flow.
Clients can then collect values from the flow. I believe the error may happen at this point.

  override fun stepDone(environment: Environment<T, P>, reaction: Actionable<T>?, time: Time, step: Long) {
      internalFlow.tryEmit(environment.toGraphQLEnvironmentSurrogate())
  }

Probably we should use the simulation thread to execute this mapping. I'll try to do something asap.