Pull clients

This page describes the three pull-inbox libraries: Go, TypeScript and C#. Each library claims the rows of one consumption: pull source, runs your handler in a transaction, and completes, retries or dead-letters the row. The README of each library holds the full text: Go, TypeScript and C#. Consume the inbox shows a worker from start to end.

What every library does

Behaviour	Rule
Claim	One transaction holds the claim and nothing else, and it commits before a handler starts.
Handler transaction	The handler receives an open transaction. The library runs the completion in that transaction, then commits.
Lost lease	A completion that updates no row means that another worker owns the row. The library rolls the handler writes back.
Renewal	The library renews the lease every third of the lease duration.
Failure	A handler error rolls the transaction back. The retry policy then retries the row later or dead-letters it.
Shutdown	A stop ends the claims at once. A handler that runs past the grace period is cancelled. The library completes nothing for it and spends no attempt, and the lease expires.
Redaction	Every log line and every last_error value passes through the redaction that QueueBox uses.
Schema	The library writes no schema. It needs the V10 schema or later, which adds the inbox headers column.
Claim token	The handler never sees the token.

Rules for a handler, in every language:

1. Write every change through the transaction that the handler receives.
2. Do not commit and do not roll back. The library owns both.
3. Fail the row with an error or an exception.
4. Honour the cancellation. It fires when the lease is lost and when a shutdown runs out of grace.
5. Deduplicate external work on the source and the idempotency key. A transaction cannot roll back a call to another system.

Message fields

Field	Meaning
id	The inbox row identifier.
source	The source name.
idempotency key	The deduplication key. The identity of the message is the source and this key together.
aggregate id	Nullable.
event type	Nullable.
payload	The JSON body.
headers	The stored headers, one string value per name. Empty when there are none.
attempt	0 on the first delivery.
correlation id	Nullable. For logs.

Go

Module	github.com/alternayte/queuebox/clients/go, package queuebox
Go	1.24 or later
Drivers	database/sql with pgx for PostgreSQL or go-mssqldb for SQL Server. The application registers the driver.
Release tag	clients/go/vX.Y.Z

go get github.com/alternayte/queuebox/clients/go

worker, err := queuebox.NewInboxWorker(db, queuebox.Options{Source: "orders", BatchSize: 10, LeaseMS: 30_000})
if err != nil {
  log.Fatal(err)
}

err = worker.Run(ctx, func(ctx context.Context, message queuebox.Message, tx *sql.Tx) error {
  var order struct {
    ID    string `json:"id"`
    Total int    `json:"total"`
  }
  if err := message.UnmarshalPayload(&order); err != nil {
    return err
  }
  _, err := tx.ExecContext(ctx, "INSERT INTO orders (id, total) VALUES ($1, $2)", order.ID, order.Total)
  return err
})

API

Name	Kind	Purpose
NewInboxWorker(db *sql.DB, options Options) (*InboxWorker, error)	function	Builds a worker. It returns an error for a setting that cannot work.
(*InboxWorker).Run(ctx, handler) error	method	Claims and handles rows until ctx is cancelled.
Handler	type	func(ctx context.Context, message Message, tx *sql.Tx) error
Message	struct	ID, Source, IdempotencyKey, AggregateID, EventType, Payload, Attempt, CorrelationID, Headers.
(Message).UnmarshalPayload(target any) error	method	Parses the payload into target.
RetryPolicy	interface	Decide(message Message, failure error) FailureAction
RetryPolicyFunc	type	A function that satisfies RetryPolicy.
NewDefaultRetryPolicy(maxAttempts, baseDelay, maxDelay, jitter)	function	The exponential policy with jitter.
RetryAfter(delay), DeadLetter()	functions	The two FailureAction values.
DefaultSchema() Schema	function	The default table and column names, to change for a mapped schema.
DialectPostgreSQL, DialectSQLServer	constants	The values of Options.Dialect.
Logger	interface	Info, Warn and Error, each with a message and a map of fields.
Sanitize, SanitizeError	functions	The redaction that the library applies.

Options

