RSB ErrorAction issue when using more than one thread

[Nazariy Kotylo]

We would like to change the RhinoServiceBus ErrorAction to a custom one that will do the retry the same way as current ErrorAction does, in a way so it does not put the message to error sub-queue and additionally write the status of processed message during retries(like message retried for the 3 time, giving up after 5 retries, etc).

I've noticed the issue in original RSB ErrorAction when using transactional MSMQ queue.

Here is the unit test that reproduces the issue:

The MultipleThreadsMsmqTestBase is just a copy of MsmqTestBase but it has 6 threads.

public class MultipleThreadsMsmqTestBase : IDisposable
  {
    private const int NumberOfThreads = 6;
 
    static MultipleThreadsMsmqTestBase()
    {
      LogManager.Adapter = new ConsoleOutLoggerFactoryAdapter();
    }
 
    private readonly string subscriptionQueuePath;
    protected readonly Endpoint SubscriptionsUri;
 
    protected readonly string testQueuePath;
    protected readonly Endpoint TestQueueUri;
 
    protected readonly string testQueuePath2;
    protected readonly Endpoint TestQueueUri2;
 
    protected readonly string transactionalTestQueuePath;
    protected readonly Endpoint TransactionalTestQueueUri;
 
    protected MessageQueue queue;
    protected MessageQueue subscriptions;
    protected MessageQueue transactionalQueue;
 
    private ITransport transactionalTransport;
    private ITransport transport;
    protected readonly MessageQueue testQueue2;
    private readonly string subbscriptionQueuePath2;
 
    public MultipleThreadsMsmqTestBase()
    {
      TestQueueUri = new Uri("msmq://localhost/test_queue").ToEndpoint();
      testQueuePath = MsmqUtil.GetQueuePath(TestQueueUri).QueuePath;
 
      TestQueueUri2 = new Uri("msmq://localhost/test_queue2").ToEndpoint();
      testQueuePath2 = MsmqUtil.GetQueuePath(TestQueueUri2).QueuePath;
 
      TransactionalTestQueueUri = new Uri("msmq://localhost/transactional_test_queue").ToEndpoint();
      transactionalTestQueuePath = MsmqUtil.GetQueuePath(TransactionalTestQueueUri).QueuePath;
 
      SubscriptionsUri2 = new Uri("msmq://localhost/test_queue2;subscriptions").ToEndpoint();
      subbscriptionQueuePath2 = MsmqUtil.GetQueuePath(SubscriptionsUri2).QueuePathWithSubQueue;
 
      SubscriptionsUri = new Uri("msmq://localhost/test_queue;subscriptions").ToEndpoint();
      subscriptionQueuePath = MsmqUtil.GetQueuePath(SubscriptionsUri).QueuePathWithSubQueue;
 
      if (MessageQueue.Exists(testQueuePath) == false)
        MessageQueue.Create(testQueuePath);
 
      if (MessageQueue.Exists(testQueuePath2) == false)
        MessageQueue.Create(testQueuePath2);
 
      if (MessageQueue.Exists(transactionalTestQueuePath) == false)
        MessageQueue.Create(transactionalTestQueuePath, true);
 
      queue = new MessageQueue(testQueuePath);
      queue.Purge();
 
      using (var errQueue = new MessageQueue(testQueuePath + ";errors"))
      {
        errQueue.Purge();
      }
 
      testQueue2 = new MessageQueue(testQueuePath2);
      testQueue2.Purge();
 
      using (var errQueue2 = new MessageQueue(testQueuePath2 + ";errors"))
      {
        errQueue2.Purge();
      }
 
      transactionalQueue = new MessageQueue(transactionalTestQueuePath);
      transactionalQueue.Purge();
 
      using (var errQueue3 = new MessageQueue(transactionalTestQueuePath + ";errors"))
      {
        errQueue3.Purge();
      }
 
      using (var discardedQueue = new MessageQueue(testQueuePath + ";discarded"))
      {
        discardedQueue.Purge();
      }
 
      using (var timeoutQueue = new MessageQueue(testQueuePath + ";timeout"))
      {
        timeoutQueue.Purge();
      }
 
      subscriptions = new MessageQueue(subscriptionQueuePath)
      {
        Formatter = new XmlMessageFormatter(new[] { typeof(string) })
      };
      subscriptions.Purge();
 
      using (var subscriptions2 = new MessageQueue(subbscriptionQueuePath2))
      {
        subscriptions2.Purge();
      }
    }
 
    public Endpoint SubscriptionsUri2 { get; set; }
 
    public ITransport Transport
    {
      get
      {
        if (transport == null)
        {
          var serializer =
              new XmlMessageSerializer(
                  new DefaultReflection(),
                  new CastleServiceLocator(new WindsorContainer()));
          transport = new MsmqTransport(serializer,
                  new SubQueueStrategy(),
                  TestQueueUri.Uri, NumberOfThreads,
                  defaultTransportActions,
                  new EndpointRouter(),
    IsolationLevel.Serializable, TransactionalOptions.FigureItOut,
                  true,
                  new MsmqMessageBuilder(serializer, new CastleServiceLocator(new WindsorContainer())));
          transport.Start();
        }
        return transport;
      }
    }
 
    public ITransport TransactionalTransport
    {
      get
      {
        if (transactionalTransport == null)
        {
          var serializer =
              new XmlMessageSerializer(new DefaultReflection(),
                                       new CastleServiceLocator(new WindsorContainer()));
          transactionalTransport = new MsmqTransport(serializer,
              new SubQueueStrategy(),
              TransactionalTestQueueUri.Uri,
              NumberOfThreads,
              defaultTransportActions,
                  new EndpointRouter(),
    IsolationLevel.Serializable, TransactionalOptions.FigureItOut,
                  true,
                  new MsmqMessageBuilder(serializer, new CastleServiceLocator(new WindsorContainer())));
          transactionalTransport.Start();
        }
        return transactionalTransport;
      }
    }
 
    private static IMsmqTransportAction[] defaultTransportActions
    {
      get
      {
        var qs = new SubQueueStrategy();
        return new IMsmqTransportAction[]
                {
                    new AdministrativeAction(),
                    new ErrorAction(5, qs),
                    new ShutDownAction(),
                    new TimeoutAction(qs)
                };
      }
    }
 
    #region IDisposable Members
 
    public virtual void Dispose()
    {
      queue.Dispose();
      transactionalQueue.Dispose();
      subscriptions.Dispose();
 
      if (transport != null)
        transport.Dispose();
      if (transactionalTransport != null)
        transactionalTransport.Dispose();
    }
 
    #endregion
  }

