Transactional Produce

Use ProduceFunc to commit a business write and its message together. Use InTransaction with ProduceInTx when one transaction must produce to several streams. Why SQLStreams explains the failure window this removes.

The examples assume registered producers and application-owned orders and invoice tables. Replace their fields with your application’s schema.

Choose the verb

Transaction	Payload already exists	Payload comes from database work
SQLStreams opens it	Produce	ProduceFunc
Your InTransaction closure owns it	ProduceInTx	ProduceFuncInTx

ProduceBatch writes several messages on one stream in one transaction, returning results in argument order. Producer lists each method’s parameters and errors.

Write and produce together

The closure you pass to ProduceFunc runs inside the message’s own transaction. Run your business writes on the tx it hands you and return the payload; both commit together, or neither does.

func CreateOrder(ctx context.Context, instance *sqlstreams.ProducerInstance[OrderCreated], order Order) error {
	_, err := instance.ProduceFunc(ctx,
		func(ctx context.Context, tx sqlstreams.Tx) (*OrderCreated, error) {
			if _, err := tx.Exec(ctx,
				`INSERT INTO orders (id, total) VALUES ($1, $2)`, order.Id, order.Total); err != nil {
				return nil, err
			}
			return &OrderCreated{OrderId: order.Id}, nil
		}, nil)
	return err
}

Returning an error from the closure rolls back its statements and the message. A commit error can still leave the caller uncertain whether the transaction landed; it does not split the business write from the message. Side Effects & Retries explains safe retries after an ambiguous result.

Three details about the closure:

• tx sqlstreams.Tx runs Exec, Query, QueryRow, SendBatch, and CopyFrom — the statement surface of a pgx transaction. For anything beyond that, tx.Raw() returns the underlying pgx.Tx (an escape hatch, not the default path).
• The payload comes back from the closure, not from a captured variable — so it can be built from what your own statements returned (a generated id, a computed total).
• Nothing hands you an idempotency key. A produce mints its own unless you set ProduceOptions.IdempotencyKey, and retrying across a crash means setting it — more in Side Effects & Retries.

Consumers cannot observe the message before its transaction commits.

Prove it to yourself

1. Choose an unused order id, such as order-9, and force a rollback after inserting it:

   _, err := instance.ProduceFunc(ctx,
   	func(ctx context.Context, tx sqlstreams.Tx) (*OrderCreated, error) {
   		if _, err := tx.Exec(ctx, `INSERT INTO orders (id) VALUES ($1)`, "order-9"); err != nil {
   			return nil, err
   		}
   		return nil, errors.New("rollback on purpose")
   	}, nil)

2. Check for that order and its message; neither should exist. Resolve the stream’s id from the catalog once (each stream owns its physical tables, named by that id; say it returned 1):

   SELECT id FROM sqlstreams.stream_config WHERE name = 'orders.created';
   
   SELECT count(*) FROM orders WHERE id = 'order-9'; -- 0
   SELECT count(*) FROM sqlstreams.message_log_1
   WHERE payload ->> 'order_id' = 'order-9'; -- 0

3. Replace the final return with return &OrderCreated{OrderId: "order-9"}, nil and run it again. Check err before proceeding. Both counts should now be 1, provided OrderCreated.OrderId uses the JSON tag order_id. While the transaction is open, consumers cannot see its message.

Several streams, one commit

To produce to more than one stream atomically, client.InTransaction opens one transaction and hands your closure a Tx that every instance’s ProduceInTx accepts. One statement per stream:

err := client.InTransaction(ctx, func(ctx context.Context, tx sqlstreams.Tx) error {
	if _, err := tx.Exec(ctx, `UPDATE orders SET status = 'paid' WHERE id = $1`, orderId); err != nil {
		return err
	}

	// produce last -- see the lock note below
	if _, err := invoices.ProduceInTx(ctx, tx, &InvoicePaid{OrderId: orderId}, nil); err != nil {
		return err
	}
	_, err := receipts.ProduceInTx(ctx, tx, &ReceiptRequested{OrderId: orderId}, nil)
	return err
})

Producing several compacted keys in one transaction adds a second ordering rule. Each key’s compaction_head row stays locked until you commit, so two transactions taking the same keys in opposite order deadlock: Postgres kills one of them (40P01) and rolls its whole transaction back. InTransaction never reruns your closure, so that retry is yours to write. Producing in message-key order avoids the cycle, and ProduceBatch sorts its items the same way, so a consistent order composes across both.

Two things stay yours after the commit guarantee — side effects the transaction can’t undo, and rerunning the closure when the outcome was ambiguous. Both have their own page: Side Effects & Retries.

Build a payload inside a shared transaction

Inside InTransaction you already hold tx, so ProduceInTx usually suffices. Use ProduceFuncInTx when an existing producer function owns the write and constructs the payload from its result:

err := client.InTransaction(ctx, func(ctx context.Context, tx sqlstreams.Tx) error {
	_, err := invoices.ProduceFuncInTx(ctx, tx,
		func(ctx context.Context, tx sqlstreams.Tx) (*InvoicePaid, error) {
			var invoiceId int64
			if err := tx.QueryRow(ctx,
				`INSERT INTO invoices (order_id) VALUES ($1) RETURNING id`, orderId).Scan(&invoiceId); err != nil {
				return nil, err
			}
			return &InvoicePaid{OrderId: orderId, InvoiceId: invoiceId}, nil
		}, nil)
	return err
})