Field	Default	Meaning
Source	required	The source whose rows this worker takes.
BatchSize	10	The largest number of rows in one claim.
LeaseMS	30000	The lease duration in milliseconds.
MaxConcurrency	1	The largest number of handlers at one time.
PollInterval	1 second	The wait after a claim that returned nothing.
ShutdownGrace	30 seconds	How long a stop waits for running handlers.
Dialect	DialectPostgreSQL	The database dialect.
Schema	the default names	A *Schema for a mapped table or mapped columns.
RetryPolicy	5 attempts, 1 second base, 5 minute ceiling, 0.2 jitter	What happens to a row whose handler failed.
Logger	none	Without a logger, the library prints nothing.

On SQL Server, give each claim call a context deadline of at least 30 seconds.

TypeScript

Package	@alternayte/queuebox-inbox on npm, ESM and CJS
Runtime	Node 22 or later, Bun, or Deno
Drivers	pg for PostgreSQL or mssql for SQL Server, through an adapter. The package imports no driver.
Release tag	typescript-vX.Y.Z

npm install @alternayte/queuebox-inbox

import pg from "pg";
import { InboxWorker, fromPg } from "@alternayte/queuebox-inbox";

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
const worker = new InboxWorker(fromPg(pool), { source: "orders", batchSize: 10, leaseMs: 30_000 });

await worker.run(async (message, tx) => {
  const payload = message.payload as { id: string; total: number };
  await tx.query("INSERT INTO orders (id, total) VALUES ($1, $2)", [payload.id, payload.total]);
});

API

Name	Kind	Purpose
new InboxWorker(connection, options)	class	Builds a worker.
worker.run(handler, signal?)	method	Claims and handles rows until the signal aborts.
InboxHandler	type	(message, tx, signal) => Promise<void>
fromPg(pool)	function	The adapter for a pg pool.
fromMssql(mssql, pool)	function	The adapter for an mssql pool. It takes the mssql module itself.
InboxMessage	type	id, source, idempotencyKey, aggregateId, eventType, payload, headers, attempt, correlationId.
InboxRetryPolicy	type	(message, failure) => InboxFailureAction
defaultRetryPolicy(options?)	function	The exponential policy with jitter.
retryAfter(ms), deadLetter()	functions	The two InboxFailureAction values.
defaultSchema	value	The default table and column names.
inboxSql	function	The SQL statements for a dialect and a schema.
sanitize, sanitizeError, MAX_ERROR_LENGTH	functions and value	The redaction that the library applies.
resolveOptions	function	The options with their defaults filled in.

Options

Field	Default	Meaning
source	required	The source whose rows this worker takes.
batchSize	10	The largest number of rows in one claim.
leaseMs	30000	The lease duration in milliseconds.
maxConcurrency	1	The largest number of handlers at one time.
pollIntervalMs	1000	The wait after a claim that returned nothing.
shutdownGraceMs	30000	How long a stop waits for running handlers.
dialect	"postgresql"	"postgresql" or "sqlserver".
schema	defaultSchema	The names for a mapped table or mapped columns.
retryPolicy	defaultRetryPolicy()	5 attempts, 1000 ms base, 300000 ms ceiling, 0.2 jitter.
logger	none	Without a logger, the library prints nothing.

On SQL Server, set the mssql request timeout to at least 30 seconds. Its default of 15 seconds is too short.

C#

Packages	QueueBox.Inbox, and QueueBox.Inbox.DependencyInjection for a hosted service, on NuGet
Framework	.NET 8.0 or later
Drivers	Npgsql for PostgreSQL or Microsoft.Data.SqlClient for SQL Server, through System.Data.Common. The package depends on no driver.
Release tag	csharp-vX.Y.Z

dotnet add package QueueBox.Inbox

await using var dataSource = NpgsqlDataSource.Create(connectionString);

var worker = new InboxWorker(
    InboxConnections.From(dataSource),
    new InboxOptions { Source = "orders", BatchSize = 10, LeaseMs = 30_000 });

await worker.RunAsync(async (message, transaction, cancellationToken) =>
{
    await using var command = transaction.CreateCommand();
    command.CommandText = "INSERT INTO orders (id, total) VALUES (@id, @total)";
    command
        .WithParameter("@id", message.Payload.GetProperty("id").GetString())
        .WithParameter("@total", message.Payload.GetProperty("total").GetInt32());
    await command.ExecuteNonQueryAsync(cancellationToken);
});

API

