fix(incarnation-sapere): fix race condition in LsaNode causing ConcurrentModificationException in GraphQL subscriptions

alchemist-graphql/src/jvmTest/kotlin/it/unibo/alchemist/boundary/graphql/schema/model/LsaNodeGraphQLConcurrencyTest.kt

@@ -0,0 +1,110 @@
+/*
+ * Copyright (C) 2010-2025, Danilo Pianini and contributors
+ * listed, for each module, in the respective subproject's build.gradle.kts file.
+ *
+ * This file is part of Alchemist, and is distributed under the terms of the
+ * GNU General Public License, with a linking exception,
+ * as described in the file LICENSE in the Alchemist distribution's top directory.
+ */
+
+package it.unibo.alchemist.boundary.graphql.schema.model
+
+import it.unibo.alchemist.boundary.GraphQLTestEnvironments
+import it.unibo.alchemist.boundary.graphql.schema.model.surrogates.toGraphQLNodeSurrogate
+import it.unibo.alchemist.model.Position
+import it.unibo.alchemist.model.geometry.Vector
+import it.unibo.alchemist.model.sapere.nodes.LsaNode
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.Timeout
+
+/**
+ * Tests that verify the fix for race conditions in LsaNode when accessed via GraphQL.
+ * This reproduces the original ConcurrentModificationException that occurred when
+ * GraphQL subscriptions accessed node contents while simulation reactions modified the same nodes.
+ */
+class LsaNodeGraphQLConcurrencyTest<T, P> where T : Any, P : Position<P>, P : Vector<P> {
+
+    @Test
+    @Timeout(value = 2, unit = TimeUnit.MINUTES)
+    fun `GraphQL node contents access should not cause ConcurrentModificationException during simulation`() {
+        GraphQLTestEnvironments.loadTests<T, P> { environment ->
+            val nodes = environment.nodes.filterIsInstance<LsaNode>()
+            // Skip test if no SAPERE nodes are available
+            check(nodes.isNotEmpty())
+            val threadCount = 20
+            val operationsPerThread = 500
+            val latch = CountDownLatch(threadCount * 2) // Double for both reader and writer threads
+            val exceptions = AtomicInteger(0)
+            val successfulReads = AtomicInteger(0)
+            val successfulWrites = AtomicInteger(0)
+            // Start reader threads that simulate GraphQL subscription access
+            repeat(threadCount) {
+                Thread {
+                    try {
+                        repeat(operationsPerThread) {
+                            // This is what GraphQL does when accessing node contents
+                            val randomNode = nodes.random()
+                            val nodeSurrogate = randomNode.toGraphQLNodeSurrogate()
+                            // This calls LsaNode.getContents() which was causing ConcurrentModificationException
+                            val contents = nodeSurrogate.contents()
+                            // Verify we can access the size safely
+                            contents.size
+                            successfulReads.incrementAndGet()
+                        }
+                    } catch (e: ConcurrentModificationException) {
+                        exceptions.incrementAndGet()
+                        throw e
+                    } catch (e: Exception) {
+                        // Other exceptions are acceptable in this test context
+                        println("Reader thread caught non-CME exception: ${e::class.simpleName}")
+                    } finally {
+                        latch.countDown()
+                    }
+                }.start()
+            }
+            // Start writer threads that simulate simulation reactions modifying nodes
+            repeat(threadCount) {
+                Thread {
+                    try {
+                        repeat(operationsPerThread) {
+                            // Simulate reactions modifying node contents
+                            val randomNode = nodes.random()
+                            // This modifies the instances collection, potentially causing race conditions
+                            try {
+                                val firstMolecule = randomNode.lsaSpace.firstOrNull()
+                                if (firstMolecule != null) {
+                                    randomNode.setConcentration(firstMolecule)
+                                }
+                            } catch (e: Exception) {
+                                // Expected in some cases during concurrent access
+                                println("Writer operation caught exception: ${e::class.simpleName}")
+                            }
+                            successfulWrites.incrementAndGet()
+                        }
+                    } catch (e: Exception) {
+                        // Exceptions during writes are acceptable for this test
+                        println("Writer thread caught exception: ${e::class.simpleName}")
+                    } finally {
+                        latch.countDown()
+                    }
+                }.start()
+            }
+            // Wait for all threads to complete
+            assertTrue(latch.await(90, TimeUnit.SECONDS), "Test should complete within timeout")
+            // The key assertion: no ConcurrentModificationException should occur
+            assertEquals(0, exceptions.get(), "No ConcurrentModificationException should occur with the fix")
+            // Verify that operations actually happened
+            assertTrue(successfulReads.get() > 0, "Should have successful reads")
+            assertTrue(successfulWrites.get() > 0, "Should have successful writes")
+            println("Test completed successfully:")
+            println("- Successful reads: ${successfulReads.get()}")
+            println("- Successful writes: ${successfulWrites.get()}")
+            println("- ConcurrentModificationExceptions: ${exceptions.get()}")
+        }
+    }
+}

