Handling concurrency

[brendt]

After fixing #170, we need to think about better ways of handling concurrency. Here's the problem briefly described:

#170 fixes an issue where two instances of the same aggregate root could persist different events at the same time. Imagine two HTTP requests coming in, both triggering an event on the same AR instance (that is, one with the same UUID). If both events are recorded simultaneously, it could result in a corrupted state, because they didn't know of each other.

The solution to prevent such invalid state is the ensure there can always only be one event per AR uuid and AR version on the database level, which is how #170 was fixed. While invalid state is prevented, there's still an error for the developer to deal with. That's what we want to improve in v5.

---

The cleanest solution is to use commands (a basic command bus is already added in v5). We could schedule commands on a queue that's only able to process one command at a time, and forces the aggregate root to be retrieved right before execution. That way we're sure always the latest version of any given AR is used.

While this solution works, it's not the most performant: if every concurrent command in the system needs to be processed by a single queue, there's a significant potential for bottlenecks.

Instead, if we'd have a mechanism in place that only allows one command to run per AR instance, we're safe as well. We could run x-amount of commands in parallel, but only have one command per AR at any given time.

This begs the question though: is it possible to achieve this behaviour in Horizon, or do we need to write our own scheduler? I want to avoid rewriting parts of Horizon if possible, so I'm looking for feedback before diving into this.

---

Edit: we've decided to only support basic command bus middlewares for now, so app(CommandBus::class)->middleware(new RetryMiddleware()); is possible, but we won't add a custom horizon queuing mechanism for now. We might add it later, but probably not in v5

[robertvansteen]

Interesting idea. But that would only work for actions on the AR that can happen async. If you have a simple action that modifies the AR during the lifecycle of the request so you can return the result immediately this won't solve the issue of the second request failing due to concurrency.

[brendt]

Exactly, that's more or less how we solved it in our application. However, if your system works with the command bus pattern, a simple RetryingCommandBus decorator or middleware could help ensure that this doesn't need to be configured for every operation.

@jdrieghe I think I see what you're getting at: you want to have a way of running commands in sync as well as opposed to what I said about running every one of them on the queue? I don't think I have any problems with that, that was mostly me thinking out loud :)

[brendt]

Yes true. It might be a sign of an AR that's too big if it doesn't matter but we haven't been able to find a good domain solution for that (yet).

@rovansteen so I don't think our design should take those cases into account. I'm not saying they aren't important but I think there are other solutions to solve that problem.

[jdrieghe]

Yep, a queue already solves this problem if you can guarantee a single worker for a single AR. The problem only shows up when you apply multiple concurrent requests synchronously.

[brendt]

That's good feedback, thanks both!

[robertvansteen]

I totally understand that this is something that you don't want to encourage as a library. So if this is something that I could solve by adding a middleware like I shared above in my own code and register it that would be good enough for me.

[brendt]

I little visualisation of what the command flow would look like

command-bus.mov

[brendt]

Some design decisions based on this:

• We'll store commands in a database table keeping their serialised data, state (pending, progress, done, failed), AR type and uuid.
• We could have a finished job look for the next one to be scheduled, but that adds some flexibility when there are no jobs in the queue and you want to start a few at once. An alternative would be a long running script that watches over commands that should be run next.
• If a command for a given AR fails, all next commands for that AR can't be processed, we can rely on horizon's retry functionality, maybe we need to come up with other/smarter recovery strategies as well.
• I think that our command bus implementation should always be async. This makes a clear distinction between running something async or not
• When persisting an AR, we'll need to do an extra check whether there are previously pending commands for that AR, if that's the case we can't allow persists.

[brendt]

Also, based on the discussions in the thread above this one:

• We need a way to still run commands synchronously, but have a retry middleware that allows us to re-retrieve outdated ARs without having to run the command on Horizon.

[iosifch]

What if my application needs to execute the commands in a synchronous way?
Would not a locking system be helpful here? A middleware should ensure that a aggregate can be modified before executing the command.

[brendt]

I think some of the comments in the above thread also addressed the same question, and I agree that some kind of middleware can be useful.

[dilab]

What if my command triggers multiple ARs, for example:

MyCommand -> Aggregate A -> ReactorA(async) -> Aggregate B

Even though MyCommand is handled in sequence, ReactorA is async, you can not guarantee its order again, now you are back to square one.

[brendt]

I don't think you're use a command to trigger multiple ARs, you'd wrap them in a process manager (which, granted, this package only has partial support for, but we might look into that after this is done)

