From 69c281ccd11e7829df21cd5816a30567774c2ffe Mon Sep 17 00:00:00 2001
From: Jasper <jasper.lammers@gmail.com>
Date: Sat, 3 Jan 2026 16:44:01 +0100
Subject: [PATCH] Implemented lock for pipeline node output changes.

---
 src/CodeCasa.AutomationPipelines/Pipeline.cs | 17 +++++++++++++----
 1 file changed, 13 insertions(+), 4 deletions(-)

diff --git a/src/CodeCasa.AutomationPipelines/Pipeline.cs b/src/CodeCasa.AutomationPipelines/Pipeline.cs
index c93fc2c..362eadd 100644
--- a/src/CodeCasa.AutomationPipelines/Pipeline.cs
+++ b/src/CodeCasa.AutomationPipelines/Pipeline.cs
@@ -7,6 +7,7 @@ namespace CodeCasa.AutomationPipelines;
 /// </summary>
 public class Pipeline<TState> : PipelineNode<TState>, IPipeline<TState>
 {
+    private readonly Lock _lock = new();
     private readonly List<IPipelineNode<TState>> _nodes = new();
     private readonly ILogger<Pipeline<TState>>? _logger;
 
@@ -136,8 +137,12 @@ public IPipeline<TState> RegisterNode(IPipelineNode<TState> node)
             var destinationIndex = _nodes.Count;
             previousNode.OnNewOutput.Subscribe(output =>
             {
-                _logger?.LogTrace($"{LogPrefix}[Node {sourceIndex}] ({previousNode}) passed value [{output?.ToString() ?? "NULL"}] to [Node {destinationIndex}] ({node}).");
-                node.Input = output;
+                lock (_lock)
+                {
+                    _logger?.LogTrace(
+                        $"{LogPrefix}[Node {sourceIndex}] ({previousNode}) passed value [{output?.ToString() ?? "NULL"}] to [Node {destinationIndex}] ({node}).");
+                    node.Input = output;
+                }
             });
 
             _logger?.LogTrace($"{LogPrefix}Passing [Node {sourceIndex}] ({previousNode}) value [{previousNode.Output?.ToString() ?? "NULL"}] to [Node {destinationIndex}] ({node}).");
@@ -157,8 +162,12 @@ public IPipeline<TState> RegisterNode(IPipelineNode<TState> node)
         var nodeIndex = _nodes.Count - 1;
         _subscription = node.OnNewOutput.Subscribe(o =>
         {
-            _logger?.LogTrace($"{LogPrefix}[Node {nodeIndex}] ({node}) passed value [{o?.ToString() ?? "NULL"}] to pipeline output.");
-            SetOutputAndCallActionWhenApplicable(o);
+            lock (_lock)
+            {
+                _logger?.LogTrace(
+                    $"{LogPrefix}[Node {nodeIndex}] ({node}) passed value [{o?.ToString() ?? "NULL"}] to pipeline output.");
+                SetOutputAndCallActionWhenApplicable(o);
+            }
         });
 
         var newOutput = node.Output;