alchemist-incarnation-sapere/src/main/java/it/unibo/alchemist/model/sapere/nodes/LsaNode.java

@@ -50,7 +50,9 @@ public LsaNode(final Environment<List<ILsaMolecule>, ?> environment) {
     @Override
     public boolean contains(@Nonnull final Molecule molecule) {
         if (molecule instanceof final ILsaMolecule toMatch) {
-            return instances.stream().anyMatch(mol -> mol.matches(toMatch));
+            synchronized (instances) {
+                return instances.stream().anyMatch(mol -> mol.matches(toMatch));
+            }
         }
         return false;
     }
@@ -63,7 +65,9 @@ protected List<ILsaMolecule> createT() { // NOPMD: this must return null, not an
 
     @Override
     public int getMoleculeCount() {
-        return instances.size();
+        synchronized (instances) {
+            return instances.size();
+        }
     }
 
     @Override
@@ -72,9 +76,11 @@ public List<ILsaMolecule> getConcentration(@Nonnull final Molecule m) {
             throw new IllegalArgumentException(m + " is not a compatible molecule type");
         }
         final ArrayList<ILsaMolecule> listMol = new ArrayList<>();
-        for (final ILsaMolecule instance : instances) {
-            if (mol.matches(instance)) {
-                listMol.add(instance);
+        synchronized (instances) {
+            for (final ILsaMolecule instance : instances) {
+                if (mol.matches(instance)) {
+                    listMol.add(instance);
+                }
             }
         }
         return listMol;
@@ -83,8 +89,13 @@ public List<ILsaMolecule> getConcentration(@Nonnull final Molecule m) {
     @Override
     @Nonnull
     public Map<Molecule, List<ILsaMolecule>> getContents() {
-        final Map<Molecule, List<ILsaMolecule>> res = new HashMap<>(instances.size(), 1.0f);
-        for (final ILsaMolecule m : instances) {
+        // Create a defensive copy to avoid ConcurrentModificationException
+        final List<ILsaMolecule> instancesCopy;
+        synchronized (instances) {
+            instancesCopy = new ArrayList<>(instances);
+        }
+        final Map<Molecule, List<ILsaMolecule>> res = new HashMap<>(instancesCopy.size(), 1.0f);
+        for (final ILsaMolecule m : instancesCopy) {
             final List<ILsaMolecule> l;
             if (res.containsKey(m)) {
                 /*
@@ -105,15 +116,19 @@ public Map<Molecule, List<ILsaMolecule>> getContents() {
 
     @Override
     public List<ILsaMolecule> getLsaSpace() {
-        return Collections.unmodifiableList(instances);
+        synchronized (instances) {
+            return Collections.unmodifiableList(new ArrayList<>(instances));
+        }
     }
 
     @Override
     public boolean removeConcentration(final ILsaMolecule matchedInstance) {
-        for (int i = 0; i < instances.size(); i++) {
-            if (matchedInstance.matches(instances.get(i))) {
-                instances.remove(i);
-                return true;
+        synchronized (instances) {
+            for (int i = 0; i < instances.size(); i++) {
+                if (matchedInstance.matches(instances.get(i))) {
+                    instances.remove(i);
+                    return true;
+                }
             }
         }
         throw new IllegalStateException("Tried to remove missing " + matchedInstance + " from " + this);
@@ -122,7 +137,9 @@ public boolean removeConcentration(final ILsaMolecule matchedInstance) {
     @Override
     public void setConcentration(final ILsaMolecule inst) {
         if (inst.isIstance()) {
-            instances.add(inst);
+            synchronized (instances) {
+                instances.add(inst);
+            }
         } else {
             throw new IllegalStateException("Tried to insert uninstanced " + inst + " into " + this);
         }

alchemist-incarnation-sapere/src/test/java/it/unibo/alchemist/model/sapere/nodes/LsaNodeConcurrencyTest.java

@@ -0,0 +1,125 @@
+/*
+ * Copyright (C) 2010-2023, Danilo Pianini and contributors
+ * listed, for each module, in the respective subproject's build.gradle.kts file.
+ *
+ * This file is part of Alchemist, and is distributed under the terms of the
+ * GNU General Public License, with a linking exception,
+ * as described in the file LICENSE in the Alchemist distribution's top directory.
+ */
+
+package it.unibo.alchemist.model.sapere.nodes;
+
+import it.unibo.alchemist.model.Environment;
+import it.unibo.alchemist.model.Molecule;
+import it.unibo.alchemist.model.sapere.ILsaMolecule;
+import it.unibo.alchemist.model.sapere.SAPEREIncarnation;
+import it.unibo.alchemist.model.sapere.molecules.LsaMolecule;
+import it.unibo.alchemist.model.environments.Continuous2DEnvironment;
+import it.unibo.alchemist.model.positions.Euclidean2DPosition;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Test for concurrent access to LsaNode to ensure thread safety.
+ */
+class LsaNodeConcurrencyTest {
+
+    private static final int MIN_MOLECULES = 5;
+    private static final int TIMEOUT_SECONDS = 30;
+
+    @Test
+    void testConcurrentGetContentsAndModification() throws InterruptedException {
+        final SAPEREIncarnation<Euclidean2DPosition> incarnation = new SAPEREIncarnation<>();
+        final Environment<List<ILsaMolecule>, Euclidean2DPosition> environment = new Continuous2DEnvironment<>(incarnation);
+        final LsaNode node = new LsaNode(environment);
+        // Add some initial molecules
+        for (int i = 0; i < 10; i++) {
+            node.setConcentration(new LsaMolecule("molecule" + i));
+        }
+        final int numberOfThreads = 10;
+        final int numberOfOperations = 100;
+        final ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);
+        final CountDownLatch latch = new CountDownLatch(numberOfThreads);
+        final AtomicBoolean exceptionOccurred = new AtomicBoolean(false);
+        // Start threads that modify the node while others read from it
+        for (int i = 0; i < numberOfThreads; i++) {
+            final int threadId = i;
+            executor.submit(() -> {
+                try {
+                    for (int j = 0; j < numberOfOperations; j++) {
+                        if (threadId % 2 == 0) {
+                            // Reader threads - should not throw ConcurrentModificationException
+                            final Map<Molecule, List<ILsaMolecule>> contents = node.getContents();
+                            assertNotNull(contents);
+                            final int moleculeCount = node.getMoleculeCount();
+                            assertTrue(moleculeCount >= 0);
+                            final List<ILsaMolecule> lsaSpace = node.getLsaSpace();
+                            assertNotNull(lsaSpace);
+                        } else {
+                            // Writer threads
+                            final ILsaMolecule newMolecule = new LsaMolecule("thread" + threadId + "_" + j);
+                            node.setConcentration(newMolecule);
+                            // Sometimes remove molecules to simulate real concurrent modification
+                            if (j % 10 == 0 && node.getMoleculeCount() > MIN_MOLECULES) {
+                                try {
+                                    node.removeConcentration(newMolecule);
+                                } catch (final IllegalStateException e) {
+                                    // Expected if molecule was already removed by another thread
+                                }
+                            }
+                        }
+                    }
+                } catch (final IllegalStateException | java.util.ConcurrentModificationException e) {
+                    exceptionOccurred.set(true);
+                    e.printStackTrace();
+                } finally {
+                    latch.countDown();
+                }
+            });
+        }
+        // Wait for all threads to complete
+        assertTrue(latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS), "Test should complete within 30 seconds");
+        executor.shutdown();
+        // No exceptions should have occurred (especially no ConcurrentModificationException)
+        assertFalse(exceptionOccurred.get(), "No exceptions should occur during concurrent access");
+        // Verify the node is still in a valid state
+        final Map<Molecule, List<ILsaMolecule>> finalContents = node.getContents();
+        assertNotNull(finalContents);
+        assertTrue(node.getMoleculeCount() >= 0);
+    }
+
+    @Test
+    void testBasicFunctionalityPreserved() {
+        final SAPEREIncarnation<Euclidean2DPosition> incarnation = new SAPEREIncarnation<>();
+        final Environment<List<ILsaMolecule>, Euclidean2DPosition> environment = new Continuous2DEnvironment<>(incarnation);
+        final LsaNode node = new LsaNode(environment);
+        final ILsaMolecule testMolecule = new LsaMolecule("test");
+        // Test basic operations still work correctly
+        assertEquals(0, node.getMoleculeCount());
+        assertFalse(node.contains(testMolecule));
+        node.setConcentration(testMolecule);
+        assertEquals(1, node.getMoleculeCount());
+        assertTrue(node.contains(testMolecule));
+        final Map<Molecule, List<ILsaMolecule>> contents = node.getContents();
+        assertNotNull(contents);
+        assertEquals(1, contents.size());
+        final List<ILsaMolecule> lsaSpace = node.getLsaSpace();
+        assertNotNull(lsaSpace);
+        assertEquals(1, lsaSpace.size());
+        assertTrue(node.removeConcentration(testMolecule));
+        assertEquals(0, node.getMoleculeCount());
+        assertFalse(node.contains(testMolecule));
+    }
+}