Masstransit: In-memory outbox behavior changed for batch consumers from 7.0.6 to 7.0.7

[gschuager]

Context:

We use a unit of work filter to do 2 things:

• handle domain events (which generally publish integration messages)
• commit db transaction

For this, we relied on the outbox to only publish outgoing messages once the db is updated.

This used to work correctly before 7.0.7.
With the latest verion, the outbox is ignored and the messages are published (and handled) immediately.

I put togheter a reproduction here https://github.com/gschuager/MassTransitBatchTest/tree/uow/MassTransitTest

Output log for each versions is as follows:

Looking at the pipeline configuration I see the following differences:

The mentioned UnitOfWorkFilter works correctly for non-batching consumers.
Also, messages published directly from the consumer correctly respect the outbox, the problem is only present when publishing message from the filter.

I'm not sure if this is a MT issue or there is something wrong with the way I'm hooking the UoW filter into the pipeline.

[phatboyg]

So it isn't adding the outbox? I'll have a look at the test batch test and see what's up.

[gschuager]

The outbox was previously added at 2 different places, but now it is gone from one of those.

The reproduction is in the uow branch.

[phatboyg]

That's what I wondered, there was a null configurator passed to an observer previously, and that was removed. I'm guessing that the change has put a single outbox at another point in the pipeline. Can you attach a Gist of the entire consume pipe for the receive endpoint?

[gschuager]

https://gist.github.com/gschuager/79efba2a127fa9bcf2dc2b594704d203

[phatboyg]

So, the outbox is there, and it's only on the batch. It would seem that your UoW would need to get registered after the outbox. How is your UoW filter being added? Is that in the example?

[phatboyg]

https://github.com/MassTransit/MassTransit/blob/develop/src/MassTransit/Configuration/PipeConfigurators/InMemoryOutboxConfigurationObserver.cs

I'm thinking this is the issue, but I'll try to test with your sample.

[phatboyg]

Changed files and order is correct: https://gist.github.com/phatboyg/9d814ec52e6513de06faec1cd84a6f4b

[gschuager]

I'm trying to understand to understand what did you do...

Anyway, with those changes alone, it doesn't work as expected for me.

[phatboyg]

With those changes, the pipeline looks correct:

                "consumePipe": {
                  "filters": {
                    "filterType": "dispatchPipe",
                    "outputType": "MassTransit.ConsumeContext<MassTransitTest.DoWork>",
                    "consumer": {
                      "type": "MassTransit.Pipeline.ConsumerFactories.BatchConsumer<MassTransitTest.DoWork>",
                      "consumerFactory": {
                        "source": "batch",
                        "consumerType": "MassTransit.IConsumer<MassTransitTest.DoWork>",
                        "timeLimit": "00:00:01",
                        "messageLimit": 10,
                        "consumerFactory": {
                          "source": "scope",
                          "consumerType": "MassTransitTest.DoWorkConsumer",
                          "provider": "dependencyInjection"
                        },
                        "batchCollector": {
                          "consumer": {
                            "type": "MassTransitTest.DoWorkConsumer",
                            "consumerFactory": {
                              "source": "scope",
                              "consumerType": "MassTransitTest.DoWorkConsumer",
                              "provider": "dependencyInjection"
                            },
                            "filters": [
                              {
                                "filterType": "split",
                                "consumerType": "MassTransitTest.DoWorkConsumer",
                                "filters": {
                                  "filterType": "logging"
                                }
                              },
                              {
                                "filterType": "split",
                                "messageType": "MassTransit.Batch<MassTransitTest.DoWork>",
                                "filters": {
                                  "filterType": "outbox",
                                  "type": "in-memory"
                                }
                              },
                              {
                                "filterType": "split",
                                "messageType": "MassTransit.Batch<MassTransitTest.DoWork>",
                                "filters": {
                                  "filterType": "uow"
                                }
                              }
                            ],
                            "consume": {
                              "method": "Consume(ConsumeContext<MassTransit.Batch<MassTransitTest.DoWork>> context)"
                            }
                          }
                        }
                      },
                      "consume": {
                        "method": "Consume(ConsumeContext<MassTransitTest.DoWork> context)"
                      }
                    }
                  }
                }
              }

[gschuager]

With those changes, I get the following error "GreenPipes.PayloadNotFoundException: The payload was not found: System.IServiceProvider"

