Concurrency issues with Postgre and IContentManager

[Jad-EL]

Describe the bug

Hello, I'm currently facing some issues with the content manager. OrchardCore 1.5
I'm supposed to listen to RabbitMQ messages, and to do something according to the type of the message.
I created a ModularTenantEvent who starts the listener when the tenant get activated, in the "ActivatedAsync" and I just dispatch the message with MediatR to a CommandHandler, as you can see :

    private async Task DispatchMessage<T>(T message) where T : IRabbitMQMessage, INotification
    {
        var scope = await _shellHost.GetScopeAsync(_shellSettings);
        await scope.UsingServiceScopeAsync(async (scope) =>
        {
            var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
            _ = await mediator.Send(message, _cancellationTokenSource.Token);
        });
    }

In my command handlers, I use to create / update some contents in DB with IContentManager. that I inject.
Everything is working well with SQLite, But I tried to switch on PostGre and now it just doesn't work,

I have that kind of exception : A command is already in progress: insert into "Document" ("Id", "Type", "Content", "Version") values (@Id_1, @Type_1, @Content_1, @Version_1);insert into "Document" ("Id", "Type", "Content", "Version")

Firstly, I thought the problem was probably the scopes. So I decided to create an overlay to Session, making it Transient, and did the same with IContentManager.

The result is an ITransientContentManager (kind of proxy or decorator) that I can inject into my services, to ensure that connections are properly isolated.

public class TransientContentManager : ITransientContentManager
{
    private readonly IContentManager _contentManager;
    private readonly ITransientSession _transientSession;

    public TransientContentManager(ITransientSession transientSession, IServiceProvider serviceProvider)
    {
        _transientSession = transientSession ?? throw new ArgumentNullException(nameof(transientSession));
        _contentManager = new DefaultContentManager(
                contentDefinitionManager: serviceProvider.GetRequiredService<IContentDefinitionManager>(),
                contentManagerSession: serviceProvider.GetRequiredService<IContentManagerSession>(),
                handlers: serviceProvider.GetRequiredService<IEnumerable<IContentHandler>>(),
                session: _transientSession.TransientSession,
                idGenerator: serviceProvider.GetRequiredService<IContentItemIdGenerator>(),
                logger: serviceProvider.GetRequiredService<ILogger<DefaultContentManager>>(),
                clock: serviceProvider.GetRequiredService<IClock>()
            );
    }
}

But now there is still an exception, on WorkflowManager.I tried to analyze and realized that the problem came from the ContentHandler, which is triggered after each ContentManager update/publish.
The ContentHandler does a "TriggerEventAsync", and it throws a concurrency exception on this line

var haltedWorkflows = await _workflowStore.ListByActivityNameAsync(name, correlationId, isAlwaysCorrelated);

It seems that the problem is on the "ProduceAsync" of the YesSQL Store, but I'm not sure..

I'm also wondering if my way to create the shellscope and to handle my message is right.

I can give you more information if needed.

Thanks in advance !

To Reproduce

Steps to reproduce the behavior:

1. Use PostGre (with Pooling enabled)
2. Turn on workflows
3. Create a ModularTenantEvent
4. In the "Activated", create a shellscope and try to create / update some contents with ContentManager (ex : await _contentManager.CreateAsync(newContentItem, isPublished ? VersionOptions.Published : VersionOptions.Draft);

[sebastienros]

The repro steps should be good enough to repro the issue and understand what is happening.

[Jad-EL]

Hi,

My bad, I'll try to be more precise and clear.

I need to have a background task that listens to a rabbit mq server and processes incoming messages, making bdd interactions.
For example: when I receive a message of type "HelloWorld", I need to create a contentItem in the database of type HelloWorldPart, and then other processing in the chain... For each message.

In accordance with the documentation, I've implemented ModularTenantEvents to start listening to rabbitMQ when the tenant is activated (I use EasyNetQ to listen).

Each time I receive a message, I create a ShellScope, then dispatch the message with MediatR. A handler receives it and performs various operations.

I receive about 15 messages per second.

This diagram illustrates my current implementation:

It's working with Sqlite. But with Postgre. I have some concurrency exceptions. (A command is already in progress). The exception comes from this line :
var haltedWorkflows = await _workflowStore.ListByActivityNameAsync(name, correlationId, isAlwaysCorrelated);

in the "TriggerEventAsync" method of the

All my DB calls with contentManager are awaited.

First of all, I'm wondering if I'm doing this correctly. Is this the best way to start a background service that dispatches rabbitMQ messages using Orchard ?

Anyway, Here are the reproduce steps :

1. You have to use Postgre, since I'm not facing the problem with sqlite.
2. Create a service that implements IModularTenantEvents.
3. In my case, I Subscribe to RabbitMQ messages in the "OnActivatedAsync()" method, giving a callback, but I think you can bypass this step.
4. This call back creates a shell scope, to get a mediatR, and dispatch the message.
5. Implement a message handler, that will inject a contentManager.
6. In the MessageHandler, use the injected IContentManager to create a contentItem (multiple times in my case).

I'm not using the last version of Orchard, and my issue is similar to #7449

[sebastienros]

@Jad-EL I didn't mean that it was not, literally it was already self-explanatory but thank you for detailing it even more.

[sebastienros]

can you share the stack trace for when you get A command is already in progress: insert into ...

Note that a DefaultContentManager instance can't be shared, and each ISession can't be either. That is probably the issue you are getting.

[Piedone]

@Jad-EL can you reply here, please?

[github-actions]

It seems that this issue didn't really move for quite a while. Is this something you'd like to revisit any time soon or should we close? Please reply.

[github-actions]

Closing this issue because it didn't receive further feedback from the author for very long. If you think this is still relevant, feel free to reopen it with the requested details.