Write outbox rows

This guide shows how an application hands a message to QueueBox. The application runs one INSERT into the outbox table, inside the transaction of its business write. QueueBox reads the table, routes each row to a destination, and delivers it.

The outbox table lists every column, its type and its default. This guide shows the columns an application writes.

Note

app/src/test/kotlin/docs/IntegrationDocSqlTest.kt runs every SQL block on this page. The PostgreSQL blocks run against a PostgreSQL container, and the SQL Server blocks run against a SQL Server container. The running poller then delivers every inserted row to an HTTP destination.

The insert shares the business transaction

Open one transaction, write the business rows, insert the outbox row, then commit.

This shared transaction is the reason for the outbox pattern. The database commits the message and the business state together, so the two never disagree.

• If the transaction commits, the business row and the outbox row both exist. QueueBox then delivers the message at least once.
• If the transaction rolls back, neither row exists. No message announces an order that does not exist.

An application that publishes to a broker outside the transaction has no such guarantee. The commit can fail after the publish, and the message then announces an order that does not exist. The publish can fail after the commit, and the message is lost. No order of the two writes removes the failure, because they go to two systems.

Two more rules follow from this one:

• Do not open a second connection for the outbox insert. A second connection is a second transaction, and the guarantee is gone.
• Keep the transaction short. QueueBox sees the row only after the commit.

Insert a row

The examples use one business table, orders. The examples create it, because the test runs every statement. Your application has its own table already.

PostgreSQL

CREATE TABLE IF NOT EXISTS orders (
    id          UUID PRIMARY KEY,
    customer_id VARCHAR(64) NOT NULL,
    amount      NUMERIC(12, 2) NOT NULL,
    created_at  TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP
);

payload and headers are JSONB, so cast each string literal.

BEGIN;

INSERT INTO orders (id, customer_id, amount)
VALUES ('11111111-1111-1111-1111-111111111111', 'cust-42', 99.99);

INSERT INTO outbox (topic, key, payload, headers, aggregate_type)
VALUES (
    'order.created',
    'cust-42',
    '{"orderId":"11111111-1111-1111-1111-111111111111","amount":99.99}'::jsonb,
    '{"X-Tenant":"acme"}'::jsonb,
    'order'
);

COMMIT;

SQL Server

IF OBJECT_ID('orders', 'U') IS NULL
CREATE TABLE orders (
    id          UNIQUEIDENTIFIER PRIMARY KEY,
    customer_id NVARCHAR(64) NOT NULL,
    amount      DECIMAL(12, 2) NOT NULL,
    created_at  DATETIME2 NOT NULL DEFAULT GETUTCDATE()
);

payload and headers are NVARCHAR(MAX). key is a reserved word, so put it in brackets.

BEGIN TRANSACTION;

INSERT INTO orders (id, customer_id, amount)
VALUES ('11111111-1111-1111-1111-111111111111', N'cust-42', 99.99);

INSERT INTO outbox (topic, [key], payload, headers, aggregate_type)
VALUES (
    N'order.created',
    N'cust-42',
    N'{"orderId":"11111111-1111-1111-1111-111111111111","amount":99.99}',
    N'{"X-Tenant":"acme"}',
    N'order'
);

COMMIT TRANSACTION;

The insert names five columns:

Column	What to write
topic	Required. The name that routes the row, for example order.created.
payload	Required. The message body, as a JSON object.
key	Optional. The unit of order. See Key and order.
headers	Optional. A JSON object of headers that travel with the message. See Headers.
aggregate_type	Optional. The kind of business entity, for example order.

aggregate_type lets a broker destination choose its address from the row. A RabbitMQ exchange, a Kafka topic or a NATS subject can be a template such as public.orders.{{ aggregateType }}.v1. The destination can also read the column through exchangeFrom, topicFrom or subjectFrom. See Bridge a broker.

The shortest legal insert names only topic and payload. Every other column has a default or accepts null.

PostgreSQL

BEGIN;

INSERT INTO orders (id, customer_id, amount)
VALUES ('22222222-2222-2222-2222-222222222222', 'cust-7', 12.00);

INSERT INTO outbox (topic, payload)
VALUES ('order.paid', '{"orderId":"22222222-2222-2222-2222-222222222222"}'::jsonb);

COMMIT;

SQL Server

BEGIN TRANSACTION;

INSERT INTO orders (id, customer_id, amount)
VALUES ('22222222-2222-2222-2222-222222222222', N'cust-7', 12.00);

INSERT INTO outbox (topic, payload)
VALUES (N'order.paid', N'{"orderId":"22222222-2222-2222-2222-222222222222"}');

COMMIT TRANSACTION;

Columns the application must not write