[brendt]

And to answer to your question: MyCommand wouldn't be able to be handled in sequence, it's the same thinking I had here (this was within the context of one AR, but it also applies here):

as soon as there's one operation susceptible to concurrent operations, I'd say all other operations within that AR are also, since there's that one operation that might depend on others. So in my mind, it's either the whole AR or nothing.

[dilab]

Maybe I misunderstood your command bus with my command.

My understanding of command here is a simple Command that will call the AR. When I say multiple ARs, I meant Aggregate B is trigger by an async reactor.

[hettiger]

What about the WithoutOverlapping middleware? (https://laravel.com/docs/8.x/queues#preventing-job-overlaps)

I'm on Laravel Vapor and thought this might be a solution for me. (Queues on Vapor are NOT FIFO)
However, I'm not able to add a proper key at the moment.

Example:

class OfferLoanReactor extends Reactor implements ShouldQueue
{
    public function middleware()
    {
        return [(new WithoutOverlapping('NO IDEA HOW TO GET $event->aggregateRootUuid() HERE'))->expireAfter(180)];
    }

    public function __invoke(MoreMoneyNeeded $event)
    {
        $account = Account::where('uuid', $event->aggregateRootUuid())->first();

        Mail::to($account->user)->send(new LoanProposalMail());
    }
}

Maybe a generic middleware() implementation could even become part of Reactor so users of the package just don't have to bother with this …?

Also please keep in mind Horizon is not available on Vapor. It would really be awesome if Vapor would be supported.

[hettiger]

Just to give some more context about Amazon SQS / Laravel Vapor Queue behavior:

A standard queue makes a best effort to preserve the order of messages

⇒ order is not guaranteed

and you might get that message copy again

⇒ jobs should be idempotent

See: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/standard-queues.html

Will I run into trouble using spatie/laravel-event-sourcing on Vapor with that behavior right now?
I can set a queue-concurrency: 1 to prevent concurrency issues but what about the order and the idempotency requirement?
I guess in the current state Vapor is not an option?

AWS SQS supports FIFO queues but Laravel Vapor does not. Also Laravel Vapor does not support other queue drivers. You must use AWS SQS with standard queues right now.

[brendt]

We haven't looked into supporting Vapor yet and currently have no plans to do so.

If you allow me to think out loud: if you're running serverless Laravel there are, I think, better alternatives than trying to get this event sourcing package working on it. You coul, for example, look at running Event Store as a separate service and communicate with it. I think a micro service setup would offer much more benefits if you're going serverless anyway, why not embrace it?

[hettiger]

Thank you brendt. I'm still researching what would be the best option for us. Probably you're right. I looked into spatie/laravel-event-sourcing specifically because I expected the best dev experience going that route. Vapor support would be awesome though I can understand that you have to set priorities.

[rmcdaniel]

I think https://github.com/temporalio/sdk-php is a great solution for this. My reactors just send signals into currently running temporal workflows.

[rmcdaniel]

I also wrote https://github.com/laravel-workflow/laravel-workflow which is the same thing as temporal but it uses Laravel queues and nothing else.

[rmcdaniel]

Here's how to do a saga with Laravel Workflow.

class TransferWorkflow extends Workflow
{
    public function execute($fromAccountId, $toAccountId, $amount)
    {
        try {
            yield ActivityStub::make(WithdrawActivity::class, $fromAccountId, $amount);
            $this->addCompensation(fn () => ActivityStub::make(DepositActivity::class, $fromAccountId, $amount));

            yield ActivityStub::make(DepositActivity::class, $toAccountId, $amount);
            $this->addCompensation(fn () => ActivityStub::make(WithdrawActivity::class, $toAccountId, $amount));
        } catch (Throwable $th) {
            yield from $this->compensate();
            throw $th;
        }
    }
}

class DepositActivity extends Activity
{
    public $tries = 3;

    public function execute($accountId, $amount)
    {
        app(CommandBus::class)->dispatch(new DepositCurrency($accountId, $amount));
    }
}

class WithdrawActivity extends Activity
{
    public $tries = 3;

    public function execute($accountId, $amount)
    {
        app(CommandBus::class)->dispatch(new WithdrawCurrency($accountId, $amount));
    }
}

If the withdrawal succeeds but the deposit fails then the initial withdrawal will be undone via a compensating deposit command of the same amount.

[sebastiandedeyne]

I recently opened a draft PR to improve on how we handle concurrency: #424