Specs/grains observation

.editorconfig

@@ -140,6 +140,7 @@ dotnet_diagnostic.CA2000.severity = none
 dotnet_diagnostic.CA2201.severity = none
 dotnet_diagnostic.CA2211.severity = none
 dotnet_diagnostic.CA2252.severity = none
+dotnet_diagnostic.CS0108.severity = none
 dotnet_diagnostic.CS8600.severity = none
 dotnet_diagnostic.CS8601.severity = none
 dotnet_diagnostic.CS8602.severity = none
@@ -153,6 +154,7 @@ dotnet_diagnostic.IDE0005.severity = none
 dotnet_diagnostic.IDE0044.severity = none
 dotnet_diagnostic.IDE0051.severity = none
 dotnet_diagnostic.IDE0052.severity = none
+dotnet_diagnostic.IDE0058.severity = none
 dotnet_diagnostic.IDE0060.severity = none
 dotnet_diagnostic.IDE0240.severity = none
 dotnet_diagnostic.IDE1006.severity = none
@@ -196,3 +198,4 @@ dotnet_diagnostic.MA0069.severity = none
 dotnet_diagnostic.MA0096.severity = none
 dotnet_diagnostic.MA0097.severity = none
 dotnet_diagnostic.MA0132.severity = none
+dotnet_diagnostic.MA0134.severity = none

Integration/Orleans.InProcess/for_Reactors/when_appending_event/and_waiting_for_observer_to_be_active.cs

@@ -74,9 +74,6 @@ async Task Because()
     [Fact]
     void should_have_correct_observer_state_next_event_sequence_number_for_event_types() => Context.ReactorObserverState.NextEventSequenceNumberForEventTypes.Value.ShouldEqual(EventSequenceNumber.Unavailable.Value);
 
-    [Fact]
-    void should_have_correct_observer_state_handled_number() => Context.ReactorObserverState.Handled.ShouldEqual(1);
-
     [Fact]
     void should_not_have_failed_partitions() => Context.FailedPartitions.HasFailedPartitions.ShouldBeFalse();
 }

Source/Api/Observation/ObserverCommands.cs

@@ -46,7 +46,7 @@ public async Task TryRecoverFailedPartition(
         [FromRoute] ObserverId observerId,
         [FromRoute] string partition)
     {
-        await grainFactory.GetGrain<IObserver>(new ObserverKey(observerId, eventStore, @namespace, EventSequenceId.Log)).TryRecoverFailedPartition(partition);
+        await grainFactory.GetGrain<IObserver>(new ObserverKey(observerId, eventStore, @namespace, EventSequenceId.Log)).TryStartRecoverJobForFailedPartition(partition);
     }
 
     /// <summary>

Source/Infrastructure/Properties/ArrayIndexers.cs

@@ -15,6 +15,8 @@ public class ArrayIndexers
 
     readonly IDictionary<PropertyPath, ArrayIndexer> _arrayIndexers = new Dictionary<PropertyPath, ArrayIndexer>();
 
+    int? _computedHashCode;
+
     /// <summary>
     /// Initializes a new instance of the <see cref="ArrayIndexers"/> class.
     /// </summary>
@@ -37,7 +39,7 @@ public ArrayIndexers(IEnumerable<ArrayIndexer> arrayIndexers)
     public int Count => _arrayIndexers.Count;
 
     /// <summary>
-    /// Gets a value indicating whether or not it is empty.
+    /// Gets a value indicating whether it is empty.
     /// </summary>
     public bool IsEmpty => _arrayIndexers.Count == 0;
 
@@ -81,7 +83,15 @@ public override bool Equals(object? obj)
     }
 
     /// <inheritdoc/>
-    public override int GetHashCode() => base.GetHashCode();
+    public override int GetHashCode()
+    {
+        // From https://stackoverflow.com/a/8094931
+        unchecked
+        {
+            _computedHashCode ??= All.Aggregate(19, (current, item) => (current * 31) + item.GetHashCode());
+            return _computedHashCode.Value;
+        }
+    }
 
     void ThrowIfMissingArrayIndexerForPropertyPath(PropertyPath propertyPath)
     {

Source/Kernel/Concepts/Observation/FailedPartition.cs

@@ -37,14 +37,14 @@ public IEnumerable<FailedPartitionAttempt> Attempts
     }
 
     /// <summary>
-    /// Gets or sets whether or not the failure is resolved.
+    /// Gets or sets whether the failure is resolved.
     /// </summary>
     public bool IsResolved { get; set; }
 
     /// <summary>
     /// Gets the last attempt for the failed partition.
     /// </summary>
