Race conditions in transactions with SyncIndexWithObjectChangeListener

[sneakyvv]

The issue is that during a long or nested transaction, the sendUpdateIndexMessage() is called during a postPersist event, while the data may not yet be available in the database. The subsequent processing of the sent message (Doctrine/Queue/SyncIndexWithObjectChangeProcessor.php line 87) does not find the object and will send a delete command to Elastic instead of an insert.

This happens for example during a Doctrine fixtures load, which is actually several flushes inside nested transactions, but I have also seen it happen during a simple request (with queued processing). The problem is that the message is sent before the $conn->commit() is done in the (outer-)transaction's flush call. So none of the Doctrine lifecycle events is gonna guarantee there will be no race conditions.

The solution imo is to allow the message to be delivered with a delay.

[sneakyvv]

PR => php-enqueue/enqueue-elastica-bundle#19

[aumel]

I encountered the same problem. I agree with @sneakyvv 's explanation. In PostPersist/PostUpdate, the DB transaction is not necessarily completed.

Unless I'm mistaken, there is indeed the usable PostFlush event. The PostFlush event is dispatched after the $conn->commit() (see: UnitOfWork.php).

Doctrine doc:

The postFlush event occurs when the EntityManager#flush() operation is invoked and after all actual database operations are executed successfully.

In SyncIndexWithObjectChangeListener.php, we could do this:

...
public function postUpdate(LifecycleEventArgs $args)
    {
        if ($args->getObject() instanceof $this->modelClass) {
            $this->triggers[] = ['action' => SyncProcessor::UPDATE_ACTION, 'args' => $args];
        }
    }

    public function postPersist(LifecycleEventArgs $args)
    {
        if ($args->getObject() instanceof $this->modelClass) {
            $this->triggers[] = ['action' => SyncProcessor::INSERT_ACTION, 'args' => $args];
        }
    }

    public function preRemove(LifecycleEventArgs $args)
    {
        if ($args->getObject() instanceof $this->modelClass) {
            $this->triggers[] = ['action' => SyncProcessor::REMOVE_ACTION, 'args' => $args];
        }
    }

    public function postFlush(PostFlushEventArgs $event)
    {
        if (!empty($this->triggers)) {
            foreach ($this->triggers as $trigger) {
                $this->sendUpdateIndexMessage($trigger['action'], $trigger['args']);
            }

            $this->triggers = [];
        }
    }
...

IMO, the solution with PostFlush is "safer" than the delay.

[sneakyvv]

Even the flush event cannot help during nested transactions. Every flush() is sending this event, but only the outer transaction really commits it to the database.

So, I guess this is a Doctrine issue (too?).

[aumel]

There are two problems (or contexts):

1. Non-nested transaction and the data may not yet be available in the database.
2. Nested transaction and the data may not yet be available in the database.

For the first problem, I think that enqueue-elastica should at least follow the implementation of FosElasticaBundle (see: FOS\ElasticaBundle\Doctrine\Listener) and sendUpdateIndexMessage on PostFlush.

For the second, you're right. Doctrine events cannot help during nested transactions. It does not emit a PostFlush event in this case. You will encounter also this problem with FosElasticaBundle (without enqueue). You could dispatch the PostFlush event yourself. Maybe this could be useful: doctrine/orm#7103 (comment).

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.