// The unit test that reproduces the issue

public class MessageThatWillThrowException
  {
    public int Id { get; set; }
 
    public override string ToString()
    {
      return base.ToString()+Id;
    }
  }
 
  public class ErrorActionWithMultipleThreads: MultipleThreadsMsmqTestBase
  {
    [Fact]
    public void message_should_arrive_expected_number_of_times_using_transactional_queue()
    {
      //retry count is 5
 
      int count = 0;
      TransactionalTransport.MessageArrived += o =>
      {
        Console.WriteLine("MessageArrived " + o.MessageId);
        Interlocked.Increment(ref count);
        throw new InvalidOperationException();
      };
      const int numberOfMessages = 10;
      Parallel.For(1, numberOfMessages + 1, (i, state) => TransactionalTransport.Send(TransactionalTestQueueUri,
                                                                                      new object[]
                                                                                        {
                                                                                          new MessageThatWillThrowException
                                                                                            {Id = i}
                                                                                        }));
 
      Thread.Sleep(TimeSpan.FromSeconds(20));
      Assert.Equal(5 * numberOfMessages, count); //passes for non-transactional queue
    }
 
    [Fact]
    public void message_should_arrive_expected_number_of_times_using_non_transactional_queue()
    {
      //retry count is 5
 
      int count = 0;
      Transport.MessageArrived += o =>
      {
        Console.WriteLine("MessageArrived " + o.MessageId);
        Interlocked.Increment(ref count);
        throw new InvalidOperationException();
      };
      const int numberOfMessages = 10;
      Parallel.For(1, numberOfMessages + 1, (i, state) => Transport.Send(TestQueueUri,
                                                                                    new object[]
                                                                                      {
                                                                                        new MessageThatWillThrowException { Id = i }
                                                                                      }));
 
      Thread.Sleep(TimeSpan.FromSeconds(20));
      Assert.Equal(5 * numberOfMessages, count); //Fails here as message arrived is called more times then expected
    }
  }

The first test pass. The second one fails.

The reason for the fail is that MessageArrived event is called more times than expected. As a consequence I got the logs that sometimes message arrived/failed more than 5 times, etc... 

Perhaps the problem is related in a way how AbstractMsmqListener works?

Here is the log. So message is "peeked" by 6 threads simultaneously...

AbstractMsmqListener - Got message Rhino.ServiceBus.Tests.MessageThatWillThrowException7 on msmq://pc-nko/transactional_test_queue from FORMATNAME:DIRECT=OS:PC-NKO\private$\transactional_test_queue
25.10.2012 10:34:20 [DEBUG] Rhino.ServiceBus.Msmq.AbstractMsmqListener - Got message Rhino.ServiceBus.Tests.MessageThatWillThrowException7 on msmq://pc-nko/transactional_test_queue from FORMATNAME:DIRECT=OS:PC-NKO\private$\transactional_test_queue
25.10.2012 10:34:20 [DEBUG] Rhino.ServiceBus.Msmq.AbstractMsmqListener - Got message Rhino.ServiceBus.Tests.MessageThatWillThrowException7 on msmq://pc-nko/transactional_test_queue from FORMATNAME:DIRECT=OS:PC-NKO\private$\transactional_test_queue
25.10.2012 10:34:20 [DEBUG] Rhino.ServiceBus.Msmq.AbstractMsmqListener - Got message Rhino.ServiceBus.Tests.MessageThatWillThrowException7 on msmq://pc-nko/transactional_test_queue from FORMATNAME:DIRECT=OS:PC-NKO\private$\transactional_test_queue
25.10.2012 10:34:20 [DEBUG] Rhino.ServiceBus.Msmq.AbstractMsmqListener - Got message Rhino.ServiceBus.Tests.MessageThatWillThrowException7 on msmq://pc-nko/transactional_test_queue from FORMATNAME:DIRECT=OS:PC-NKO\private$\transactional_test_queue
25.10.2012 10:34:20 [DEBUG] Rhino.ServiceBus.Msmq.AbstractMsmqListener - Got message Rhino.ServiceBus.Tests.MessageThatWillThrowException7 on msmq://pc-nko/transactional_test_queue from FORMATNAME:DIRECT=OS:PC-NKO\private$\transactional_test_queue

Strange is that even if message is "peeked" up simultaneously by 6 threads for both transactional and non-transactional there is a difference is number of MessageArrived events...

Any idea why do we have different behavior or how this can be fixed?