-    public FailedPartitionAttempt? LastAttempt => Attempts.LastOrDefault();
+    public FailedPartitionAttempt LastAttempt { get; private set; } = FailedPartitionAttempt.NoAttempt;
 
     /// <summary>
     /// Add an attempt to the failed partition.
@@ -53,5 +53,6 @@ public IEnumerable<FailedPartitionAttempt> Attempts
     public void AddAttempt(FailedPartitionAttempt attempt)
     {
         Attempts = Attempts.Append(attempt);
+        LastAttempt = attempt;
     }
 }

Source/Kernel/Concepts/Observation/FailedPartitionAttempt.cs

@@ -10,6 +10,11 @@ namespace Cratis.Chronicle.Concepts.Observation;
 /// </summary>
 public class FailedPartitionAttempt
 {
+    /// <summary>
+    /// Gets the empty <see cref="FailedPartitionAttempt"/>.
+    /// </summary>
+    public static FailedPartitionAttempt NoAttempt => new();
+
     /// <summary>
     /// Gets or sets when the attempt occurred.
     /// </summary>

Source/Kernel/Concepts/Observation/FailedPartitionId.cs

@@ -9,6 +9,8 @@ namespace Cratis.Chronicle.Concepts.Observation;
 /// <param name="value">The inner value.</param>
 public record FailedPartitionId(Guid value) : ConceptAs<Guid>(value)
 {
+    public static implicit operator FailedPartitionId(Guid value) => new(value);
+
     /// <summary>
     /// Creates a new instance of the <see cref="FailedPartitionId"/> class with a new unique value.
     /// </summary>

Source/Kernel/Concepts/Observation/FailedPartitions.cs

@@ -1,6 +1,7 @@
 // Copyright (c) Cratis. All rights reserved.
 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
 
+using System.Diagnostics.CodeAnalysis;
 using Cratis.Chronicle.Concepts.Events;
 using Cratis.Chronicle.Concepts.Keys;
 
@@ -12,46 +13,49 @@ namespace Cratis.Chronicle.Concepts.Observation;
 public class FailedPartitions
 {
     readonly List<FailedPartition> _resolvedPartitions = [];
-    List<FailedPartition> _partitions = [];
+    readonly Dictionary<Key, FailedPartition> _partitions = [];
 
     /// <summary>
     /// Gets or sets the failed partitions for the observer.
     /// </summary>
     public IEnumerable<FailedPartition> Partitions
     {
-        get => _partitions;
-        set => _partitions = new(value);
+        get => _partitions.Values;
+        set
+        {
+            _partitions.Clear();
+            foreach (var failedPartition in value)
+            {
+                var key = failedPartition.Partition;
+                _partitions.Add(key, failedPartition);
+            }
+        }
     }
 
     /// <summary>
-    /// Gets or sets the resolved partitions for the observer.
+    /// Gets the resolved partitions for the observer.
     /// </summary>
     public IEnumerable<FailedPartition> ResolvedPartitions => _resolvedPartitions;
 
     /// <summary>
-    /// Gets whether or not there are any failed partitions.
+    /// Gets a value indicating whether there are any failed partitions.
     /// </summary>
     public bool HasFailedPartitions => _partitions.Count > 0;
 
     /// <summary>
-    /// Add a failed partition.
-    /// </summary>
-    /// <param name="failedPartition"><see cref="FailedPartition"/> to add.</param>
-    public void Add(FailedPartition failedPartition) => _partitions.Add(failedPartition);
-
-    /// <summary>
-    /// Check whether or not a partition is failed.
+    /// Check whether a partition is failed.
     /// </summary>
     /// <param name="partition">Partition to check.</param>
     /// <returns>True if failed, false if not.</returns>
-    public bool IsFailed(Key partition) => _partitions.Exists(_ => _.Partition.Equals(partition));
+    public bool IsFailed(Key partition) => _partitions.ContainsKey(partition);
 
     /// <summary>
-    /// Gets a failed partition by its partition identifier.
+    /// Try to get a failed partition by its partition identifier.
     /// </summary>
     /// <param name="partition">Partition to get.</param>
-    /// <returns>The failed partition.</returns>
-    public FailedPartition? Get(Key partition) => _partitions.Find(_ => _.Partition == partition);
+    /// <param name="failedPartition">The optional failed partition.</param>
+    /// <returns>True when failed partition exists, false if not.</returns>
+    public bool TryGet(Key partition, [NotNullWhen(true)]out FailedPartition? failedPartition) => _partitions.TryGetValue(partition, out failedPartition);
 
     /// <summary>
     /// Register an attempt for a partition.
@@ -67,19 +71,13 @@ public FailedPartition RegisterAttempt(
         IEnumerable<string> messages,
         string stackTrace)
     {
-        FailedPartition failure;
-        if (IsFailed(partition))
+        if (!TryGet(partition, out var failure))
         {
-            failure = Get(partition)!;
-        }
-        else
-        {
-            failure = new FailedPartition
+            failure = new()
             {
                 Id = FailedPartitionId.New(),
                 Partition = partition
             };
-
             Add(failure);
         }
 
@@ -100,11 +98,11 @@ public FailedPartition RegisterAttempt(
     /// <param name="partition"><see cref="Key"/> to remove.</param>
     public void Remove(Key partition)
     {
-        var failedPartition = Get(partition);
-        if (failedPartition != null)
-        {
-            _resolvedPartitions.Add(failedPartition);
-        }
-        _partitions.RemoveAll(_ => _.Partition == partition);
+        if (!TryGet(partition, out var failedPartition)) return;
+
+        _resolvedPartitions.Add(failedPartition);
+        _partitions.Remove(partition);
     }
+
+    void Add(FailedPartition failedPartition) => _partitions.Add(failedPartition.Partition, failedPartition);
 }

Source/Kernel/Concepts/Observation/ObserverInformation.cs

@@ -16,7 +16,6 @@ namespace Cratis.Chronicle.Concepts.Observation;
 /// <param name="NextEventSequenceNumber">The next event sequence number the observer will observe.</param>
 /// <param name="LastHandledEventSequenceNumber">The event sequence number the observer last handled.</param>
 /// <param name="RunningState">The running state of the observer.</param>
-/// <param name="Handled">Number of events handled by the observer.</param>
 /// <param name="FailedPartitions">Collection of <see cref="FailedPartition"/>.</param>
 public record ObserverInformation(
     ObserverId ObserverId,
@@ -26,5 +25,4 @@ public record ObserverInformation(
     EventSequenceNumber NextEventSequenceNumber,
     EventSequenceNumber LastHandledEventSequenceNumber,
     ObserverRunningState RunningState,
-    EventCount Handled,
     IEnumerable<FailedPartition> FailedPartitions);

Source/Kernel/Grains.Interfaces/EventSequences/IEventSequence.cs

@@ -55,7 +55,7 @@ public interface IEventSequence : IGrainWithStringKey
     /// <p>If no event is found, it will return <see cref="EventSequenceNumber.Unavailable"/>.</p>
     /// </returns>
     [AlwaysInterleave]
-    Task<EventSequenceNumber> GetNextSequenceNumberGreaterOrEqualThan(EventSequenceNumber sequenceNumber, IEnumerable<EventType>? eventTypes = null, EventSourceId? eventSourceId = null);
+    Task<EventSequenceNumber> GetNextSequenceNumberGreaterOrEqualTo(EventSequenceNumber sequenceNumber, IEnumerable<EventType>? eventTypes = null, EventSourceId? eventSourceId = null);
 
     /// <summary>
     /// Append a single event to the event store.

Source/Kernel/Grains.Interfaces/Jobs/IJob.cs

@@ -108,7 +108,7 @@ public interface IJob : IGrainWithGuidCompoundKey
 }
 
 /// <summary>
-/// Represents a job that typically runs as long running with <see cref="IJobStep{TRequest, TResult}"/>.
+/// Represents a job that typically runs as long-running with <see cref="IJobStep{TRequest, TResult}"/>.
 /// </summary>
 /// <typeparam name="TRequest">Type of request object that gets passed to job.</typeparam>
 public interface IJob<TRequest> : IJob

Source/Kernel/Grains.Interfaces/Jobs/JobStepResult.cs

@@ -15,7 +15,7 @@ namespace Cratis.Chronicle.Grains.Jobs;
 public record JobStepResult(JobStepStatus Status, IEnumerable<string> Messages, string ExceptionStackTrace, object? Result = null)
 {
     /// <summary>
-    /// Gets whether or not the job step was successful.
+    /// Gets whether the job step was successful.
     /// </summary>
     public bool IsSuccess => Status == JobStepStatus.Succeeded;
 

Source/Kernel/Grains.Interfaces/Observation/IObserver.cs

@@ -15,7 +15,7 @@ namespace Cratis.Chronicle.Grains.Observation;
 public interface IObserver : IStateMachine<ObserverState>, IGrainWithStringKey
 {
     /// <summary>
-    /// Ensure the observers existence.
+    /// Ensure the observer existence.
     /// </summary>
     /// <returns>Awaitable task.</returns>
     Task Ensure();
@@ -29,10 +29,9 @@ public interface IObserver : IStateMachine<ObserverState>, IGrainWithStringKey
     /// <summary>
     /// Set the handled stats for the observer.
     /// </summary>
-    /// <param name="handled">Number of handled events.</param>
     /// <param name="lastHandledEventSequenceNumber">The last handled event sequence number.</param>
     /// <returns>Awaitable task.</returns>
-    Task SetHandledStats(EventCount handled, EventSequenceNumber lastHandledEventSequenceNumber);
+    Task SetHandledStats(EventSequenceNumber lastHandledEventSequenceNumber);
 
     /// <summary>
     /// Get the subscription for the observer.
@@ -99,8 +98,9 @@ Task Subscribe<TObserverSubscriber>(
     /// Notify that the partition has been replayed.
     /// </summary>
     /// <param name="partition">The partition that has been replayed.</param>
+    /// <param name="lastHandledEventSequenceNumber">The event sequence number of the last event that as handled in the catchup.</param>
     /// <returns>Awaitable task.</returns>
-    Task PartitionReplayed(Key partition);
+    Task PartitionReplayed(Key partition, EventSequenceNumber lastHandledEventSequenceNumber);
 
     /// <summary>
     /// Notify that the partition has failed.
@@ -116,15 +116,24 @@ Task Subscribe<TObserverSubscriber>(
     /// Notify that the partition has recovered.
     /// </summary>
     /// <param name="partition">The partition that has recovered.</param>
+    /// <param name="lastHandledEventSequenceNumber">The event sequence number of the last event that as handled in the catchup.</param>
     /// <returns>Awaitable task.</returns>
-    Task FailedPartitionRecovered(Key partition);
+    Task FailedPartitionRecovered(Key partition, EventSequenceNumber lastHandledEventSequenceNumber);
+
+    /// <summary>
+    /// Notify that the partition was caught.
+    /// </summary>
+    /// <param name="partition">The partition that has caught up.</param>
+    /// <param name="lastHandledEventSequenceNumber">The event sequence number of the last event that as handled in the catchup.</param>
+    /// <returns>Awaitable task.</returns>
+    Task PartitionCaughtUp(Key partition, EventSequenceNumber lastHandledEventSequenceNumber);
 
     /// <summary>
     /// Attempt to recover a failed partition.
     /// </summary>
     /// <param name="partition">The partition that is failed.</param>
     /// <returns>Awaitable task.</returns>
-    Task TryRecoverFailedPartition(Key partition);
+    Task TryStartRecoverJobForFailedPartition(Key partition);
 
     /// <summary>
     /// Attempt to recover all failed partitions.
@@ -139,11 +148,4 @@ Task Subscribe<TObserverSubscriber>(
     /// <param name="events">Collection of <see cref="AppendedEvent"/>.</param>
     /// <returns>Awaitable task.</returns>
     Task Handle(Key partition, IEnumerable<AppendedEvent> events);
-
-    /// <summary>
-    /// Report a count of events that has been handled.
-    /// </summary>
-    /// <param name="count"><see cref="EventCount"/> to increase the handled count with.</param>
-    /// <returns>Awaitable task.</returns>
-    Task ReportHandledEvents(EventCount count);
 }

Source/Kernel/Grains.Interfaces/Observation/Jobs/CatchUpObserverPartitionRequest.cs

@@ -0,0 +1,24 @@
+// Copyright (c) Cratis. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using Cratis.Chronicle.Concepts.Events;
+using Cratis.Chronicle.Concepts.Keys;
+using Cratis.Chronicle.Concepts.Observation;
+namespace Cratis.Chronicle.Grains.Observation.Jobs;
+
+/// <summary>
+/// Represents the request for a <see cref="ICatchUpObserverPartition"/>.
+/// </summary>
+/// <param name="ObserverId">The identifier of the observer to replay.</param>
+/// <param name="ObserverKey">The additional <see cref="ObserverKey"/> for the observer to catch up.</param>
+/// <param name="ObserverSubscription">The <see cref="ObserverSubscription"/> for the observer.</param>
+/// <param name="Key"><see cref="Key">Partition</see> to retry.</param>
+/// <param name="FromSequenceNumber">From <see cref="EventSequenceNumber"/> to catch up from.</param>
+/// <param name="EventTypes">The event types.</param>
+public record CatchUpObserverPartitionRequest(
+    ObserverId ObserverId,
+    ObserverKey ObserverKey,
+    ObserverSubscription ObserverSubscription,
+    Key Key,
+    EventSequenceNumber FromSequenceNumber,
+    IEnumerable<EventType> EventTypes);