test(canopen): wait on observables instead of sleeping

src/CanKit.Pro.CANopen/ObjectDictionary.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using CanKit.Pro.CANopen.Sdo;
 
 namespace CanKit.Pro.CANopen;
@@ -43,6 +44,17 @@ public sealed class ObjectDictionary
     // may. Readers never take this lock, so a validator reading under _sync cannot deadlock.
     private readonly object _writeGate = new();
 
+    // Test seam (#171): the number of callers that have reached the write gate and are not yet
+    // inside it — WriteUnsigned's and Add's own <c>lock (_writeGate)</c>, the only two entry
+    // points a concurrent test "hammer" reaches. While a test holds the gate, every caller
+    // counted here is blocked on it; tests poll it as that signal instead of a fixed sleep that
+    // only ever guessed how long reaching the gate takes. Not counted for Transaction /
+    // WriteRawUnchecked / Declare, which no such test hammers.
+    private int _writeGateWaiters;
+
+    /// <summary>Test seam (#171): see <see cref="_writeGateWaiters"/>.</summary>
+    internal int WriteGateWaiters => Volatile.Read(ref _writeGateWaiters);
+
     /// <summary>
     /// Internal hook for <see cref="CanOpenNode"/>: raised after a value-mutating write
     /// (<see cref="WriteRaw"/> / <see cref="WriteUnsigned"/>, or an <c>Add*</c> call that
@@ -394,8 +406,10 @@ public void WriteUnsigned(ushort index, byte subindex, uint value)
         // The type is resolved under the write gate, so a re-declaration is either fully before
         // or fully after this write: the value is encoded and range-checked against the
         // declaration it lands on (Codex on #133).