QueueBox owns state, attempt, claimed_at, claim_token, lease_expires_at and last_error. The database fills sequence. Write state only with the value pending, which is the default. A row that starts in another state can stall, or it can deliver twice.

Choose a topic

QueueBox matches topic against the topicPattern of each route, in configuration order. The first route that matches wins. A topic that matches no route goes to dead.

Use a dotted, lower case topic, for example order.created. The column holds 255 characters at most. Fan out over HTTP shows how a pattern matches a topic.

Key and order

The rows of one non-empty key arrive in insert order, one row at a time. Rows with a null or empty key, and rows of different keys, have no order between them.

Use the identifier of the entity whose events must stay in order, for example the order ID or the customer ID. The two rows below share a key, so QueueBox delivers order.created before order.paid.

PostgreSQL

BEGIN;

INSERT INTO outbox (topic, key, payload, aggregate_type)
VALUES (
    'order.created',
    'order-33333333',
    '{"orderId":"33333333-3333-3333-3333-333333333333","step":1}'::jsonb,
    'order'
);

INSERT INTO outbox (topic, key, payload, aggregate_type)
VALUES (
    'order.paid',
    'order-33333333',
    '{"orderId":"33333333-3333-3333-3333-333333333333","step":2}'::jsonb,
    'order'
);

COMMIT;

SQL Server

BEGIN TRANSACTION;

INSERT INTO outbox (topic, [key], payload, aggregate_type)
VALUES (
    N'order.created',
    N'order-33333333',
    N'{"orderId":"33333333-3333-3333-3333-333333333333","step":1}',
    N'order'
);

INSERT INTO outbox (topic, [key], payload, aggregate_type)
VALUES (
    N'order.paid',
    N'order-33333333',
    N'{"orderId":"33333333-3333-3333-3333-333333333333","step":2}',
    N'order'
);

COMMIT TRANSACTION;

These rules decide what a key costs and what it gives:

• The database fills sequence on insert, and the claim orders the rows of one key by it. Two rows that one transaction writes keep their insert order.
• A row that waits for a retry holds back the later rows of its key. A dead row releases its key.
• One row of a key is in flight at a time. The throughput of one key is one publish round trip per row.
• The order holds for one writer per key. Two transactions that insert rows of one key at the same time can commit out of sequence order.

Spread a busy stream over more keys only when the consumer needs no order across them. Ordering explains the rule in full.

Headers

headers holds one JSON object. Each key is a header name, and each value is a string. The column is NOT NULL with the default '{}'.

• To send no header, omit the column. An explicit NULL makes the insert fail.
• An HTTP destination sends each row header as an HTTP header. A row header wins over a static destination header and over the authentication header of the same name.
• A broker destination sends the row headers as message headers.
• Put X-Correlation-Id in headers to follow one message through every log line.

Headers lists the headers that QueueBox adds itself.

Delay a message

Set scheduled_at to delay a message. QueueBox claims a row only when scheduled_at has passed. The default is the moment of the insert.

max_attempts sets the retry ceiling of one row. The block below delays a reminder by five seconds and gives it ten attempts.

PostgreSQL

BEGIN;

INSERT INTO outbox (topic, payload, scheduled_at, max_attempts)
VALUES (
    'order.reminder',
    '{"orderId":"22222222-2222-2222-2222-222222222222"}'::jsonb,
    CURRENT_TIMESTAMP + INTERVAL '5 seconds',
    10
);

COMMIT;

SQL Server

BEGIN TRANSACTION;

INSERT INTO outbox (topic, payload, scheduled_at, max_attempts)
VALUES (
    N'order.reminder',
    N'{"orderId":"22222222-2222-2222-2222-222222222222"}',
    DATEADD(SECOND, 5, SYSUTCDATETIME()),
    10
);

COMMIT TRANSACTION;

On SQL Server, scheduled_at is DATETIME2 and carries no time zone. QueueBox compares it with the wall clock of the host that runs QueueBox. Run every QueueBox instance in UTC, and write UTC values.

Set the retry ceiling

QueueBox compares the attempt column of a row with the max_attempts column of the same row. attempt is 0 on the first delivery, and each failed delivery raises it. The ceiling comes from the first of these that applies:

1. The max_attempts value that the application writes on the row.
2. outbox.maxAttempts, which QueueBox writes on every row it creates itself, for example a row that the inbox relay forwards. inbox.relay.maxAttempts overrides it for relayed rows.
3. The column default, 5.

Set max_attempts on the insert to give a slow destination more attempts than the rest of the system. A row that reaches its ceiling goes to dead. See Dead letters.

Write the row from application code

