feat: make shared signals use asynchronous signal trees (#23559)

* feat: make shared signals use asynchronous signal trees

Shared signals (SharedValueSignal, SharedListSignal, SharedMapSignal,
SharedNodeSignal) now use LocalAsynchronousSignalTree instead of
SynchronousSignalTree. This aligns with future clustered implementations
where confirmation happens asynchronously via an event log.

Key changes:
- Add LocalAsynchronousSignalTree that dispatches confirm() on
  Vaadin's service executor
- Adjust confirm() to skip observer notification when confirmed
  commands were at the head of the unconfirmed queue (no state change)

As a side effect, all signals in a transaction must be from the same tree.

Fixes #23545

Author: Artur-

flow-server/src/main/java/com/vaadin/flow/signals/shared/SharedListSignal.java

@@ -34,8 +34,8 @@
 import com.vaadin.flow.signals.function.TransactionTask;
 import com.vaadin.flow.signals.operations.InsertOperation;
 import com.vaadin.flow.signals.operations.SignalOperation;
+import com.vaadin.flow.signals.shared.impl.LocalAsynchronousSignalTree;
 import com.vaadin.flow.signals.shared.impl.SignalTree;
-import com.vaadin.flow.signals.shared.impl.SynchronousSignalTree;
 
 /**
  * A signal containing a list of values. Supports atomic updates to the list
@@ -158,7 +158,7 @@ private static Id idOf(@Nullable AbstractSignal<?> signal) {
      *            the element type, not <code>null</code>
      */
     public SharedListSignal(Class<T> elementType) {
-        this(new SynchronousSignalTree(false), Id.ZERO, ANYTHING_GOES,
+        this(new LocalAsynchronousSignalTree(), Id.ZERO, ANYTHING_GOES,
                 elementType);
     }
 

flow-server/src/main/java/com/vaadin/flow/signals/shared/SharedMapSignal.java

@@ -37,8 +37,8 @@
 import com.vaadin.flow.signals.operations.PutIfAbsentResult;
 import com.vaadin.flow.signals.operations.SignalOperation;
 import com.vaadin.flow.signals.shared.impl.CommandResult.NodeModification;
+import com.vaadin.flow.signals.shared.impl.LocalAsynchronousSignalTree;
 import com.vaadin.flow.signals.shared.impl.SignalTree;
-import com.vaadin.flow.signals.shared.impl.SynchronousSignalTree;
 
 /**
  * A signal containing a map of values with string keys. Supports atomic updates
@@ -62,7 +62,7 @@ public class SharedMapSignal<T>
      *            the element type, not <code>null</code>
      */
     public SharedMapSignal(Class<T> elementType) {
-        this(new SynchronousSignalTree(false), Id.ZERO, ANYTHING_GOES,
+        this(new LocalAsynchronousSignalTree(), Id.ZERO, ANYTHING_GOES,
                 elementType);
     }
 

flow-server/src/main/java/com/vaadin/flow/signals/shared/SharedNodeSignal.java

@@ -35,8 +35,8 @@
 import com.vaadin.flow.signals.operations.PutIfAbsentResult;
 import com.vaadin.flow.signals.operations.SignalOperation;
 import com.vaadin.flow.signals.shared.SharedListSignal.ListPosition;
+import com.vaadin.flow.signals.shared.impl.LocalAsynchronousSignalTree;
 import com.vaadin.flow.signals.shared.impl.SignalTree;
-import com.vaadin.flow.signals.shared.impl.SynchronousSignalTree;
 
 /**
  * A signal representing a node in a tree structure. The {@link #get()} of a
@@ -142,7 +142,7 @@ public Map<String, SharedNodeSignal> mapChildren() {
      * node structure. The signal does not support clustering.
      */
     public SharedNodeSignal() {
-        this(new SynchronousSignalTree(false), Id.ZERO, ANYTHING_GOES);
+        this(new LocalAsynchronousSignalTree(), Id.ZERO, ANYTHING_GOES);
     }
 
     /**

flow-server/src/main/java/com/vaadin/flow/signals/shared/SharedValueSignal.java

@@ -34,8 +34,8 @@
 import com.vaadin.flow.signals.impl.Transaction;
 import com.vaadin.flow.signals.operations.CancelableOperation;
 import com.vaadin.flow.signals.operations.SignalOperation;
+import com.vaadin.flow.signals.shared.impl.LocalAsynchronousSignalTree;
 import com.vaadin.flow.signals.shared.impl.SignalTree;
-import com.vaadin.flow.signals.shared.impl.SynchronousSignalTree;
 
 /**
  * A signal containing a value. The value is updated as a single atomic change.
@@ -59,7 +59,7 @@ public class SharedValueSignal<T> extends AbstractSignal<T> {
      */
     @SuppressWarnings("unchecked")
     public SharedValueSignal(T initialValue) {
-        this(new SynchronousSignalTree(false), Id.ZERO, ANYTHING_GOES,
+        this(new LocalAsynchronousSignalTree(), Id.ZERO, ANYTHING_GOES,
                 (Class<T>) initialValue.getClass());
         set(initialValue);
     }
@@ -72,7 +72,7 @@ public SharedValueSignal(T initialValue) {
      *            the value type, not <code>null</code>
      */
     public SharedValueSignal(Class<T> valueType) {
-        this(new SynchronousSignalTree(false), Id.ZERO, ANYTHING_GOES,
+        this(new LocalAsynchronousSignalTree(), Id.ZERO, ANYTHING_GOES,
                 Objects.requireNonNull(valueType));
     }
 

flow-server/src/main/java/com/vaadin/flow/signals/shared/impl/AsynchronousSignalTree.java

@@ -73,17 +73,24 @@ public void confirm(List<SignalCommand> commands) {
 
             confirmed = new Snapshot(builder);
 
+            /*
+             * Check if the confirmed commands were at the head of
+             * unconfirmedCommands. If so, the same commands were already
+             * applied during latency compensation with the same results, so
+             * submitted doesn't change and we can skip re-applying and
+             * notifying observers.
+             */
+            boolean confirmedFromHead = wereAtHead(
+                    unconfirmedCommands.getCommands(), commands);
+
             // Remove any pending commands that are now confirmed from the queue
             unconfirmedCommands.removeHandledCommands(results.keySet());
 
             Snapshot oldSubmitted = submitted;
 
-            /*
-             * TODO: could skip this part if the newly confirmed commands were
-             * at the head of unconfirmedCommands since submitted doesn't change
-             * in that case
-             */
-            if (!unconfirmedCommands.isEmpty()) {
+            if (confirmedFromHead) {
+                // submitted doesn't change so no need to rebuild or notify
+            } else if (!unconfirmedCommands.isEmpty()) {
                 // Re-apply pending commands that remain in the queue
                 builder.apply(unconfirmedCommands.getCommands());
 
@@ -92,14 +99,50 @@ public void confirm(List<SignalCommand> commands) {
                 submitted = confirmed;
             }
 
-            notifyObservers(oldSubmitted, submitted);
+            if (!confirmedFromHead) {
+                notifyObservers(oldSubmitted, submitted);
+            }
 
             unconfirmedCommands.notifyResultHandlers(results, commands);
 
             notifyProcessedCommandSubscribers(commands, results);
         });
     }
 
+    /**
+     * Checks whether all confirmed commands appear at the head of the
+     * unconfirmed command list. This means the confirmed commands were the
+     * first commands submitted to this tree and no reordering or external
+     * commands were involved. Uses the top-level command list to determine the
+     * count rather than the results key set, since results may also include
+     * sub-command IDs from transaction commands.
+     * <p>
+     * When this returns {@code true} and there are additional unconfirmed
+     * commands beyond the confirmed ones, submitted still doesn't need to be
+     * rebuilt. If unconfirmed was [A, B, C] and confirmed is [A], then the
+     * current submitted was built as {@code old_confirmed + A + B + C}. After
+     * confirming A, the new confirmed is {@code old_confirmed + A}, so
+     * rebuilding would give {@code (old_confirmed + A) + B + C} which is the
+     * same as the current submitted.
+     */
+    private static boolean wereAtHead(List<SignalCommand> unconfirmedCommands,
+            List<SignalCommand> confirmedCommands) {
+        int confirmedCount = confirmedCommands.size();
+        if (confirmedCount == 0) {
+            return true;
+        }
+        if (confirmedCount > unconfirmedCommands.size()) {
+            return false;
+        }
+        for (int i = 0; i < confirmedCount; i++) {
+            if (!unconfirmedCommands.get(i).commandId()
+                    .equals(confirmedCommands.get(i).commandId())) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     @Override
     public PendingCommit prepareCommit(CommandsAndHandlers changes) {
         assert hasLock();

flow-server/src/main/java/com/vaadin/flow/signals/shared/impl/LocalAsynchronousSignalTree.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2000-2026 Vaadin Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.vaadin.flow.signals.shared.impl;
+
+import java.util.List;
+
+import com.vaadin.flow.signals.SignalCommand;
+import com.vaadin.flow.signals.SignalEnvironment;
+
+/**
+ * An asynchronous signal tree for single-JVM use that dispatches
+ * {@link #confirm(List)} using the default effect dispatcher from
+ * {@link SignalEnvironment}. This makes the behavior consistent with clustered
+ * implementations where confirmation happens asynchronously.
+ */
+public class LocalAsynchronousSignalTree extends AsynchronousSignalTree {
+
+    @Override
+    protected void submit(List<SignalCommand> commands) {
+        SignalEnvironment.getDefaultEffectDispatcher()
+                .execute(() -> confirm(commands));
+    }
+}

flow-server/src/test/java/com/vaadin/flow/signals/impl/ComputedSignalTest.java

@@ -299,7 +299,7 @@ void transaction_readInCommittedTransaction_notCoumptedAgainAfterTransaction() {
     }
 
     @Test
-    void transaction_readInAbortedTransaction_notCoumptedAgainAfterTransaction() {
+    void transaction_readInAbortedTransaction_valueRestoredAfterRejection() {
         SharedValueSignal<String> source = new SharedValueSignal<>("value");
         AtomicInteger count = new AtomicInteger();
 
@@ -308,20 +308,26 @@ void transaction_readInAbortedTransaction_notCoumptedAgainAfterTransaction() {
             return source.get();
         });
 
-        signal.get();
+        assertEquals("value", signal.get());
         assertEquals(1, count.get());
 
         Transaction.runInTransaction(() -> {
             source.set("update");
 
-            signal.get();
+            assertEquals("update", signal.get());
             assertEquals(2, count.get());
 
             source.verifyValue("other");
         });
 
-        signal.get();
-        assertEquals(2, count.get());
+        /*
+         * Count is 3 because the computed signal's dependency was captured with
+         * the in-transaction value ("update"). After the rejected transaction,
+         * the submitted value is still "value", which differs from the captured
+         * value, so the computed signal must recompute.
+         */
+        assertEquals("value", signal.get());
+        assertEquals(3, count.get());
     }
 
     @Test

flow-server/src/test/java/com/vaadin/flow/signals/impl/EffectTest.java

@@ -194,24 +194,6 @@ void changeTracking_multipleWritesInTransaction_effectRunOnce() {
         assertEquals(List.of("", "second"), invocations);
     }
 
-    @Test
-    void changeTracking_multipleSignalsInTransaction_effectRunOnce() {
-        SharedValueSignal<String> signal1 = new SharedValueSignal<>("");
-        SharedValueSignal<String> signal2 = new SharedValueSignal<>("");
-        ArrayList<String> invocations = new ArrayList<>();
-
-        Signal.unboundEffect(() -> {
-            invocations.add(signal1.get() + signal2.get());
-        });
-
-        Signal.runInTransaction(() -> {
-            signal1.set("one ");
-            signal2.set("two");
-        });
-
-        assertEquals(List.of("", "one two"), invocations);
-    }
-
     @Test
     void changeTracking_changeOtherPartOfNode_effectNotRunAgain() {
         SharedValueSignal<String> signal = new SharedValueSignal<>("value");

flow-server/src/test/java/com/vaadin/flow/signals/shared/impl/LocalAsynchronousSignalTreeTest.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2000-2026 Vaadin Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.vaadin.flow.signals.shared.impl;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+
+import com.vaadin.flow.signals.SignalCommand;
+import com.vaadin.flow.signals.SignalTestBase;
+import com.vaadin.flow.signals.TestUtil;
+import com.vaadin.flow.signals.shared.impl.SignalTree.Type;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class LocalAsynchronousSignalTreeTest extends SignalTestBase {
+
+    @Test
+    void type_isAsynchronous() {
+        LocalAsynchronousSignalTree tree = new LocalAsynchronousSignalTree();
+        assertEquals(Type.ASYNCHRONOUS, tree.type());
+    }
+
+    @Test
+    void submit_singleCommand_immediatelyConfirmed() {
+        LocalAsynchronousSignalTree tree = new LocalAsynchronousSignalTree();
+        AtomicReference<CommandResult> result = new AtomicReference<>();
+
+        SignalCommand command = TestUtil.writeRootValueCommand();
+        tree.commitSingleCommand(command, result::set);
+
+        assertTrue(result.get().accepted());
+    }
+
+    @Test
+    void submitted_afterCommand_immediatelyUpdated() {
+        LocalAsynchronousSignalTree tree = new LocalAsynchronousSignalTree();
+
+        tree.commitSingleCommand(TestUtil.writeRootValueCommand());
+
+        assertNotNull(TestUtil.readSubmittedRootValue(tree));
+    }
+
+    @Test
+    void confirmed_afterCommand_immediatelyUpdated() {
+        LocalAsynchronousSignalTree tree = new LocalAsynchronousSignalTree();
+
+        tree.commitSingleCommand(TestUtil.writeRootValueCommand());
+
+        assertNotNull(TestUtil.readConfirmedRootValue(tree));
+    }
+
+    @Test
+    void multipleCommands_allImmediatelyConfirmed() {
+        LocalAsynchronousSignalTree tree = new LocalAsynchronousSignalTree();
+
+        for (int i = 0; i < 3; i++) {
+            AtomicReference<CommandResult> result = new AtomicReference<>();
+            tree.commitSingleCommand(
+                    TestUtil.writeRootValueCommand("value" + i), result::set);
+            assertTrue(result.get().accepted());
+        }
+    }
+
+    @Test
+    void submit_withAsyncDispatcher_confirmedOnlyAfterDispatch() {
+        TestExecutor dispatcher = useTestEffectDispatcher();
+        LocalAsynchronousSignalTree tree = new LocalAsynchronousSignalTree();
+
+        tree.commitSingleCommand(TestUtil.writeRootValueCommand());
+
+        assertNotNull(TestUtil.readSubmittedRootValue(tree));
+        assertNull(TestUtil.readConfirmedRootValue(tree));
+
+        dispatcher.runPendingTasks();
+
+        assertNotNull(TestUtil.readConfirmedRootValue(tree));
+    }
+}