AdvanceTime with multiple timeouts throws exception

[ngallegos]

Symptoms

Adding another OrderBilledCall (or any new message handler that adds another timeout to the saga) and calling AdvanceTime in this test throws an exception:

System.InvalidOperationException : Failed to compare two elements in the array.

To reproduce, I added a second call to handle OrderBilled after the first call:

[Test]
public async Task TestBasicSaga()
{
    var testableSaga = new TestableSaga<ShippingPolicy, ShippingPolicyData>();

    var placeResult = await testableSaga.Handle(new OrderPlaced { OrderId = "abc" });
    var billResult = await testableSaga.Handle(new OrderBilled { OrderId = "abc" });
    var billResult2 = await testableSaga.Handle(new OrderBilled { OrderId = "abc" });
}

I know it doesn't make sense to handle order billed twice in a row like this, but this was the easiest way to reproduce the issue. My use case involves the timeout handler taking action depending on some saga data that can be updated by an event being handled multiple times during the life of the saga.

Who's affected

All tests using the TestableSaga helper class using the AdvanceTime(...) API for sagas that have multiple delayed messages and/or timeouts.

Root cause

A mistake during backporting this functionality from the master branch to the release-7.4 branch mixed up the tuple properties for an OrderBy comparison, resulting in incorrect types being compared.

[timbussmann]

Hi @ngallegos

Looking at your repro, I seem to get a different error: System.Exception : No mapped value found from message, could not look up saga data. It seems that error is caused by the fact that OrderBilled triggers a timeout that, once completed, marks the saga as completed. Since you send this message twice, 2 timeouts will be scheduled and once the second timeout message comes around, it won't find a saga.

Do I have to make further changes to reproduce the error that you're seeing?

[ngallegos]

Eesh, I'm sorry, @timbussmann , I think I got caught up in figuring out the problem and didn't realize the other places I'd modified the test. See the full file below, but here's a summary of what else I changed:

• Added a DateTimeOffset BilledOn property to the ShippingDelay timeout
• Added a DateTimeOffset BilledOn property to the ShippingPolicyData saga data
• Set the new saga property to DateTimeOffset.UtcNow in the OrderBilled handler
• Added a shipping delay message when requesting the timeout in TimeToShip with the current BilledOn date in the message
• Check that the ticks in the Data.BilledOn match the state.BilledOn ticks in the Timeout handler and only mark the saga complete if they are the same
• Added an await Task.Delay(1000) before calling the second OrderBilled handler to make sure ticks would differ by a second.

Here's the modified file:

namespace NServiceBus.Testing.Tests.Sagas
{
    using System;
    using System.Linq;
    using System.Threading.Tasks;
    using NUnit.Framework;

    [TestFixture]
    public class BasicSaga
    {
        [Test]
        public async Task TestBasicSaga()
        {
            var testableSaga = new TestableSaga<ShippingPolicy, ShippingPolicyData>();

            var placeResult = await testableSaga.Handle(new OrderPlaced { OrderId = "abc" });
            var billResult = await testableSaga.Handle(new OrderBilled { OrderId = "abc" });
            await Task.Delay(1000);
            var billResult2 = await testableSaga.Handle(new OrderBilled { OrderId = "abc" });

            Assert.That(placeResult.Completed, Is.False);
            Assert.That(billResult.Completed, Is.False);

            // Snapshots of data should still be assertable even after multiple operations have occurred.
            Assert.That(placeResult.SagaDataSnapshot.OrderId, Is.EqualTo("abc"));
            Assert.That(placeResult.SagaDataSnapshot.Placed, Is.True);
            Assert.That(placeResult.SagaDataSnapshot.Billed, Is.False);

            var noResults = await testableSaga.AdvanceTime(TimeSpan.FromMinutes(10));
            Assert.That(noResults.Length, Is.EqualTo(0));

            var timeoutResults = await testableSaga.AdvanceTime(TimeSpan.FromHours(1));

            Assert.That(timeoutResults.Length, Is.EqualTo(1));

            var shipped = timeoutResults.First().FindPublishedMessage<OrderShipped>();
            Assert.That(shipped.OrderId == "abc");
        }