(I'm still trying to understand the cause)

If I comment the body of the method UnitOfWorkConfigurationObserver.MessageConfigured to ingore the previous error, I get the following:

which indicates that the outbox is still being ignored for the domain event.

[phatboyg]

What does bus.json look like with 7.0.6?

[gschuager]

https://gist.github.com/gschuager/7c0471db509d5438eb4b3c46963cc06e

[phatboyg]

So, the problem with the 7.0.6 code is that every message has its own outbox. So, in your scenarios, like 1000 outboxes, with no connection between them. With the 7.0.7 code it is creating the outbox prior to the batch consumer, just one, since that's the only consumer that's configured.

[gschuager]

Ok, I understand that.

What I don't understand is how to configure the pipeline in 7.0.7 so that messages published in the uow filter go to the single outbox instead of being published right away (this works fine for non-batching scenarios).

[phatboyg]

Are you calling publish on the ConsumeContext<Batch<T>> or the individual consume contexts of each message? If it's the latter, that's why.

[gschuager]

BatchConsumeContext<T> which implements ConsumeContext<Batch<T>>

[phatboyg]

So it seems your UnitOfWork, taking ConsumeContext on the constructor, is getting the wrong one. And I'm wondering if it's because the UOW is starting before the batch consumer, thereby not having a service provider available at the batch consumer level. Because if you look in the consumer, it's an outbox consume context wrapped in a consumer consume context.

So if you were able to create the unit of work in the filter using the proper consume context it would resolve the issue. I can't figure out why, except that the consumer factory is where the scope is created and the consumer scope isn't created until after the unit of work is added.

[phatboyg]

PR submitted that works.

[gschuager]

I just checked your changes and I see that it still doesn't work as expected: the domain event is sent (and consumed) before the batch consumer completes:

But even if the issue can be worked around in this simple case by passing the correct context around, in a real application (like the one were I detected this) you'd have domain event handlers and their dependencies (all omitted in the reproduction) where you'd need access to the correct IPublishEndpoint: having to pass it as a parameter would break those interface (I wouldn't like to add an IPublishEndpoint to my domain event handler interface because that's an implementation detail and not all of them need it).

I'm looking at the code to see if I can find where the problem is but I'm having a hard time following some things.

[phatboyg]

Looking at this with a fresh perspective (and a night's sleep), the pipeline for a regular consumer resolved from the container using the in-memory outbox looks like this one:

"consumePipe": {
  "filters": {
    "filterType": "dispatchPipe",
    "outputType": "MassTransit.ConsumeContext<MassTransit.TestFramework.Messages.PingMessage>",
    "filters": {
      "filterType": "outbox",
      "type": "in-memory"
    },
    "consumer": {
      "type": "MassTransit.Containers.Tests.Common_Tests.ConsumeContextTestSubjects.DependentConsumer",
      "consumerFactory": {
        "source": "scope",
        "consumerType": "MassTransit.Containers.Tests.Common_Tests.ConsumeContextTestSubjects.DependentConsumer",
        "provider": "dependencyInjection"
      },
      "consume": {
        "method": "Consume(ConsumeContext<MassTransit.TestFramework.Messages.PingMessage> context)"
      }
    }
  }
}

You can see, the outbox is right before the consumer factory that resolves from the container. I'll need to look at creating unit tests for the batch consumer to ensure that batch consumer dependencies get the proper consumeContext.

[phatboyg]

Test confirmed, the pipeline isn't correct (and wasn't correct _before_ either, it just didn't get fixed properly in 7.0.7).

"consumePipe": {
  "filters": {
    "filterType": "dispatchPipe",
    "outputType": "MassTransit.ConsumeContext<MassTransit.TestFramework.Messages.PingMessage>",
    "consumer": {
      "type": "MassTransit.Pipeline.ConsumerFactories.BatchConsumer<MassTransit.TestFramework.Messages.PingMessage>",
      "consumerFactory": {
        "source": "batch",
        "consumerType": "MassTransit.IConsumer<MassTransit.TestFramework.Messages.PingMessage>",
        "timeLimit": "00:00:00.2000000",
        "messageLimit": 4,
        "consumerFactory": {
          "source": "scope",
          "consumerType": "MassTransit.Containers.Tests.Common_Tests.ConsumeContextTestSubjects.DependentBatchConsumer",
          "provider": "dependencyInjection"
        },
        "batchCollector": {
          "consumer": {
            "type": "MassTransit.Containers.Tests.Common_Tests.ConsumeContextTestSubjects.DependentBatchConsumer",
            "consumerFactory": {
              "source": "scope",
              "consumerType": "MassTransit.Containers.Tests.Common_Tests.ConsumeContextTestSubjects.DependentBatchConsumer",
              "provider": "dependencyInjection"
            },
            "filters": {
              "filterType": "split",
              "messageType": "MassTransit.Batch<MassTransit.TestFramework.Messages.PingMessage>",
              "filters": {
                "filterType": "outbox",
                "type": "in-memory"
              }
            },
            "consume": {
              "method": "Consume(ConsumeContext<MassTransit.Batch<MassTransit.TestFramework.Messages.PingMessage>> context)"
            }
          }
        }
      },
      "consume": {
        "method": "Consume(ConsumeContext<MassTransit.TestFramework.Messages.PingMessage> context)"
      }
    }
  }
}

[gschuager]

I think I understand.

I added the retry filter to see where it gets added in each case ... is this what we should be targeting?

[phatboyg]

So, not really. Your unit of work expects the container scope to be available, which means it needs to be after the consumer factory. Which sort of creates a problem, that I'm trying to understand.

I have made a change to the batch configurator, so that ConsumeContext<T> filters are added before the consumer factory resulting in the following pipeline. The unit test verifies that the dependent services are getting the outbox, instead of the undecorated ConsumeContext, IPublishEndpoint, etc.

"consumePipe": {
  "filters": {
    "filterType": "dispatchPipe",
    "outputType": "MassTransit.ConsumeContext<MassTransit.TestFramework.Messages.PingMessage>",
    "consumer": {
      "type": "MassTransit.Pipeline.ConsumerFactories.BatchConsumer<MassTransit.TestFramework.Messages.PingMessage>",
      "consumerFactory": {
        "source": "batch",
        "consumerType": "MassTransit.IConsumer<MassTransit.TestFramework.Messages.PingMessage>",
        "timeLimit": "00:00:00.2000000",
        "messageLimit": 4,
        "batchCollector": {
          "filters": {
            "filterType": "outbox",
            "type": "in-memory"
          },
          "consumer": {
            "type": "MassTransit.Containers.Tests.Common_Tests.ConsumeContextTestSubjects.DependentBatchConsumer",
            "consumerFactory": {
              "source": "scope",
              "consumerType": "MassTransit.Containers.Tests.Common_Tests.ConsumeContextTestSubjects.DependentBatchConsumer",
              "provider": "dependencyInjection"
            },
            "consume": {
              "method": "Consume(ConsumeContext<MassTransit.Batch<MassTransit.TestFramework.Messages.PingMessage>> context)"
            }
          }
        }
      },
      "consume": {
        "method": "Consume(ConsumeContext<MassTransit.TestFramework.Messages.PingMessage> context)"
      }
    }
  }
}

The issue I need to confirm is if it is _consistent_ with how the other filters perform. Also, you would need to use your UnitOfWork filter with knowledge of the consumer type in order for it to work right. That's my next challenge.

[phatboyg]

Okay, this looks right (updated to include message retry, to make sure it goes in the right place).

json"consumePipe": { "filters": { "filterType": "dispatchPipe", "outputType": "MassTransit.ConsumeContext<MassTransit.TestFramework.Messages.PingMessage>", "consumer": { "type": "MassTransit.Pipeline.ConsumerFactories.BatchConsumer<MassTransit.TestFramework.Messages.PingMessage>", "consumerFactory": { "source": "batch", "consumerType": "MassTransit.IConsumer<MassTransit.TestFramework.Messages.PingMessage>", "timeLimit": "00:00:00.2000000", "messageLimit": 4, "batchCollector": { "filters": [ { "filterType": "retry", "retry-consumeContext": { "policy": "None" } }, { "filterType": "outbox", "type": "in-memory" } ], "consumer": { "type": "MassTransit.Containers.Tests.Common_Tests.ConsumeContextTestSubjects.DependentBatchConsumer", "consumerFactory": { "source": "scope", "consumerType": "MassTransit.Containers.Tests.Common_Tests.ConsumeContextTestSubjects.DependentBatchConsumer", "provider": "dependencyInjection" }, "filters": { "filterType": "split", "consumerType": "MassTransit.Containers.Tests.Common_Tests.ConsumeContextTestSubjects.DependentBatchConsumer", "filters": { "filterType": "uow" } }, "consume": { "method": "Consume(ConsumeContext<MassTransit.Batch<MassTransit.TestFramework.Messages.PingMessage>> context)" } } } }, "consume": { "method": "Consume(ConsumeContext<MassTransit.TestFramework.Messages.PingMessage> context)" } } } }

[phatboyg]

So this is built, develop packages are available. You might give it a shot.

[gschuager]

Excellent!! It works flawlessly now.

Once again, thank you very much!