Documentation
¶
Overview ¶
Package postgres is the Postgres-backed queue backend. It is wire- and schema-compatible with existing deployments, so a Go router can be dropped in and drain the SAME queue_messages table that the existing producers write to (and vice versa).
Schema (created by InitSchema; matches the pre-existing layout):
CREATE TABLE queue_messages (
id TEXT NOT NULL,
queue_name TEXT NOT NULL,
message_group_id TEXT,
receipt_handle TEXT,
visible_at BIGINT NOT NULL, -- unix epoch seconds
payload TEXT NOT NULL, -- JSON-encoded common.Message
created_at BIGINT NOT NULL, -- unix epoch seconds
receive_count INTEGER DEFAULT 0,
PRIMARY KEY (queue_name, id)
);
CREATE INDEX idx_queue_visible
ON queue_messages (queue_name, visible_at, message_group_id);
Semantics:
- Claim is keyed on visible_at, NOT on receipt_handle being NULL. A claimed message becomes eligible again once its visibility window lapses, so a crashed consumer's messages are redelivered (at-least-once) without an explicit NACK.
- FIFO per message group: only the earliest visible message of each group (COALESCE(message_group_id, id)) is eligible at a time, so a NULL group behaves as a singleton keyed by the message id.
Index ¶
- func InitSchema(ctx context.Context, pool *pgxpool.Pool) error
- func InsertBatch(ctx context.Context, pool *pgxpool.Pool, rows []Row) error
- type Queue
- func (q *Queue) Ack(ctx context.Context, receipt string, _ string) error
- func (q *Queue) Counters() *queue.Metrics
- func (q *Queue) Defer(ctx context.Context, receipt string, delaySeconds *uint32) error
- func (q *Queue) Healthy() bool
- func (q *Queue) HonoursDelayedReturn() bool
- func (q *Queue) Identifier() string
- func (q *Queue) InitSchema(ctx context.Context) error
- func (q *Queue) Metrics(ctx context.Context) (*queue.Metrics, error)
- func (q *Queue) Nack(ctx context.Context, receipt string, delaySeconds *uint32) error
- func (q *Queue) Poll(ctx context.Context, maxMessages uint32) ([]common.QueuedMessage, error)
- func (q *Queue) Publish(ctx context.Context, m common.Message) (string, error)
- func (q *Queue) PublishBatch(ctx context.Context, msgs []common.Message) ([]string, error)
- func (q *Queue) Stop()
- type Row
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func InitSchema ¶
InitSchema creates the queue tables on an existing pool. Exported for the party that OWNS the database the queue lives on — the platform, whose dispatch scheduler publishes there. A router process must not call it against the platform's database: the platform owns that schema, and running DDL from a consumer build fails wherever the shared pool cannot hand out a connection.
func InsertBatch ¶
InsertBatch writes one row per Row in a single statement, so a failure never partially applies. Each row may name a different queue.
Exported so the dispatch scheduler's Postgres publisher writes through the same column mapping this backend's own publisher uses: two mappings for one table is how a producer and its consumer drift apart. ON CONFLICT DO NOTHING keeps at-least-once publish semantics — re-publishing a message still sitting on the queue is a no-op rather than a duplicate.
Types ¶
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue is the Postgres-backed queue (both consumer + publisher).
func (*Queue) Ack ¶
Ack deletes the message permanently. brokerMessageID is unused: a Postgres receipt handle embeds the row id and the row is deleted outright, so there is no at-least-once redelivery to guard against.
func (*Queue) HonoursDelayedReturn ¶
HonoursDelayedReturn is true (R5, docs/spec/router-deferral-handback.md): Nack's visible_at update really does hold the row back for delay, and the claim query's second NOT EXISTS clause (see Poll's doc comment) blocks a nacked-with-delay group head's successors from claiming ahead of it.
func (*Queue) InitSchema ¶
InitSchema creates the queue table and index (idempotent). The DDL matches the pre-existing layout exactly so it is a no-op when the table was already provisioned by the existing system.
func (*Queue) Poll ¶
Poll claims up to maxMessages eligible messages from this queue.
Eligibility = visible_at <= now AND the message is the earliest ELIGIBLE message in its group (COALESCE(message_group_id, id)). Each claimed row gets a unique receipt handle (<pollUUID>:<id>) and its visibility window is pushed out by the configured timeout. We use a correlated NOT EXISTS rather than a windowed CTE because FOR UPDATE SKIP LOCKED cannot be applied over a ROW_NUMBER() result under Postgres CTE inlining; the resulting claim set is equivalent.
Two NOT EXISTS clauses gate "earliest eligible in its group", not one:
- an earlier row that is CLAIMED (receipt_handle IS NOT NULL) does NOT block — an in-flight head's cross-poll ordering is still not enforced by the broker, same as always;
- an earlier row that was NACKED WITH A DELAY (R4, owner ruling 2026-09-17, docs/spec/router-deferral-handback.md: receipt_handle IS NULL AND visible_at > now) DOES block. Before R4 this state was treated the same as "claimed" — not visible, so not blocking — which let a returned group head's own successors overtake it on the very next poll, exactly the ordering violation an ordered group exists to prevent. A delay-bearing MediationDeferred hand-back (R1) sits in precisely this state, which is what made the gap load-bearing rather than academic.
func (*Queue) Publish ¶
Publish writes a single message. Uses ON CONFLICT DO NOTHING so a duplicate id is a no-op (at-least-once publish semantics).
func (*Queue) PublishBatch ¶
PublishBatch writes a batch of messages, all bound for this queue, in one statement.
Source Files