        public class ShippingPolicy : NServiceBus.Saga<ShippingPolicyData>,
            IAmStartedByMessages<OrderPlaced>,
            IAmStartedByMessages<OrderBilled>,
            IHandleTimeouts<ShippingDelay>
        {
            protected override void ConfigureHowToFindSaga(SagaPropertyMapper<ShippingPolicyData> mapper)
            {
                mapper.MapSaga(saga => saga.OrderId)
                    .ToMessage<OrderPlaced>(msg => msg.OrderId)
                    .ToMessage<OrderBilled>(msg => msg.OrderId);
            }

            public Task Handle(OrderPlaced message, IMessageHandlerContext context)
            {
                Data.Placed = true;
                return TimeToShip(context);
            }

            public Task Handle(OrderBilled message, IMessageHandlerContext context)
            {
                Data.Billed = true;
                Data.BilledOn = DateTimeOffset.UtcNow;
                return TimeToShip(context);
            }

            public async Task TimeToShip(IMessageHandlerContext context)
            {
                if (Data.Placed && Data.Billed)
                {
                    await RequestTimeout<ShippingDelay>(context, TimeSpan.FromMinutes(15)
                        , new ShippingDelay { BilledOn = Data.BilledOn });
                }
            }

            public async Task Timeout(ShippingDelay state, IMessageHandlerContext context)
            {
                if (Data.BilledOn.Ticks == state.BilledOn.Ticks)
                {
                    await context.Publish(new OrderShipped { OrderId = Data.OrderId });
                    MarkAsComplete();
                }
            }
        }

        public class ShippingPolicyData : ContainSagaData
        {
            public string OrderId { get; set; }
            public bool Placed { get; set; }
            public bool Billed { get; set; }
            public DateTimeOffset BilledOn { get; set; }
        }

        public class OrderPlaced : IEvent
        {
            public string OrderId { get; set; }
        }

        public class OrderBilled : IEvent
        {
            public string OrderId { get; set; }
        }

        public class OrderShipped : IEvent
        {
            public string OrderId { get; set; }
        }

        public class ShippingDelay
        {
            public DateTimeOffset BilledOn { get; set; }
        }
    }
}

[timbussmann]

thanks for the additional details @ngallegos , I'll try to look closer into this as soon as possible.

[timbussmann]

@ngallegos I've tried to run the code that you shared but that seems to work without errors. The test is failing due to an assertion but I don't see the exception message that you've described 🤔

[ngallegos]

Strange, I just tried again, pasting the exact code above and I'm getting that error in both Visual Studio 2022 and in Jetbrains Rider:

It's happening on this line:

var timeoutResults = await testableSaga.AdvanceTime(TimeSpan.FromHours(1));

Some additional details that I wouldn't think would matter - I'm using nunit and .net 5 (though I did try upgrading the project to .net 6 and it still gave me the error).

[mikesigs]

@timbussmann
I ran into the same issue. It's rather simple to reproduce, and appears to be caused by the .OrderBy() of the storedTimeouts in the test saga. OrderBy requires an object to implement IComparable, but the OutgoingMessage type doesn't implement this interface.

    public class SomeSaga : NServiceBus.Saga<SomeSagaData>,
        IAmStartedByMessages<SomeMessage>,
        IHandleTimeouts<SomeTimeout>
    {
        public async Task Handle(SomeMessage message, IMessageHandlerContext context)
        {
            await RequestTimeout<SomeTimeout>(context, TimeSpan.FromMinutes(1));
        }

        protected override void ConfigureHowToFindSaga(SagaPropertyMapper<SomeSagaData> mapper)
        {
            mapper.ConfigureMapping<SomeMessage>(m => m.SomeId).ToSaga(d => d.SomeId);
        }

        public Task Timeout(SomeTimeout state, IMessageHandlerContext context)
        {
            return Task.CompletedTask;
        }
    }

    public class SomeMessage
    {
        public Guid SomeId { get; set; }
    }

    public class SomeTimeout { }

    public class SomeSagaData : ContainSagaData
    {
        public Guid SomeId { get; set; }
    }

    

    public class SomeSagaTests
    {
        [Fact]
        public async Task Test()
        {
            var saga = new TestableSaga<SomeSaga, SomeSagaData>();
            await saga.Handle(new SomeMessage());
            await saga.Handle(new SomeMessage());
            await saga.Handle(new SomeMessage());

            await saga.AdvanceTime(TimeSpan.FromMinutes(2)); // throws exception
        }
    }**

Here are the details of the exception:

