Fix/error with restarting

.github/workflows/pull-requests.yml

@@ -80,10 +80,10 @@ jobs:
         id: yarn-cache-dir-path
         run: echo "::set-output name=dir::$(yarn cache dir)"
 
-      - name: Setup node v16
+      - name: Setup node
         uses: actions/setup-node@v4
         with:
-          node-version: 16.x
+          node-version: 20.x
           registry-url: "https://registry.npmjs.org"
 
       - uses: actions/cache@v3
@@ -99,16 +99,13 @@ jobs:
             ${{ runner.os }}-yarn-
 
       - name: Yarn install
+        working-directory: ./Source/Workbench/Web
         run: yarn
 
       - name: Build JS/TS
-        run: |
-          export NODE_OPTIONS="--max-old-space-size=4096"
-          yarn build
-
-      - name: Build Workbench
         working-directory: ./Source/Workbench/Web
         run: |
+          export NODE_OPTIONS="--max-old-space-size=4096"
           yarn build
 
   publish-nuget-packages:
@@ -152,51 +149,6 @@ jobs:
             https://github.com/cratis/chronicle/packages/1655206?version=${{ needs.release.outputs.version }}
           allow-repeats: false
 
-  publish-npm-packages:
-    if: needs.release.outputs.publish == 'true'
-    runs-on: ubuntu-latest
-    needs: [release]
-
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@v4
-
-      - name: Setup node v16
-        uses: actions/setup-node@v4
-        with:
-          node-version: 16.x
-          registry-url: "https://registry.npmjs.org"
-
-
-      # - name: Get yarn cache directory path
-      #   id: yarn-cache-dir-path
-      #   run: echo "::set-output name=dir::$(yarn cache dir)"
-      # - uses: actions/cache@v3
-      #   id: yarn-cache
-      #   with:
-      #     path: |
-      #       **/node_modules
-      #       **/.eslintcache
-      #       **/yarn.lock
-      #       ${{ steps.yarn-cache-dir-path.outputs.dir }}
-      #     key: ${{ runner.os }}-yarn-${{ hashFiles('**/yarn.lock') }}
-      #     restore-keys: |
-      #       ${{ runner.os }}-yarn-
-
-      # - name: Yarn install
-      #   run: yarn
-
-      # - name: Publish NPM packages
-      #   env:
-      #     NPM_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
-      #     NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
-      #   run: |
-      #     yarn publish-version ${{ needs.release.outputs.version }}
-
-      - name: Git reset (package.json files changed)
-        run: |
-          git reset --hard
-
   publish-docker:
     if: needs.release.outputs.publish == 'true'
     runs-on: ubuntu-latest

Directory.Packages.props

@@ -84,7 +84,7 @@
         <PackageVersion Include="Cratis.Specifications.XUnit" Version="3.0.4" />
         <PackageVersion Include="Testcontainers" Version="4.1.0" />
         <PackageVersion Include="xunit" Version="2.9.3" />
-        <PackageVersion Include="xunit.runner.visualstudio" Version="3.0.0" />
+        <PackageVersion Include="xunit.runner.visualstudio" Version="2.8.2" />
         <PackageVersion Include="NSubstitute" Version="5.3.0" />
         <PackageVersion Include="Microsoft.NET.Test.SDK" Version="17.12.0" />
     </ItemGroup>

Integration/Orleans.InProcess/ObserverHelpers.cs

