Added Contains(value) for durable ring and time-window buffers

Author: ledjon-behluli

DurableStateMachines.Tests/DurableRingBufferTests.cs

@@ -15,6 +15,7 @@ public interface IDurableRingBufferGrain : IGrainWithStringKey
         Task<int> GetCount();
         Task<bool> IsEmpty();
         Task<bool> IsFull();
+        Task<bool> Contains(string item);
         Task<List<string>> GetAll();
         Task<(int, string[])> CopyToArray(int arraySize, int arrayIndex);
         Task<(int, string[])> DrainToArray(int arraySize, int arrayIndex);
@@ -57,6 +58,7 @@ public async Task Clear()
         public Task<int> GetCount() => Task.FromResult(state.Count);
         public Task<bool> IsEmpty() => Task.FromResult(state.IsEmpty);
         public Task<bool> IsFull() => Task.FromResult(state.IsFull);
+        public Task<bool> Contains(string item) => Task.FromResult(state.Contains(item));
         public Task<List<string>> GetAll() => Task.FromResult(state.ToList());
 
         public Task<(int, string[])> CopyToArray(int arraySize, int arrayIndex)
@@ -128,6 +130,33 @@ public async Task BasicOperations()
         Assert.Equal("three", items2.Last());
     }
 
+    [Fact]
+    public async Task Contains()
+    {
+        var grain = GetGrain("contains");
+        await grain.SetCapacity(3);
+
+        Assert.False(await grain.Contains("one"));
+
+        await grain.Enqueue("one");
+        await grain.Enqueue("two");
+
+        Assert.True(await grain.Contains("one"));
+        Assert.True(await grain.Contains("two"));
+        Assert.False(await grain.Contains("three"));
+
+        await grain.Enqueue("three");
+        await grain.Enqueue("four");
+
+        Assert.False(await grain.Contains("one"));
+        Assert.True(await grain.Contains("four"));
+
+        await grain.TryDequeue();
+
+        Assert.False(await grain.Contains("two"));
+        Assert.True(await grain.Contains("three"));
+    }
+
     [Fact]
     public async Task TryDequeueRemovesOldestItem()
     {

DurableStateMachines.Tests/DurableTimeWindowBufferTests.cs

@@ -14,6 +14,7 @@ public interface IDurableTimeWindowBufferGrain : IGrainWithStringKey
         Task<TimeSpan> GetWindow();
         Task<int> GetCount();
         Task<bool> IsEmpty();
+        Task<bool> Contains(string value);
         Task<List<string>> GetAll();
         Task<(int, string[])> CopyToArray(int arraySize, int arrayIndex);
         Task<(int, string[])> DrainToArray(int arraySize, int arrayIndex);
@@ -55,6 +56,7 @@ public async Task Clear()
         public Task<TimeSpan> GetWindow() => Task.FromResult(state.Window);
         public Task<int> GetCount() => Task.FromResult(state.Count);
         public Task<bool> IsEmpty() => Task.FromResult(state.IsEmpty);
+        public Task<bool> Contains(string value) => Task.FromResult(state.Contains(value));
         public Task<List<string>> GetAll() => Task.FromResult(state.ToList());
 
         public Task<(int, string[])> CopyToArray(int arraySize, int arrayIndex)
@@ -117,6 +119,35 @@ public async Task BasicOperations()
         Assert.Equal("two", items1.Last());
     }
 
+    [Fact]
+    public async Task Contains()
+    {
+        var grain = GetGrain("contains");
+        await grain.SetWindow(TimeSpan.FromSeconds(10));
+
+        Assert.False(await grain.Contains("one"));
+
+        await grain.Enqueue("one"); // t=0
+        fixture.TimeProvider.Advance(TimeSpan.FromSeconds(5));
+        await grain.Enqueue("two"); // t=5
+
+        Assert.True(await grain.Contains("one"));
+        Assert.True(await grain.Contains("two"));
+        Assert.False(await grain.Contains("three"));
+
+        // Test after item expiration
+        fixture.TimeProvider.Advance(TimeSpan.FromSeconds(6)); // Time is now t=11. "one" is 11s old and expired.
+        await grain.Enqueue("three"); // Enqueue triggers purge. "one" is removed.
+
+        Assert.False(await grain.Contains("one"));
+        Assert.True(await grain.Contains("two"));
+        Assert.True(await grain.Contains("three"));
+
+        await grain.TryDequeue();
+        Assert.False(await grain.Contains("two"));
+        Assert.True(await grain.Contains("three"));
+    }
+
     [Fact]
     public async Task TryDequeueRemovesOldestItem()
     {

DurableStateMachines/DurableRingBuffer.cs

@@ -54,6 +54,16 @@ public interface IDurableRingBuffer<T> : IEnumerable<T>, IReadOnlyCollection<T>
     /// <returns><c>true</c> if an element was removed and returned from the buffer successfully; otherwise, <c>false</c>.</returns>
     bool TryDequeue([MaybeNullWhen(false)] out T result);
 
+    /// <summary>
+    /// Determines whether the buffer contains a specific value.
+    /// <para>
+    /// The comparison is performed using <see cref="EqualityComparer{T}.Default"/>.
+    /// </para>
+    /// </summary>
+    /// <param name="item">The object to locate in the buffer. The value can be <c>null</c> for reference types.</param>
+    /// <returns><c>true</c> if <paramref name="item"/> is found in the buffer; otherwise, <c>false</c>.</returns>
+    bool Contains(T item);
+
     /// <summary>
     /// Copies the elements of the buffer to an array, starting at a particular array index.
     /// The elements are copied in their logical order (from oldest to newest).
@@ -292,6 +302,7 @@ public void Clear()
         }
     }
 