+        Interlocked.Increment(ref _writeGateWaiters);
         lock (_writeGate)
         {
+            Interlocked.Decrement(ref _writeGateWaiters);
             OdDataType type;
             lock (_sync)
             {
@@ -437,8 +451,10 @@ private OdEntry Add(ushort index, byte subindex, OdDataType type, OdAccess acces
         bool replaced;
         // Under the write gate: a write validated against the entry being replaced is stored on
         // it before the replacement lands, never on the new declaration (Codex on #133).
+        Interlocked.Increment(ref _writeGateWaiters);
         lock (_writeGate)
         {
+            Interlocked.Decrement(ref _writeGateWaiters);
             lock (_sync)
             {
                 var key = Key(index, subindex);

tests/CanKit.Pro.Tests/TestCases/CANopen/CanOpenCommunicationProfileTests.cs

@@ -91,6 +91,25 @@ private static TaskCompletionSource<T> NewTcs<T>()
 
     private static uint HeartbeatEntry(byte nodeId, ushort milliseconds) => ((uint)nodeId << 16) | milliseconds;
 
+    /// <summary>
+    /// Spin-waits, synchronously, until <see cref="ObjectDictionary.WriteGateWaiters"/> (the
+    /// write-gate test seam, #171) reaches at least <paramref name="count"/> — real evidence that
+    /// another writer's own <c>lock (_writeGate)</c> attempt has reached the gate this call is
+    /// itself holding, rather than a fixed sleep that only ever guessed how long that takes. Used
+    /// from inside a <c>WriteValidator</c> callback, which runs synchronously on the holding
+    /// thread, so the wait is a blocking spin, not an awaited one.
+    /// </summary>
+    private static void SpinUntilAtTheWriteGate(ObjectDictionary od, int count = 1)
+    {
+        var deadline = DateTime.UtcNow + ShortTimeout;
+        while (od.WriteGateWaiters < count)
+        {
+            if (DateTime.UtcNow >= deadline)
+                throw new TimeoutException($"Fewer than {count} writer(s) reached the write gate within {ShortTimeout}.");
+            Thread.Sleep(1);
+        }
+    }
+
     /// <summary>
     /// Counts the boot-up frames (<c>00h</c> on <c>700h + producer</c>) one node sees from another.
     /// A node sends one when it is opened and one on every NMT reset; a test that waits for the
@@ -439,7 +458,7 @@ public async Task A_Redeclaration_Waits_For_The_Write_In_Flight_On_The_Entry()
             if (index == 0x2000 && redeclare is null)
             {
                 redeclare = Task.Run(() => od.AddU8(0x2000, 0x00, 0x01));
-                Thread.Sleep(200); // a re-declaration that did not wait for the gate would land here
+                SpinUntilAtTheWriteGate(od); // the re-declaration is genuinely blocked on this gate
             }
             return inner(index, subindex, value);
         };
@@ -499,8 +518,8 @@ public async Task A_Typed_Write_Resolves_Its_Type_Under_The_Write_Gate()
             {
                 gateHeld.Set();
                 writeStarted.Wait(ShortTimeout);
-                Thread.Sleep(200);          // the typed write has started and waits for the gate
-                od.AddU32(0x2000, 0x00, 0); // re-declared while the gate is held
+                SpinUntilAtTheWriteGate(od); // the typed write is genuinely blocked on this gate
+                od.AddU32(0x2000, 0x00, 0);  // re-declared while the gate is held
             }
             return inner(index, subindex, value);
         };
@@ -642,7 +661,7 @@ public async Task A_Direct_Write_Cannot_Land_Inside_A_ConfigureTpdo_Transaction(
             if (index == 0x1800 && subindex == 0x01 && !sequenceStarted.IsSet)
             {
                 sequenceStarted.Set();
-                Thread.Sleep(100); // the hammers are at the gate before the sequence goes on
+                SpinUntilAtTheWriteGate(od, count: 4); // all four hammers are genuinely blocked on this gate
             }
             return inner(index, subindex, value);
         };

tests/CanKit.Pro.Tests/TestCases/CANopen/CanOpenDynamicMappingTests.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using CanKit.Abstractions.API.Can;
@@ -51,6 +52,23 @@ private static Task CreatePdoAsync(ICanOpenNode client, byte serverNodeId, ushor
         return client.SdoDownloadAsync(serverNodeId, commIndex, 0x01, U32Bytes(cobId)).WithTimeoutAsync(ShortTimeout);
     }
 
+    // The barrier for "the NMT Start each node was just sent has been dequeued off its bus and
+    // applied by ApplyNmtTransition": ICanOpenNode.State round-trips through the node's actor
+    // (CanOpenNode.State getter posts to the actor and returns what it reads there), so polling
+    // it cannot observe a state the actor has not actually reached yet. It replaces a fixed
+    // Task.Delay that only ever guessed how long the dequeue-and-apply hop would take.
+    private static async Task WaitUntilOperationalAsync(params ICanOpenNode[] nodes)
+    {
+        var deadline = DateTime.UtcNow + ShortTimeout;
+        while (true)
+        {
+            if (nodes.All(node => node.State == NmtState.Operational)) return;
+            if (DateTime.UtcNow >= deadline)
+                throw new TimeoutException($"Node(s) did not reach Operational within {ShortTimeout}.");
+            await Task.Delay(5);
+        }
+    }
+
     private static byte[] MappingEntryBytes(ushort index, byte subindex, byte bitLength)
     {
         uint raw = ((uint)index << 16) | ((uint)subindex << 8) | bitLength;
@@ -122,7 +140,7 @@ await WriteMappingAsync(consumer, serverNodeId: 0x11, mapIndex: 0x1A00,
 
         await consumer.SendNmtCommandAsync(NmtCommand.Start, targetNodeId: 0x11);
         await producer.SendNmtCommandAsync(NmtCommand.Start, targetNodeId: 0x01);
-        await Task.Delay(50);
+        await WaitUntilOperationalAsync(producer, consumer);
         await producer.TriggerTpdoAsync(1);
 
         var payload = await received.Task.WithTimeoutAsync(ShortTimeout);
@@ -168,7 +186,7 @@ await WriteMappingAsync(producer, serverNodeId: 0x01, mapIndex: 0x1600,
 
         await consumer.SendNmtCommandAsync(NmtCommand.Start, targetNodeId: 0x11);
         await producer.SendNmtCommandAsync(NmtCommand.Start, targetNodeId: 0x01);
-        await Task.Delay(50);
+        await WaitUntilOperationalAsync(producer, consumer);
         await producer.TriggerTpdoAsync(1);
 
         await received.Task.WithTimeoutAsync(ShortTimeout);
@@ -371,7 +389,7 @@ public async Task Tpdo_ChangeOfState_Emits_On_ApplicationOdWrite()
 
         await consumer.SendNmtCommandAsync(NmtCommand.Start, targetNodeId: 0x11);
         await producer.SendNmtCommandAsync(NmtCommand.Start, targetNodeId: 0x01);
-        await Task.Delay(50);
+        await WaitUntilOperationalAsync(producer, consumer);
 
         // Application-originated OD write — no TriggerTpdoAsync.
         producer.ObjectDictionary.WriteUnsigned(0x2000, 0x00, 0x1234u);
@@ -420,7 +438,7 @@ public async Task Tpdo_ChangeOfState_DoesNotEcho_On_RpdoUnpack()
 
         await consumer.SendNmtCommandAsync(NmtCommand.Start, targetNodeId: 0x11);
         await producer.SendNmtCommandAsync(NmtCommand.Start, targetNodeId: 0x01);
-        await Task.Delay(50);
+        await WaitUntilOperationalAsync(producer, consumer);
         await producer.TriggerTpdoAsync(1);
         await Task.Delay(300); // give any (wrong) echo ample time to appear