Kinetis ORM 1.13: Save Your Data and Queue a Job in One Transaction#

Most applications have this bug somewhere. A request marks an order as shipped, then queues a job that emails the customer. That is two separate writes, and either one can fail without the other:

  • The order is saved but the job never gets queued, so the customer never hears that their order shipped.

  • The job is queued but the order update rolls back, so the customer gets an email about a shipment that never happened.

The fix is to put both writes in the same database transaction. With Kinetis ORM 1.13 and the SQL queue, you can.

How it works#

The callback you pass to transaction() can now take a second argument: the transaction the ORM is using. Hand it to the SQL queue’s pushOn(), and the job is written inside that transaction, next to your entity changes:

use Kinetis\Orm\EntityManager;
use Kinetis\Orm\SessionTransaction;

$this->orm->transaction(function (EntityManager $entities, SessionTransaction $tx) use ($id): void {
    $entities->repository(Order::class)->findOrFail($id)->ship();
    $this->queue->pushOn($tx, new SendShippingNotice($id));
    $entities->flush();
});

Here $this->queue is a SqlQueue. The queue-sql guide shows how to register it.

Now the order and the job succeed or fail together:

  • If the callback finishes and the commit goes through, both are saved.

  • If anything throws, neither is.

  • If the connection drops during the commit, you get CommitNotAcknowledgedException, as before. The database may or may not have committed, and that now covers the job too.

Callbacks that only take the entity manager keep working. You don’t need to change existing code.

Before this release, the ORM kept its transaction to itself. To get the same guarantee you had to build an outbox: save the job as an entity, then push it to the queue after the transaction committed.

Things to know#

  • The queue table must be in the same database. pushOn() inserts the job into kinetis_queue_jobs through the ORM’s connection, so that database needs the table. It never switches to the queue’s own connection.

  • Only the SQL queue can do this. Redis, SQS and RabbitMQ can’t take part in a database transaction. A plain push() always writes on its own, on every backend.

  • New rows don’t have their generated IDs yet. The ORM fills in an auto-increment ID only after the transaction commits, so a job can’t carry the ID of an entity you created in the same callback. Pass an ID you already know instead, such as an existing record’s or a UUID you generate. If you need the generated ID, keep using an outbox.

  • You can’t end the transaction yourself. $tx runs queries, but commit() and rollback() throw. The ORM decides when the transaction ends, and once it has, $tx refuses to run anything.

  • Catching your own query errors is risky. If a query you run on $tx fails and you catch the exception, the ORM doesn’t know about it. MySQL and MariaDB carry on with the rest of the transaction. PostgreSQL has already aborted it, so nothing is saved.

The ORM guide covers the details in Enqueueing a job with the session.

Upgrading#

Update kinetis/orm to 1.13.0. kinetis/queue-sql 1.6.5 only updates its documentation: pushOn() already accepted a transaction, and the ORM now gives you one to pass it.

Kinetis Framework 1.23: A PHPStan Rule for Process and SAPI Globals   Kinetis RoadRunner Adapter 1.3: New RoadRunner Package Names