Name	Kind	Purpose
InboxWorker(connections, options, logger?)	class	Builds a worker. logger is an ILogger<InboxWorker>.
InboxWorker.RunAsync(handler, cancellationToken)	method	Claims and handles rows until the token fires.
InboxHandler	delegate	(InboxMessage message, DbTransaction transaction, CancellationToken token) => Task
InboxConnections.From(DbDataSource)	method	A connection source for a data source, for example Npgsql.
InboxConnections.From(DbProviderFactory, string)	method	A connection source for a factory and a connection string, for example SqlClientFactory.Instance.
InboxConnections.From(Func<CancellationToken, ValueTask<DbConnection>>)	method	A connection source for your own open function.
IInboxConnectionSource	interface	The connection source that a worker uses.
InboxMessage	class	Id, Source, IdempotencyKey, AggregateId, EventType, Payload as JsonElement, Attempt, CorrelationId, Headers.
IInboxRetryPolicy	interface	InboxFailureAction Decide(InboxMessage message, Exception failure)
DefaultRetryPolicy(maxAttempts, baseDelay, maxDelay, jitter)	class	The exponential policy with jitter.
InboxFailureAction.Retry(TimeSpan), InboxFailureAction.DeadLetter()	methods	The two failure actions.
InboxSchema.Default	record	The default names. Change them with a with expression.
SqlDialect	enum	PostgreSql or SqlServer.
transaction.CreateCommand(), command.WithParameter(name, value)	extension methods	Helpers for a command on the handler transaction.
services.AddQueueBoxInbox(name, options, handler, connections?)	extension method	QueueBox.Inbox.DependencyInjection. Registers a named worker as a hosted service.
InboxDbContextFactory.CreateOn(transaction, build)	method	QueueBox.Inbox.DependencyInjection. Builds an Entity Framework Core context on the handler transaction.

Options

Property	Default	Meaning
Source	required	The source whose rows this worker takes.
BatchSize	10	The largest number of rows in one claim.
LeaseMs	30000	The lease duration in milliseconds.
MaxConcurrency	1	The largest number of handlers at one time.
PollInterval	1 second	The wait after a claim that returned nothing.
ShutdownGrace	30 seconds	How long a stop waits for running handlers.
Dialect	PostgreSql	PostgreSql or SqlServer.
Schema	InboxSchema.Default	The names for a mapped table or mapped columns.
RetryPolicy	DefaultRetryPolicy	5 attempts, 1 second base, 5 minute ceiling, 0.2 jitter.

On SQL Server, keep the Microsoft.Data.SqlClient command timeout at 30 seconds or more. The default is 30 seconds. Entity Framework Core opens its own connection by default. Build the context through InboxDbContextFactory.CreateOn, so the application write and the completion share one transaction.

The retry policy

The default policy of each library retries while attempt is below the ceiling of 5, and dead-letters after that. The delay is the base delay times 2^attempt, at most the ceiling delay, with a random jitter of 20 percent. The row starts at attempt 0, so the default allows six deliveries. Write your own policy when an error must never retry, for example a payload that does not parse.

The SQL contract

A language without a library can use the SQL statements in examples/pull/sql directly. The pull example README holds the full rules.

File	Parameters	Effect
claim.sql	:source, :batch, :lease_ms, :cand_limit	Claims up to batch rows of the source, one per aggregate. Returns the rows with their claim token.
renew.sql	:id, :token, :lease_ms	Extends the lease.
complete.sql	:id, :token	Marks the row processed. Run it in the transaction of the business writes.
retry.sql	:id, :token, :delay_ms, :error	Returns the row to pending after the delay, and raises attempt by one.
dead.sql	:id, :token, :error	Marks the row dead.

• Bind every value as a parameter. Do not build the SQL text from values.
• Compute cand_limit as LEAST(GREATEST(3 * batch, 50), 500), and bind it on every call.
• Renew, complete, retry and dead-letter must each update exactly one row. Zero rows means that the claim was lost. Stop the work.
• Remove secrets from error and cut its length before you bind it.

On SQL Server, the claim takes an application lock per source with a 10 second lock timeout. A lock failure raises Msg 51000. Treat it as a transient error, and wait before the next claim. Set the driver timeout to at least 30 seconds, so the server raises Msg 51000 before the driver aborts the call. An aborted call leaves the lock held until the connection resets. A SQL Server claim needs READ COMMITTED isolation. It also works with READ_COMMITTED_SNAPSHOT ON.