Any database library can write the row, because the row is one insert into one table. QueueBox publishes no library for the outbox side. Check two things in your code:

1. The insert joins the transaction of the business write.
2. payload and headers hold a JSON object. A JSON string that holds another JSON string breaks a transform and a routing key template.

Go

func createOrder(ctx context.Context, db *sql.DB, order Order) error {
  tx, err := db.BeginTx(ctx, nil)
  if err != nil {
    return err
  }
  defer tx.Rollback()

  if _, err := tx.ExecContext(ctx,
    "INSERT INTO orders (id, customer_id, amount) VALUES ($1, $2, $3)",
    order.ID, order.CustomerID, order.Amount); err != nil {
    return err
  }

  payload, err := json.Marshal(map[string]any{"orderId": order.ID, "amount": order.Amount})
  if err != nil {
    return err
  }

  if _, err := tx.ExecContext(ctx,
    `INSERT INTO outbox (topic, key, payload, headers, aggregate_type)
     VALUES ($1, $2, $3::jsonb, $4::jsonb, $5)`,
    "order.created", order.CustomerID, string(payload), `{"X-Tenant":"acme"}`, "order"); err != nil {
    return err
  }

  return tx.Commit()
}

TypeScript

import pg from "pg";

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });

export async function createOrder(order: { id: string; customerId: string; amount: number }) {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    await client.query(
      "INSERT INTO orders (id, customer_id, amount) VALUES ($1, $2, $3)",
      [order.id, order.customerId, order.amount],
    );
    await client.query(
      "INSERT INTO outbox (topic, key, payload, headers, aggregate_type) VALUES ($1, $2, $3, $4, $5)",
      [
        "order.created",
        order.customerId,
        JSON.stringify({ orderId: order.id, amount: order.amount }),
        JSON.stringify({ "X-Tenant": "acme" }),
        "order",
      ],
    );
    await client.query("COMMIT");
  } catch (error) {
    await client.query("ROLLBACK");
    throw error;
  } finally {
    client.release();
  }
}

C#

var orderId = Guid.NewGuid();

await using var transaction = await db.Database.BeginTransactionAsync(cancellationToken);

db.Orders.Add(new Order { Id = orderId, CustomerId = "cust-42", Amount = 99.99m });
db.OutboxMessages.Add(new OutboxMessage
{
    Id = Guid.NewGuid(),
    Topic = "order.created",
    Key = "cust-42",
    Payload = JsonSerializer.Serialize(new { orderId, amount = 99.99m }),
    Headers = """{"X-Tenant":"acme"}""",
    AggregateType = "order",
});

await db.SaveChangesAsync(cancellationToken);
await transaction.CommitAsync(cancellationToken);

The next section holds the OutboxMessage entity and its mapping.

The Entity Framework Core mapping

Entity Framework Core writes the properties of an entity, not the columns of a table. An entity without a Headers property or an AggregateType property never writes those columns. The default convention also does not turn AggregateType into aggregate_type, so name every column explicitly.

public class OutboxMessage
{
    public Guid Id { get; set; }
    public string Topic { get; set; } = default!;
    public string? Key { get; set; }
    public string Payload { get; set; } = default!;
    public string Headers { get; set; } = "{}";
    public string? AggregateType { get; set; }
}

public class OutboxMessageConfiguration : IEntityTypeConfiguration<OutboxMessage>
{
    public void Configure(EntityTypeBuilder<OutboxMessage> builder)
    {
        builder.ToTable("outbox");
        builder.Property(m => m.Id).HasColumnName("id");
        builder.Property(m => m.Topic).HasColumnName("topic");
        builder.Property(m => m.Key).HasColumnName("key");
        builder.Property(m => m.Payload).HasColumnName("payload").HasColumnType("jsonb");
        builder.Property(m => m.Headers).HasColumnName("headers").HasColumnType("jsonb");
        builder.Property(m => m.AggregateType).HasColumnName("aggregate_type");
    }
}

• HasColumnType("jsonb") applies to PostgreSQL only. On SQL Server, use nvarchar(max) for Payload and Headers, and nvarchar(255) for AggregateType.
• Set Headers to a JSON object string, never to null. The column is NOT NULL.
• Map AggregateType even when no destination reads it today. An unmapped property leaves the column empty for a destination that reads it later.
• QueueBox owns and migrates the outbox table. Do not generate an Entity Framework Core migration from this entity.

Use QueueBox from an EF Core app shows the SQL Server mapping, and the inbox side with InboxDbContextFactory.

Next steps

• Fan out over HTTP routes the rows to HTTP endpoints.
• Bridge a broker publishes the rows to RabbitMQ, Kafka or NATS.
• Delivery semantics states what QueueBox promises for a committed row.