+    public bool Contains(T item) => _buffer.Contains(item);
     public int CopyTo(T[] array, int arrayIndex) => _buffer.CopyTo(array, arrayIndex);
     public int CopyTo(Span<T> destination) => _buffer.CopyTo(destination);
 
@@ -356,6 +367,50 @@ internal sealed class RingBuffer<T> : IEnumerable<T>
     public bool IsEmpty => _count == 0;
     public bool IsFull => _count == Capacity;
 
+    public bool Contains(T item)
+    {
+        if (IsEmpty)
+        {
+            return false;
+        }
+
+        var span = new ReadOnlySpan<T>(_buffer);
+
+        // Case 1: The items are in a single contiguous block: [_, _, T, T, H, _]
+        if (_tail < _head)
+        {
+            return Exists(span, _tail, _count, item);
+        }
+
+        // Case 2: The items are wrapped around the end of the buffer: [T, T, H, _, T]
+        // This also handles the case where the buffer is full (_head == _tail).
+
+        // We search the first segment (from the tail to the end of the buffer).
+        if (Exists(span, _tail, Capacity - _tail, item))
+        {
+            return true;
+        }
+
+        // Otherwise, we search the second segment (from the start of the buffer up to the head).
+        return Exists(span, 0, _head, item);
+
+        static bool Exists(ReadOnlySpan<T> buffer, int index, int segmentLength, T item)
+        {
+            var comparer = EqualityComparer<T>.Default;
+            var end = index + segmentLength;
+
+            for (int i = index; i < end; i++)
+            {
+                if (comparer.Equals(buffer[i], item))
+                {
+                    return true;
+                }
+            }
+
+            return false;
+        }
+    }
+
     public void Enqueue(T item)
     {
         _buffer[_head] = item;

DurableStateMachines/DurableRingBufferCollection.cs

@@ -408,6 +408,7 @@ internal sealed class RingBufferProxy(TKey key,
         public bool IsEmpty => Buffer.IsEmpty;
         public bool IsFull => Buffer.IsFull;
 
+        public bool Contains(TValue item) => Buffer.Contains(item);
         public bool SetCapacity(int capacity) => collection.SetBufferCapacity(key, capacity);
         public void Enqueue(TValue item) => collection.EnqueueItem(key, item);
         public bool TryDequeue([MaybeNullWhen(false)] out TValue item) => collection.TryDequeueItem(key, out item);

DurableStateMachines/DurableTimeWindowBuffer.cs

@@ -49,6 +49,16 @@ public interface IDurableTimeWindowBuffer<T> : IEnumerable<T>, IReadOnlyCollecti
     /// <returns><c>true</c> if an element was removed and returned from the buffer successfully; otherwise, <c>false</c>.</returns>
     bool TryDequeue([MaybeNullWhen(false)] out T result);
 
+    /// <summary>
+    /// Determines whether the buffer contains a specific value.
+    /// <para>
+    /// The comparison is performed using <see cref="EqualityComparer{T}.Default"/>.
+    /// </para>
+    /// </summary>
+    /// <param name="item">The object to locate in the buffer. The value can be <c>null</c> for reference types.</param>
+    /// <returns><c>true</c> if <paramref name="item"/> is found in the buffer; otherwise, <c>false</c>.</returns>
+    bool Contains(T item);
+
     /// <summary>
     /// Copies the elements of the buffer to an array, starting at a particular array index.
     /// The elements are copied in their logical order (from oldest to newest).
@@ -297,6 +307,7 @@ public void Clear()
         }
     }
 
+    public bool Contains(T item) => _buffer.Contains(item);
     public int CopyTo(T[] array, int arrayIndex) => _buffer.CopyTo(array, arrayIndex);
     public int CopyTo(Span<T> destination) => _buffer.CopyTo(destination);
 
@@ -365,6 +376,21 @@ public static void ThrowIfTooShortWindow(long windowSeconds)
         }
     }
 
+    public bool Contains(T item)
+    {
+        var comparer = EqualityComparer<T>.Default;
+
+        foreach (var (current, _) in _buffer)
+        {
+            if (comparer.Equals(current, item))
+            {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
     public void Enqueue(T item, long timestamp)
     {
         _buffer.Enqueue((item, timestamp));

DurableStateMachines/DurableTimeWindowBufferCollection.cs

@@ -412,6 +412,7 @@ internal sealed class TimeWindowBufferProxy(TKey key,
         public bool IsEmpty => Buffer.IsEmpty;
         public TimeSpan Window => TimeSpan.FromSeconds(Buffer.WindowSeconds);
 
+        public bool Contains(TValue item) => Buffer.Contains(item);
         public bool SetWindow(TimeSpan window) => collection.SetBufferWindow(key, (long)window.TotalSeconds);
         public void Enqueue(TValue item) => collection.EnqueueItem(key, item);
         public bool TryDequeue([MaybeNullWhen(false)] out TValue item) => collection.TryDequeueItem(key, out item);