@@ -23,7 +23,7 @@ public static class ObserverHelpers
     public static async Task WaitForState(this IObserver observer, ObserverRunningState runningState, TimeSpan? timeout = default)
     {
         timeout ??= TimeSpan.FromSeconds(5);
-        var currentRunningState = ObserverRunningState.New;
+        var currentRunningState = ObserverRunningState.Unknown;
         using var cts = new CancellationTokenSource(timeout.Value);
         while (currentRunningState != runningState && !cts.IsCancellationRequested)
         {

Integration/Orleans.InProcess/for_Reactors/SomeEvent.cs

@@ -6,4 +6,4 @@
 namespace Cratis.Chronicle.Integration.Orleans.InProcess.for_Reactors;
 
 [EventType]
-public record SomeEvent(int Number);
+public record SomeEvent(int Number);

Integration/Orleans.InProcess/for_Reactors/SomeOtherEvent.cs

@@ -0,0 +1,8 @@
+// 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.Events;
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.for_Reactors;
+
+[EventType]
+public record SomeOtherEvent(int Number);

Integration/Orleans.InProcess/for_Reactors/given/a_disconnected_reactor_observing_an_event.cs

@@ -11,7 +11,7 @@ public class a_disconnected_reactor_observing_an_event(GlobalFixture globalFixtu
     public TaskCompletionSource Tcs;
     public ReactorWithoutDelay Reactor;
     public IObserver ReactorObserver;
-    public override IEnumerable<Type> EventTypes => [typeof(SomeEvent)];
+    public override IEnumerable<Type> EventTypes => [typeof(SomeEvent), typeof(SomeOtherEvent)];
 
     protected override void ConfigureServices(IServiceCollection services)
     {

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

@@ -78,9 +78,6 @@ async Task Because()
     [Fact]
     void should_have_correct_observer_state_next_event_sequence_number() => Context.ReactorObserverState.NextEventSequenceNumber.Value.ShouldEqual(1ul);
 
-    [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_not_have_failed_partitions() => Context.FailedPartitions.ShouldBeEmpty();
 }

Integration/Orleans.InProcess/for_Reactors/when_connecting/existing/with_multiple_partitions/and_reactor_has_observed_events_previously_and_is_not_behind.cs

@@ -0,0 +1,63 @@
+// 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.Events;
+using Cratis.Chronicle.EventSequences;
+using Cratis.Chronicle.Integration.Base;
+using Cratis.Chronicle.Storage.Observation;
+using context = Cratis.Chronicle.Integration.Orleans.InProcess.for_Reactors.when_connecting.existing.with_multiple_partitions.and_reactor_has_observed_events_previously_and_is_not_behind.context;
+using ObserverRunningState = Cratis.Chronicle.Concepts.Observation.ObserverRunningState;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.for_Reactors.when_connecting.existing.with_multiple_partitions;
+
+[Collection(GlobalCollection.Name)]
+public class and_reactor_has_observed_events_previously_and_is_not_behind(context context) : Given<context>(context)
+{
+    public class context(GlobalFixture globalFixture) : given.a_disconnected_reactor_observing_an_event(globalFixture)
+    {
+        public List<EventForEventSourceId> EventsToHandle;
+        public List<EventForEventSourceId> NewEvents;
+        public EventSequenceNumber LastHandledEventSequenceNumber;
+
+        public ObserverState ReactorObserverState;
+
+        public EventSequenceNumber LastEventSequenceNumberAfterDisconnect;
+
+        async Task Establish()
+        {
+            var reactor = await EventStore.Reactors.Register<ReactorWithoutDelay>();
+            ReactorObserver = GetObserverForReactor<ReactorWithoutDelay>();
+            await ReactorObserver.WaitTillActive();
+
+            EventsToHandle = EventForEventSourceIdHelpers.CreateMultiple(i => new SomeEvent(42), 10).ToList();
+            var result = await EventStore.EventLog.AppendMany(EventsToHandle);
+            LastHandledEventSequenceNumber = result.SequenceNumbers.Last();
+
+            await ReactorObserver.WaitTillReachesEventSequenceNumber(LastHandledEventSequenceNumber);
+            reactor.Disconnect();
+
+            NewEvents = EventForEventSourceIdHelpers.CreateMultiple(i => new SomeOtherEvent(42), 10).ToList();
+            result = await EventStore.EventLog.AppendMany(NewEvents);
+            LastEventSequenceNumberAfterDisconnect = result.SequenceNumbers.Last();
+        }
+
+        async Task Because()
+        {
+            await EventStore.Reactors.Register<ReactorWithoutDelay>();
+            await ReactorObserver.WaitTillActive();
+            ReactorObserverState = await ReactorObserver.GetState();
+        }
+    }
+
+    [Fact]
+    void should_have_reactor_observer_be_in_running_state() => Context.ReactorObserverState.RunningState.ShouldEqual(ObserverRunningState.Active);
+
+    [Fact]
+    void should_not_catch_up_any_new_events_added_while_disconnected() => Context.ReactorObserverState.LastHandledEventSequenceNumber.Value.ShouldEqual(Context.LastHandledEventSequenceNumber.Value);
+
+    [Fact]
+    void should_set_correct_next_event_sequence_number() => Context.ReactorObserverState.NextEventSequenceNumber.Value.ShouldEqual(Context.LastEventSequenceNumberAfterDisconnect.Next().Value);
+
+    [Fact]
+    void should_only_process_first_events() => Context.Reactor.HandledEvents.ShouldEqual(Context.EventsToHandle.Count);
+}

Integration/Orleans.InProcess/for_Reactors/when_connecting/existing/with_single_partition/and_reactor_has_observed_events_previously_and_is_not_behind.cs

@@ -0,0 +1,63 @@
+// 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.Events;
+using Cratis.Chronicle.EventSequences;
+using Cratis.Chronicle.Integration.Base;
+using Cratis.Chronicle.Storage.Observation;
+using context = Cratis.Chronicle.Integration.Orleans.InProcess.for_Reactors.when_connecting.existing.with_single_partition.and_reactor_has_observed_events_previously_and_is_not_behind.context;
+using ObserverRunningState = Cratis.Chronicle.Concepts.Observation.ObserverRunningState;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.for_Reactors.when_connecting.existing.with_single_partition;
+
+[Collection(GlobalCollection.Name)]
+public class and_reactor_has_observed_events_previously_and_is_not_behind(context context) : Given<context>(context)
+{
+    public class context(GlobalFixture globalFixture) : given.a_disconnected_reactor_observing_an_event(globalFixture)
+    {
+        public List<EventForEventSourceId> EventsToHandle;
+        public List<EventForEventSourceId> NewEvents;
+        public EventSequenceNumber LastHandledEventSequenceNumber;
+
+        public ObserverState ReactorObserverState;
+
+        public EventSequenceNumber LastEventSequenceNumberAfterDisconnect;
+
+        async Task Establish()
+        {
+            var reactor = await EventStore.Reactors.Register<ReactorWithoutDelay>();
+            ReactorObserver = GetObserverForReactor<ReactorWithoutDelay>();
+            await ReactorObserver.WaitTillActive();
+
+            EventsToHandle = EventForEventSourceIdHelpers.CreateMultiple(i => new SomeEvent(42), 10, "Partition").ToList();
+            var result = await EventStore.EventLog.AppendMany(EventsToHandle);
+            LastHandledEventSequenceNumber = result.SequenceNumbers.Last();
+
+            await ReactorObserver.WaitTillReachesEventSequenceNumber(LastHandledEventSequenceNumber);
+            reactor.Disconnect();
+
+            NewEvents = EventForEventSourceIdHelpers.CreateMultiple(i => new SomeOtherEvent(42), 10, "Partition").ToList();
+            result = await EventStore.EventLog.AppendMany(NewEvents);
+            LastEventSequenceNumberAfterDisconnect = result.SequenceNumbers.Last();
+        }
+
+        async Task Because()
+        {
+            await EventStore.Reactors.Register<ReactorWithoutDelay>();
+            await ReactorObserver.WaitTillActive();
+            ReactorObserverState = await ReactorObserver.GetState();
+        }
+    }
+
+    [Fact]
+    void should_have_reactor_observer_be_in_running_state() => Context.ReactorObserverState.RunningState.ShouldEqual(ObserverRunningState.Active);
+
+    [Fact]
+    void should_not_catch_up_any_new_events_added_while_disconnected() => Context.ReactorObserverState.LastHandledEventSequenceNumber.Value.ShouldEqual(Context.LastHandledEventSequenceNumber.Value);
+
+    [Fact]
+    void should_set_correct_next_event_sequence_number() => Context.ReactorObserverState.NextEventSequenceNumber.Value.ShouldEqual(Context.LastEventSequenceNumberAfterDisconnect.Next().Value);
+
+    [Fact]
+    void should_only_process_first_events() => Context.Reactor.HandledEvents.ShouldEqual(Context.EventsToHandle.Count);
+}

Integration/Orleans.InProcess/for_Reactors/when_connecting/non_existent/with_multiple_partitions/and_reactor_is_registered_while_there_are_no_events_to_handle.cs

@@ -0,0 +1,47 @@
+// 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.Events;
+using Cratis.Chronicle.Integration.Base;
+using Cratis.Chronicle.Storage.Observation;
+using context = Cratis.Chronicle.Integration.Orleans.InProcess.for_Reactors.when_connecting.non_existent.with_multiple_partition.and_reactor_is_registered_while_there_are_no_events_to_handle.context;
+using ObserverRunningState = Cratis.Chronicle.Concepts.Observation.ObserverRunningState;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.for_Reactors.when_connecting.non_existent.with_multiple_partition;
+
+[Collection(GlobalCollection.Name)]
+public class and_reactor_is_registered_while_there_are_no_events_to_handle(context context) : Given<context>(context)
+{
+    public class context(GlobalFixture globalFixture) : given.a_disconnected_reactor_observing_an_event(globalFixture)
+    {
+        public ObserverState ReactorObserverState;
+        public EventSequenceNumber LastEventSequenceNumber;
+
+        async Task Establish()
+        {
+            var events = EventForEventSourceIdHelpers.CreateMultiple(i => new SomeOtherEvent(42), 10).ToList();
+            var result = await EventStore.EventLog.AppendMany(events);
+            LastEventSequenceNumber = result.SequenceNumbers.Last().Next();
+        }
+
+        async Task Because()
+        {
+            ReactorObserver = GetObserverForReactor<ReactorWithoutDelay>();
+            await EventStore.Reactors.Register<ReactorWithoutDelay>();
+            await ReactorObserver.WaitTillActive();
+            ReactorObserverState = await ReactorObserver.GetState();
+        }
+    }
+
+    [Fact]
+    void should_have_reactor_observer_be_in_running_state() => Context.ReactorObserverState.RunningState.ShouldEqual(ObserverRunningState.Active);
+
+    [Fact]
+    void should_not_catch_up_any_events() => Context.ReactorObserverState.LastHandledEventSequenceNumber.Value.ShouldEqual(Concepts.Events.EventSequenceNumber.Unavailable.Value);
+
+    [Fact]
+    void should_set_next_event_sequence_number_to_next_after_last_sequence_number_of_appended_events() => Context.ReactorObserverState.NextEventSequenceNumber.Value.ShouldEqual(Context.LastEventSequenceNumber.Value);
+
+    [Fact]
+    void should_process_no_events() => Context.Reactor.HandledEvents.ShouldEqual(0);
+}

Integration/Orleans.InProcess/for_Reactors/when_connecting/non_existent/with_single_partition/and_reactor_is_registered_while_there_are_no_events_to_handle.cs

@@ -0,0 +1,47 @@
+// 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.Events;
+using Cratis.Chronicle.Integration.Base;
+using Cratis.Chronicle.Storage.Observation;
+using context = Cratis.Chronicle.Integration.Orleans.InProcess.for_Reactors.when_connecting.non_existent.with_single_partition.and_reactor_is_registered_while_there_are_no_events_to_handle.context;
+using ObserverRunningState = Cratis.Chronicle.Concepts.Observation.ObserverRunningState;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.for_Reactors.when_connecting.non_existent.with_single_partition;
+
+[Collection(GlobalCollection.Name)]
+public class and_reactor_is_registered_while_there_are_no_events_to_handle(context context) : Given<context>(context)
+{
+    public class context(GlobalFixture globalFixture) : given.a_disconnected_reactor_observing_an_event(globalFixture)
+    {
+        public ObserverState ReactorObserverState;
+        public EventSequenceNumber LastEventSequenceNumber;
+
+        async Task Establish()
+        {
+            var events = EventForEventSourceIdHelpers.CreateMultiple(i => new SomeOtherEvent(42), 10, "Partition").ToList();
+            var result = await EventStore.EventLog.AppendMany(events);
+            LastEventSequenceNumber = result.SequenceNumbers.Last().Next();
+        }
+
+        async Task Because()
+        {
+            ReactorObserver = GetObserverForReactor<ReactorWithoutDelay>();
+            await EventStore.Reactors.Register<ReactorWithoutDelay>();
+            await ReactorObserver.WaitTillActive();
+            ReactorObserverState = await ReactorObserver.GetState();
+        }
+    }
+
+    [Fact]
+    void should_have_reactor_observer_be_in_running_state() => Context.ReactorObserverState.RunningState.ShouldEqual(ObserverRunningState.Active);
+
+    [Fact]
+    void should_not_catch_up_any_events() => Context.ReactorObserverState.LastHandledEventSequenceNumber.Value.ShouldEqual(Concepts.Events.EventSequenceNumber.Unavailable.Value);
+
+    [Fact]
+    void should_set_next_event_sequence_number_to_first() => Context.ReactorObserverState.NextEventSequenceNumber.Value.ShouldEqual(Context.LastEventSequenceNumber.Value);
+
+    [Fact]
+    void should_process_no_events() => Context.Reactor.HandledEvents.ShouldEqual(0);
+}

Integration/Orleans.InProcess/for_Reactors/when_handling_event/and_it_fails/and_needs_to_catch_up.cs

@@ -14,7 +14,6 @@
 namespace Cratis.Chronicle.Integration.Orleans.InProcess.for_Reactors.when_handling_event.and_it_fails;
 
 [Collection(GlobalCollection.Name)]
-[Trait("Category", "Output")]
 public class and_needs_to_catch_up(context context) : Given<context>(context)
 {
     public class context(GlobalFixture globalFixture) : given.a_reactor_observing_an_event_that_can_fail(globalFixture, 3)
@@ -51,13 +50,13 @@ async Task Because()
             FailedPartitionsBeforeRetry = await EventStore.WaitForThereToBeFailedPartitions(ObserverId);
             Jobs = await EventStore.WaitForThereToBeJobs(waitTime);
 
-            // Wait for the second event to have been handled
+            // Wait for the first event to be handled a second time (retry)
             await Tcs[1].Task.WaitAsync(waitTime);
 
             // Append an event to be caught up
             await EventStore.EventLog.Append(EventSourceId, Event);
 
-            // Wait for the third event to have been handled
+            // Wait for the second event to have been handled for the third time after retry
             await Tcs[2].Task.WaitAsync(waitTime);
             JobsWithCatchUp = await EventStore.WaitForThereToBeJobs(waitTime);
             JobsAfterCompleted = await EventStore.WaitForThereToBeNoJobs(waitTime);

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

@@ -78,9 +78,6 @@ async Task Because()
     [Fact]
     void should_have_correct_observer_state_next_event_sequence_number() => Context.ReducerObserverState.NextEventSequenceNumber.Value.ShouldEqual(1ul);
 
-    [Fact]
-    void should_have_correct_observer_state_next_event_sequence_number_for_event_types() => Context.ReducerObserverState.NextEventSequenceNumberForEventTypes.Value.ShouldEqual(EventSequenceNumber.Unavailable.Value);
-
     [Fact]
     void should_not_have_failed_partitions() => Context.FailedPartitions.ShouldBeEmpty();
 }