System.InvalidOperationException
Failed to compare two elements in the array.
   at System.Collections.Generic.GenericArraySortHelper`1.Sort(T[] keys, Int32 index, Int32 length, IComparer`1 comparer)
   at System.Array.Sort[T](T[] array, Int32 index, Int32 length, IComparer`1 comparer)
   at System.Linq.EnumerableSorter`2.QuickSort(Int32[] keys, Int32 lo, Int32 hi)
   at System.Linq.EnumerableSorter`1.Sort(TElement[] elements, Int32 count)
   at System.Linq.OrderedEnumerable`1.ToArray()
   at System.Linq.Enumerable.ToArray[TSource](IEnumerable`1 source)
   at NServiceBus.Testing.TestableSaga`2.AdvanceTime(TimeSpan timeToAdvance, Func`2 provideContext) in D:\a\NServiceBus.Testing\NServiceBus.Testing\src\NServiceBus.Testing\Sagas\TestableSaga.cs:line 245
   at SomeService.Handler.Tests.Sagas.SomeSagaTests.Test() in C:\Code\ClearMeasure\some-service\src\SomeService.Handler.Tests\Sagas\SomeSaga_Should.cs:line 174
   at Xunit.Sdk.TestInvoker`1.<>c__DisplayClass48_1.<<InvokeTestMethodAsync>b__1>d.MoveNext() in C:\Dev\xunit\xunit\src\xunit.execution\Sdk\Frameworks\Runners\TestInvoker.cs:line 264
--- End of stack trace from previous location where exception was thrown ---
   at Xunit.Sdk.ExecutionTimer.AggregateAsync(Func`1 asyncAction) in C:\Dev\xunit\xunit\src\xunit.execution\Sdk\Frameworks\ExecutionTimer.cs:line 48
   at Xunit.Sdk.ExceptionAggregator.RunAsync(Func`1 code) in C:\Dev\xunit\xunit\src\xunit.core\Sdk\ExceptionAggregator.cs:line 90

System.ArgumentException
At least one object must implement IComparable.
   at System.Collections.Comparer.Compare(Object a, Object b)
   at System.Collections.Generic.ObjectComparer`1.Compare(T x, T y)
   at System.Linq.EnumerableSorter`2.CompareAnyKeys(Int32 index1, Int32 index2)
   at System.Collections.Generic.ComparisonComparer`1.Compare(T x, T y)
   at System.Collections.Generic.ArraySortHelper`1.SwapIfGreater(T[] keys, Comparison`1 comparer, Int32 a, Int32 b)
   at System.Collections.Generic.ArraySortHelper`1.IntroSort(T[] keys, Int32 lo, Int32 hi, Int32 depthLimit, Comparison`1 comparer)
   at System.Collections.Generic.GenericArraySortHelper`1.Sort(T[] keys, Int32 index, Int32 length, IComparer`1 comparer)

Just looking now and I see someone's made changes to the problematic code already, so this may be fixed. But in the code I've got v7.4.0 the code is trying to sort by the OutgoingMessage type itself, which doesn't implement IComparable.

var due = storedTimeouts.Where(t => t.Item1 <= advanceToTime).OrderBy(t => t.Item2).ToArray();

The fixed code is sorting by the timeout At time.

var due = storedTimeouts.Where(t => t.At <= advanceToTime).OrderBy(t => t.At).ToArray();

When do you anticipate the next patch being released for this package?

[timbussmann]

Thanks @ngallegos and @mikesigs , I wasn't able to reproduce the issue because I ran the repro code against master, assuming it uses the same code. From looking at the backport history, I suspect there has been an error when backporting the new functionality to the release-7.4 branch as it had to change some of the code using tuples due to missing support for older target platforms. I've raised this PR which should fix the issue: #379

When do you anticipate the next patch being released for this package?

I can't give you any reliable predictions for the release of the fix. I hope we can make it available to you as soon as possible.

[Tobias-08]

We have the same problem. As this issue (and #369) is blocking our adoption of the new saga scenario testing I'm wondering what is blocking the release of a new (patch) version since the fix (at least for #370) seems to have already been made.

[timbussmann]

This is being fixed with NServiceBus.Testing 7.4.1 which should be available on NuGet soon. See the release announcement for further details. Thanks for raising this issue!

[Tobias-08]

Thanks @timbussmann !
This fixed the issue on our side.