Fix SSE integration test comparison logic to use correct series pattern (#1941)

Signed-off-by: Dave Skender <[email protected]>
Signed-off-by: GitHub <[email protected]>
Co-authored-by: copilot-swe-agent[bot] <[email protected]>
Co-authored-by: DaveSkender <[email protected]>
Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

Author: 3 people

.devcontainer/devcontainer.json

@@ -21,6 +21,7 @@
     "vscode": {
       "settings": {
         "chat.tools.global.autoApprove": true,
+        "dotnet.defaultSolution": "Stock.Indicators.sln",
         "terminal.integrated.defaultProfile.linux": "zsh"
       },
       "extensions": [

docs/features/stream.md

@@ -106,7 +106,7 @@ quoteHub.Add(quote1);
 quoteHub.Add(quote2);
 
 // late-arriving data with earlier timestamp
-quoteHub.Insert(lateQuote);  // triggers recalculation
+quoteHub.Add(lateQuote);  // triggers recalculation in dependent hubs
 
 // remove incorrect quote
 quoteHub.Remove(badQuote);   // triggers recalculation
@@ -125,8 +125,8 @@ The hub automatically handles state rollback and recalculation when data arrives
 
 Stream hubs use internal locking to protect cache operations during rebuild and rollback scenarios:
 
-- **Internal cache operations** are thread-safe (Insert, RemoveAt, RemoveRange, Rebuild)
-- **External access requires synchronization** when multiple threads call Add, Insert, or Remove
+- **Internal cache operations** are thread-safe (Add, RemoveAt, RemoveRange, Rebuild)
+- **External access requires synchronization** when multiple threads call Add or Remove
 - **Single-threaded usage** requires no additional synchronization
 - **Multi-threaded usage** should synchronize external calls to hub methods
 

docs/guide.md

@@ -451,7 +451,7 @@ quoteHub.Add(quote1);
 quoteHub.Add(quote2);
 
 // Late-arriving data with earlier timestamp
-quoteHub.Insert(lateQuote);  // Triggers recalculation
+quoteHub.Add(lateQuote);  // Triggers recalculation in dependent hubs
 
 // Remove incorrect quote
 quoteHub.Remove(badQuote);   // Triggers recalculation
@@ -747,7 +747,7 @@ await Task.WhenAll(
 **You need locks or channels when:**
 
 - Sharing a single hub instance across multiple threads
-- Multiple threads call `.Add()`, `.Insert()`, or `.Remove()` on the same hub
+- Do not call `.Add()` or `.Remove()` on the same hub from multiple threads
 - Reading results from one thread while another thread updates the hub
 - Processing quotes from multiple concurrent sources into one hub
 

docs/migration.md

@@ -347,6 +347,7 @@ Popular indicators with complete streaming documentation:
 | `UseResult` | `TimeValue` | Type renamed |
 | `SmaAnalysis` | `SmaAnalysisResult` | Type renamed |
 | `UlcerIndexResult.UI` | `UlcerIndexResult.UlcerIndex` | Property renamed |
+| `StreamHub.Insert()` | `StreamHub.Add()` | Insert removed; use Add |
 | `SyncSeries()` | (removed) | Use manual alignment |
 | `Find()` / `FindIndex()` | LINQ methods | Use `.FirstOrDefault()` etc. |
 | `GetBaseQuote()` | `Use(CandlePart)` | Use utility instead |

docs/performance.md

@@ -170,7 +170,7 @@ StreamHub style adds observable patterns and state management:
 Series (502 quotes):      ~25μs total
 StreamHub (502 quotes):   ~32μs total (~28% overhead)
 Per-quote latency:        ~64ns average
-Rollback (Insert):        ~2-5μs for state rebuild
+Rollback (late Add):      ~2-5μs for state rebuild
 ```
 
 **Scaling Characteristics:**
@@ -206,7 +206,7 @@ Real-time performance targets for trading applications:
 | Single indicator per quote         | <100μs  | 60-80μs             |
 | 5 indicators on hub per quote      | <500μs  | 300-400μs           |
 | Complex chains (EMA→RSI→Slope)     | <200μs  | 120-150μs           |
-| State rebuild (Insert/Remove)      | <5ms    | 2-3ms               |
+| State rebuild (Add/Remove)         | <5ms    | 2-3ms               |
 
 ### When to Choose Each Style
 

docs/plans/test-pruning.md

@@ -123,7 +123,7 @@ For the remaining 75 indicators, add validation following these patterns:
 - **T3**: `ValidateCacheSize(lookbackPeriods * 6, Name);`
 - **Multi-parameter**: Calculate `requiredWarmup` based on indicator-specific logic, then call `ValidateCacheSize(requiredWarmup, Name);`
 
-Insert validation call immediately before `Reinitialize()` in each indicator's constructor.
+Add validation call immediately before `Reinitialize()` in each indicator's constructor.
 
 ---
 Last updated: January 26, 2026

src/_common/Enums/Act.cs

@@ -15,13 +15,8 @@ internal enum Act
     /// </summary>
     Ignore = 1,
 
-    /// <summary>
-    /// Insert item without rebuilding cache.
-    /// </summary>
-    Insert = 2,
-
     /// <summary>
     /// Reset and rebuild from marker position.
     /// </summary>
-    Rebuild = 3
+    Rebuild = 2
 }

src/_common/QuotePart/QuotePart.StreamHub.cs

@@ -19,6 +19,24 @@ CandlePart candlePart
     /// <inheritdoc/>
     public CandlePart CandlePartSelection { get; init; }
 
+    /// <inheritdoc/>
+    public override void OnAdd(IQuote item, bool notify, int? indexHint)
+    {
+        // Lock to prevent concurrent cache access.
+        lock (CacheLock)
+        {
+            (TimeValue result, int index) = ToIndicator(item, indexHint);
+
+            if (index >= 0 && index < Cache.Count)
+            {
+                InsertWithoutRebuild(result, index, notify);
+                return;
+            }
+
+            AppendCache(result, notify);
+        }
+    }
+
 
     /// <inheritdoc/>
     protected override (TimeValue result, int index)

src/_common/Quotes/Quote.StreamHub.cs

@@ -84,6 +84,17 @@ public override void OnAdd(IQuote item, bool notify, int? indexHint)
     {
         ArgumentNullException.ThrowIfNull(item);
 
+        // Reject additions that precede the current cache timeline
+        // (applies to both standalone and non-standalone QuoteHub)
+        lock (CacheLock)
+        {
+            if (Cache.Count > 0 && item.Timestamp < Cache[0].Timestamp)
+            {
+                // Silently ignore - this prevents indeterminate gaps in the timeline
+                return;
+            }
+        }
+
         // for non-standalone QuoteHub, use standard behavior (which handles locking)
         if (!_isStandalone)
         {
@@ -94,9 +105,19 @@ public override void OnAdd(IQuote item, bool notify, int? indexHint)
         // Lock for standalone QuoteHub operations
         lock (CacheLock)
         {
+
             // get result and position
             (IQuote result, int index) = ToIndicator(item, indexHint);
 
+            // Reject modifications that would affect indices before MinCacheSize
+            // to prevent corrupted rebuilds in subscribers
+            // This includes both insertions and same-timestamp replacements
+            if (index >= 0 && index < MinCacheSize && index < Cache.Count)
+            {
+                // Silently ignore all modifications before MinCacheSize
+                return;
+            }
+
             // check if this is a same-timestamp update (not a new item at the end)
             if (Cache.Count > 0 && index < Cache.Count && Cache[index].Timestamp == result.Timestamp)
             {
@@ -119,6 +140,23 @@ public override void OnAdd(IQuote item, bool notify, int? indexHint)
             }
             else
             {
+                // if out-of-order insert, insert and trigger rebuild
+                if (index >= 0 && index < Cache.Count)
+                {
+                    Cache.Insert(index, result);
+
+                    // For standalone QuoteHub, notify observers to rebuild from this timestamp
+                    // For non-standalone, this won't be reached due to earlier branch
+                    if (notify)
+                    {
+                        // Notify observers directly - they will rebuild from the updated cache
+                        // No need to call Rebuild on QuoteHub itself since cache is already updated
+                        NotifyObserversOnRebuild(result.Timestamp);
+                    }
+
+                    return;
+                }
+
                 // standard add behavior for new items
                 AppendCache(result, notify);
             }

src/_common/StreamHub/IStreamHub.cs

@@ -60,19 +60,6 @@ public interface IStreamHub<in TIn, out TOut> : IStreamObserver<TIn>, IStreamObs
     /// </param>
     void Add(IEnumerable<TIn> batchIn);
 
-    /// <summary>
-    /// Insert a new item without rebuilding the cache.
-    /// </summary>
-    /// <remarks>
-    /// This is used in situations when inserting an older item
-    /// and where newer cache entries do not need to be rebuilt.
-    /// Typically, this is only used for provider-only hubs.
-    /// </remarks>
-    /// <param name="newIn">
-    /// Item to insert
-    /// </param>
-    void Insert(TIn newIn);
-
     /// <summary>
     /// Delete an item from the cache, from a specific position.
     /// </summary>