Transactional enqueue

The classic dual-write bug: you write a business row, then enqueue a job — but the process crashes between the two, or the write rolls back after the job was queued. Now a job references data that doesn't exist (or vice-versa).

PortQueue fixes this by letting the insert join your own transaction. The job exists iff your business write commits. The NOTIFY that wakes workers fires only after commit.

The core pattern

The flow is always the same four steps: open a transaction → do your business write on it → addJob({ tx }) on the same transaction → commit. Both writes land together or not at all. If you roll back, the job insert rolls back with your business row.

import { fromKnex } from '@portqueue/core/tx';

// 1. Open your own transaction.
await knex.transaction(async (trx) => {
  // 2. Do your business write on it.
  const [order] = await trx('orders').insert({ customer: 'a@b.com', total: 4200 }).returning('*');

  // 3. Enqueue on the SAME transaction — wrap it in a PortQueue adapter.
  await pq.addJob({
    queue: 'emails',
    data: { to: order.customer },
    tx: fromKnex(trx),
  });

  // 4. Committing the transaction (returning from the callback) makes BOTH
  //    the order row and the job durable atomically. Throwing here rolls
  //    back the order AND the queued job together.
});

Because the job insert lives inside your transaction, workers are only told about it after the transaction commits — never before. There's no window where a worker can claim a job whose business row hasn't been committed yet.

Wake-up after commit is engine-specific

On PostgreSQL / PGLite the driver issues a NOTIFY immediately after your transaction commits, so LISTEN-ing workers wake and claim the job almost instantly. On engines without LISTEN/NOTIFY (SQLite, MySQL) there is no cross-process wake-up — a worker picks the job up on its next poll (limiter.pollIntervalSeconds, default 5 seconds on all drivers). The job is durable either way the moment you commit; only the latency to first claim differs.

Adapters from `@portqueue/core/tx`

An adapter exists for every supported client. Pick your tab:

import { fromKnex } from '@portqueue/core/tx';

await knex.transaction(async (trx) => {
  await trx('order').insert(order);
  await pq.addJob({ queue: 'fulfillment', data: { orderId }, tx: fromKnex(trx) });
}); // the developer's trx owns commit/rollback

import { fromPrisma } from '@portqueue/core/tx';

await prisma.$transaction(async (px) => {
  await px.order.create({ data: order });
  await pq.addJob({ queue: 'fulfillment', data: { orderId }, tx: fromPrisma(px) });
});

import { fromDrizzle } from '@portqueue/core/tx';

await db.transaction(async (tx) => {
  await tx.insert(orders).values(order);
  await pq.addJob({ queue: 'fulfillment', data: { orderId }, tx: fromDrizzle(tx) });
});

import { fromTypeorm } from '@portqueue/core/tx';

await dataSource.transaction(async (em) => {
  await em.save(Order, order);
  await pq.addJob({ queue: 'fulfillment', data: { orderId }, tx: fromTypeorm(em) });
}); // works with an EntityManager or a QueryRunner's manager

import { fromKysely } from '@portqueue/core/tx';

await db.transaction().execute(async (trx) => {
  await trx.insertInto('order').values(order).execute();
  // fromKysely wraps the transaction's raw executor (executeQuery(sql, params)).
  await pq.addJob({ queue: 'fulfillment', data: { orderId }, tx: fromKysely(trx) });
});

import { fromSequelize } from '@portqueue/core/tx';

await sequelize.transaction(async (t) => {
  await Order.create(order, { transaction: t });
  await pq.addJob({
    queue: 'fulfillment',
    data: { orderId },
    tx: fromSequelize(sequelize, t),
  });
});

import { fromMikroOrm } from '@portqueue/core/tx';

await orm.em.transactional(async (em) => {
  em.persist(new Order(order));
  await pq.addJob({ queue: 'fulfillment', data: { orderId }, tx: fromMikroOrm(em) });
});

import { fromPg } from '@portqueue/core/tx';

await client.query('BEGIN');
await client.query('INSERT INTO "order" (...) VALUES (...)');
await pq.addJob({ queue: 'fulfillment', data, tx: fromPg(client) });
await client.query('COMMIT'); // the app owns commit/rollback

import { fromOracle } from '@portqueue/core/tx';

const connection = await oracledb.getConnection();
try {
  await connection.execute('INSERT INTO "order" (...) VALUES (...)');
  await pq.addJob({ queue: 'fulfillment', data, tx: fromOracle(connection) });
  await connection.commit(); // the app owns commit/rollback
} finally {
  await connection.close();
}

import { fromBetterSqlite3 } from '@portqueue/core/tx';

// better-sqlite3 is synchronous; wrap the add in its transaction().
const tx = db.transaction((order) => {
  db.prepare('INSERT INTO "order" (id) VALUES (?)').run(order.id);
  void pq.addJob({
    queue: 'fulfillment',
    data: { orderId: order.id },
    tx: fromBetterSqlite3(db),
  });
});
tx(order);

import { fromNodeSqlite } from '@portqueue/core/tx';

db.exec('BEGIN');
db.prepare('INSERT INTO "order" (id) VALUES (?)').run(order.id);
await pq.addJob({ queue: 'fulfillment', data: { orderId: order.id }, tx: fromNodeSqlite(db) });
db.exec('COMMIT'); // the app owns commit/rollback

Adapters ship for the query builders fromKnex, fromKysely, fromDrizzle; the ORMs fromPrisma, fromTypeorm, fromSequelize, fromMikroOrm; and the raw clients fromPg, fromOracle, fromBetterSqlite3, and fromNodeSqlite.

When no tx is passed, the method runs autonomously in its own short transaction. Every job-manipulating method that writes to the database takes an optional tx — addJob, addJobs, retryJob, cancelJob, removeJob, removeJobs, and removeSchedule — so any of them can join your transaction and commit atomically with your business write. addJobs is a single insert for the whole batch, so its entries must all carry the same tx (or none at all). (Queue/instance control and createQueue are not DB mutations and take no tx.)

The core pattern​

Adapters from @portqueue/core/tx​

The core pattern

Adapters from `@portqueue/core/tx`