queue

package
v0.12.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 3, 2026 License: MIT Imports: 6 Imported by: 0

README

queue

import "github.com/kitsunium/sdk/pkg/v1/data/queue"

Package queue is the public facade for the SDK's ASYNCHRONOUS, DURABLE message queue: one process publishes work, another process picks it up later, and a crash in between loses nothing.

broker, err := queue.NewFile(queue.FileConfig{
	Dir: "/var/lib/myapp/jobs",
	Policy: queue.Policy{
		VisibilityTimeout: 30 * time.Second, // longer than the slowest handler
		MaxDeliveries:     5,                // then the dead-letter store
		RetryDelay:        2 * time.Second,
	},
})

// …in the producer, possibly in another process entirely:
msg, err := broker.Publish(ctx, payload)

// …in the consumer:
err = queue.Consume(ctx, broker, queue.ConsumerConfig{
	Handler: func(ctx context.Context, d queue.Delivery) error {
		return process(ctx, d.Message.Payload) // d.Deliveries says if this is a retry
	},
	HandlerIsIdempotent: true, // required; see below
	Parallelism:         4,
})
This is not an event bus, and the difference is the whole design

Read this before anything else in the package. pkg/v1/app/events and this package are described with the same words — publish, handler, message — and they guarantee opposite things:

                | events                     | queue (this package)
----------------|----------------------------|---------------------------
Scope           | one process                | many processes
Timing          | synchronous                | asynchronous
Goroutine       | the publisher's            | a consumer's
Transaction     | the publisher's            | its own
Durability      | none                       | the point of it
Retries / DLQ   | none                       | the point of it
If the process  | the event never happened   | the message is still
dies mid-flight |                            | there

If you want several things to happen right now, on your goroutine, inside your transaction, you want events and this package will be slower and harder for no benefit: Broker.Publish on a durable broker costs a disk flush, because that flush IS the guarantee.

The Transaction row is about where the WORK runs, and a handler always runs in its own. What NewSQL adds is that the MESSAGE can be part of yours: see "A queue in your own database" below.

Delivery is AT LEAST ONCE, and you must handle duplicates

Exactly-once delivery does not exist over a transport. What exists is at-least-once delivery plus an idempotent consumer, and any library that advertises the first is selling you the second with your half left as an exercise. This one says so instead — in three places you cannot miss:

  • Delivery.Deliveries is a field of every delivery. It counts from 1, and a value above 1 means these bytes have been handled before, by you or by a consumer in another process.
  • ConsumerConfig.HandlerIsIdempotent must be set to true. Its zero value is refused, so a handler cannot acquire the promise by omission, and `grep -rn HandlerIsIdempotent` enumerates every place in your codebase where somebody made it.
  • A message is removed at ACKNOWLEDGEMENT, never at read. That is what makes redelivery possible, and it is not negotiable: a consumer that dies after doing the work and before acknowledging it has done the work and not said so, and the queue has no way to tell that from a consumer that died before doing anything.
Ordering is NOT guaranteed, and a single retry is enough to lose it

One consumer, no failures, one durable broker: messages arrive in publication order. Change any of the three and they do not.

The one people discover in production is the retry. A message that fails is made visible again LATER, so messages published after it are delivered before it is retried. There is no per-key ordering here and no ordered mode; if two messages must be processed in order, they are one message.

What a crash actually costs

A consumer that is SIGKILLed mid-handler holds a lease it will never release. Nothing is lost: the lease lapses after Policy.VisibilityTimeout and the next consumer to look — in any process on that machine — picks the message up with Delivery.Deliveries incremented. That timeout is therefore how long a dead consumer's work sits idle, and it is the one number worth thinking about when you wire this.

A message whose consumers keep dying is dead-lettered like any other, by the same Policy.MaxDeliveries count, so a payload that crashes the process cannot loop forever taking every consumer with it.

The dead-letter store keeps the cause

After Policy.MaxDeliveries attempts the message is moved to the dead-letter store together with why: the failure's reason, its dotted-quad code, and its wire-safe Public half. The log-only Private half is NOT kept — a dead-letter store is read by whoever is investigating, often not the process or even the trust domain that failed — and Message.ID is the join key back to the log line that has it.

Read them with the DeadLetterReader capability, which every broker here implements. Reading does not remove them; evidence that a read consumes is evidence the second investigator does not get.

if reader, ok := broker.(queue.DeadLetterReader); ok {
	dead, err := reader.DeadLetters(ctx, 100)
}

What becomes of one is then a decision, and DeadLetterManager is where it is taken: ReplayDeadLetter queues it again — same ID, same payload, its count reset, so it has every attempt again — once the downstream is fixed, and DeleteDeadLetter removes it for good. An ID the store does not hold is DeadLetterNotFound.

if manager, ok := broker.(queue.DeadLetterManager); ok {
	err := manager.ReplayDeadLetter(ctx, dead[0].Message.ID)
}

A dead letter moves between brokers with DeadLetterWriter: PutDeadLetter writes one into the store as another broker recorded it — its payload, its count, its instants, its reason, cause and code — under an identifier the target mints. It is what a copy of a queue from a directory into a database needs, and nothing else does; a value no store keeps as given is InvalidDeadLetter. The live messages move the same way: MessageReader lists them — waiting and leased — without leasing one, and MessageWriter writes one as another broker recorded it, its delivery count and its instants kept; a value no broker keeps as given is InvalidMessage.

A failure no retry can fix is dead-lettered at once

A payload that does not decode will not decode on the fifth attempt either. Return DoNotRetry around the error and Consume dead-letters the message on its FIRST failure, through the broker's Rejecter, with your error as the cause the dead letter records:

var order Order
if err := json.Unmarshal(d.Message.Payload, &order); err != nil {
	return queue.DoNotRetry(err)
}

Keep it for failures that belong to the message. A downstream that is down, a lock that is held, a deadline that passed are what retries are for. The mark is recognised with errs.HasCode(err, queue.CodeNotRetryable); a broker of your own without Rejecter is nacked as before, and the message then reaches the dead-letter store after its last attempt with the same cause.

A retry when the handler says, a message for later, a publication once

Three more siblings, every broker here implementing each (ADR 0183).

A handler that knows when to try again — a downstream that answered with a Retry-After, a server that greylisted the sender — returns RetryAfter around its error, and Consume hands the message back through DelayedNacker with that delay instead of the policy's; the delivery is counted, as a nack's always is:

if resp.StatusCode == http.StatusTooManyRequests {
	return queue.RetryAfter(errThrottled, 30*time.Second)
}

Work that falls due once, later — a reminder, a trial that ends, an order cancelled if still unpaid — is published with DelayedPublisher: no consumer receives it before its instant, and it is as durable as any message. It does not recur: work that repeats on a calendar is pkg/v1/app/scheduler's, whose job may publish here.

if deferred, ok := broker.(queue.DelayedPublisher); ok {
	_, err := deferred.PublishAt(ctx, payload, trialEnds)
}

A producer that may retry a publication whose answer it never got publishes under a key with OncePublisher: within the window, a second publication under the same key publishes nothing and says so — a boolean, not an error, since it is the answer the retry was looking for. The queue still delivers at least once; this only stops the producer adding a copy.

if once, ok := broker.(queue.OncePublisher); ok {
	_, published, err := once.PublishOnce(ctx, "order-"+id+"-paid", payload, 5*time.Minute)
}
A retry delay that grows

Policy.RetryDelay is the same wait after every failure. Set Policy.MaxRetryDelay and it grows instead — RetryDelay after the first failure, doubling, never more than the ceiling — so a downstream that is down is asked less and less often:

queue.Policy{
	VisibilityTimeout: 30 * time.Second,
	MaxDeliveries:     10,
	RetryDelay:        time.Second,     // 1 s, 2 s, 4 s, …
	MaxRetryDelay:     5 * time.Minute, // … never more than five minutes
}

Zero keeps the constant delay every policy had before the field existed. A ceiling with no RetryDelay to grow from, or one below it, is refused.

Zero values are safe or refused, never inert

Policy.VisibilityTimeout and Policy.MaxDeliveries are REFUSED at zero, because each has two natural readings that are opposites and one of each pair silently destroys the guarantee — "redeliver instantly" against "never redeliver", "unlimited attempts" against "no attempts". Any value the SDK invented would be arbitrary, and a lease lifetime belongs to the work being protected.

Policy.RetryDelay and Policy.MaxMessageBytes are CLAMPED, because their zeros have one reading each and it is harmless: no extra delay, and DefaultMaxMessageBytes.

The three durations are also bounded from ABOVE by MaxDeadlineOffset, a century, and refused past it — not as a judgement about leases but because the durable brokers write every deadline as Unix nanoseconds, into a file name or a 64-bit column, which end in 2262. A deadline past that could not be read back, and the message it names would never be delivered again. The math.MaxInt64 somebody reaches for to mean "never" is 292 years, so it is refused rather than stranding the message; an extension that would reach past 2262 is refused by every broker for the same reason.

An idle consumer sleeps until there is work

Every broker here implements Waker, and Consume waits on it: a Publish, a Nack or a replay in this process wakes an idle worker at once, and a retry delay ending or a lease lapsing wakes it at that instant. ConsumerConfig .PollInterval is then an upper bound rather than a cadence — what is left for it to find is a message ANOTHER process published into a durable queue — so an idle consumer costs nothing between polls and a poll of a few seconds loses no latency inside one process:

queue.Consume(ctx, broker, queue.ConsumerConfig{
	Handler:             handle,
	HandlerIsIdempotent: true,
	PollInterval:        5 * time.Second, // bounds only a publication from another process
})

Two durable brokers over one directory — or one database and table — in one process share their wake, as they share everything else: they are one queue.

A stop hands back what it did not do, and a handler ends before its lease

When the consumer's context ends, Consume leases nothing more. The message a handler is running gets ConsumerConfig.DrainTimeout to finish; what the stop cut short, and every message of the batch never started, is handed back through Releaser — receivable at once, its delivery not counted — so a deployment no longer moves work toward the dead-letter store. Every outcome is sent past the stop, bounded by ConsumerConfig.SettleTimeout, so give the consumer a stop budget that covers both.

A handler's context ends a margin before its lease does — a tenth of it, at most SettleTimeout — with LEASE_EXPIRED as its cause, since past that instant another consumer may hold the same message. A handler that legitimately runs longer than a lease sets ConsumerConfig.AutoExtend, and its lease is renewed every third of it while it runs:

queue.Consume(ctx, broker, queue.ConsumerConfig{
	Handler:             handle,
	HandlerIsIdempotent: true,
	DrainTimeout:        10 * time.Second, // let the message in hand finish
	AutoExtend:          true,             // renew the lease while it runs
})
A queue in your own database, joined to your transaction

NewSQL keeps the queue in one table of the PostgreSQL, MySQL or SQLite database your application already writes, handed over as the [sql.Transactor] you open your own transactions with. Every call runs on the transaction its context carries — a savepoint of it — so a message published inside yours exists if and only if yours commits. That is the transactional outbox: the write and the message that announces it are never one without the other, and a process that dies between your commit and a publication loses nothing, because there is no between.

create, err := queue.SQLMigration(sql.DialectPostgres, "app__jobs", 20260930120000) // run by your Migrator
jobs, err := queue.NewSQL(queue.SQLConfig{
	Transactor: tm, Dialect: sql.DialectPostgres, Table: "app__jobs",
	Policy: queue.Policy{VisibilityTimeout: 30 * time.Second, MaxDeliveries: 5},
})
err = sql.Transact(ctx, tm, func(ctx context.Context, _ sql.Executor) error {
	if err := orders.Insert(ctx, order); err != nil {
		return err // no message either
	}
	_, err := jobs.Publish(ctx, payload) // joins this transaction
	return err
})

Idle consumers are woken once the publication's transaction commits, never for one rolled back. An idle Receive is one indexed read and takes no lock; a Receive that leases locks the rows it takes and skips those another consumer holds (FOR UPDATE SKIP LOCKED — MySQL 8.0.1, MariaDB 10.6) or, on SQLite, takes the one write lock. A failure of the database is QueueBackendFailed, with the driver's error reachable through errors.As and its text — which quotes rows — withheld; it ends Consume, as a storage failure always has, and a database fails transiently where a directory rarely does — so run a consumer over it under lifecycle's supervisor, and open SQLite with a busy timeout.

Three brokers

NewFile and NewSQL are the real ones. NewFile's state is a directory, it survives the process, and two brokers over one directory — in one process or in twenty — are one queue. NewSQL's state is a table, with the same guarantees, and the one property a directory cannot give: a publication inside your database transaction.

NewMemory is the test double. It gives your own tests the port's full semantics with no directory and no flush, which is what makes them fast enough to run on every save. It is one process and it holds nothing on a device, so a restart is a total loss: using it in production buys the asynchrony and throws away the durability, which is the exact confusion the frontier table above exists to prevent.

Index

Constants

const AckAcknowledged AckOutcome = 1

AckAcknowledged is a receipt whose message was removed for good.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.AckAcknowledged.

const AckLeaseExpired AckOutcome = 2

AckLeaseExpired is a receipt whose lease had lapsed: nothing was removed, as Broker.Ack answers LeaseExpired.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.AckLeaseExpired.

const AckNotAttempted AckOutcome = 0

AckNotAttempted is the zero AckOutcome: Acker.AckMany returned an error before it reached the receipt.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.AckNotAttempted.

const AckUnknownReceipt AckOutcome = 3

AckUnknownReceipt is a receipt the broker never issued: nothing was removed, as Broker.Ack answers UnknownReceipt.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.AckUnknownReceipt.

const CodeAcknowledged errs.Code = 136972

CodeAcknowledged is Acknowledged's code: errs.HasCode(err, CodeAcknowledged) recognises a handler's report that it acknowledged its delivery itself.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.CodeAcknowledged.

Code 0.2.23.12

const CodeDeadLetterNotFound errs.Code = 136967

CodeDeadLetterNotFound is DeadLetterNotFound's code.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.CodeDeadLetterNotFound.

Code 0.2.23.7

const CodeInvalidDeadLetter errs.Code = 136968

CodeInvalidDeadLetter is InvalidDeadLetter's code, the one sentinel ADR 0178 added, for errs.HasCode.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.CodeInvalidDeadLetter.

Code 0.2.23.8

const CodeInvalidMessage errs.Code = 136969

CodeInvalidMessage is InvalidMessage's code, for errs.HasCode.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.CodeInvalidMessage.

Code 0.2.23.9

const CodeMessageExpired errs.Code = 136971

CodeMessageExpired is MessageExpired's code: the reason and code a dead letter records for a message that outlived Policy.MaxAge.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.CodeMessageExpired.

Code 0.2.23.11

const CodeNotRetryable errs.Code = 136966

CodeNotRetryable is the code DoNotRetry adds to a failure's wrap trail: errs.HasCode(err, CodeNotRetryable) recognises the mark, where errors.Is(err, NotRetryable) sees only an origin.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.CodeNotRetryable.

Code 0.2.23.6

const CodeRetryRequested errs.Code = 136970

CodeRetryRequested is the code RetryAfter adds to a failure's wrap trail: errs.HasCode(err, CodeRetryRequested) recognises the mark.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.CodeRetryRequested.

Code 0.2.23.10

const CodeSQLQueueMisconfigured errs.Code = 210181

CodeSQLQueueMisconfigured is SQLQueueMisconfigured's code.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.CodeSQLQueueMisconfigured.

Code 0.3.53.5

const DefaultMaxMessageBytes int = 1048576

DefaultMaxMessageBytes is the payload bound applied when Policy.MaxMessageBytes is left at zero: one mebibyte.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.DefaultMaxMessageBytes.

const DefaultPollInterval time.Duration = 100000000

DefaultPollInterval is how long an idle worker waits before asking again when ConsumerConfig.PollInterval is left at zero. With a Waker broker it bounds only how late another process's publication is noticed.

Initialised with github.com/kitsunium/sdk/internal/service/data/queue.DefaultPollInterval.

const DefaultSettleTimeout time.Duration = 5000000000

DefaultSettleTimeout bounds each acknowledgement, nack, rejection or release Consume sends past a stop, when ConsumerConfig.SettleTimeout is left at zero: five seconds.

Initialised with github.com/kitsunium/sdk/internal/service/data/queue.DefaultSettleTimeout.

const MaxDeadlineOffset time.Duration = 3153600000000000000

MaxDeadlineOffset is the ceiling on Policy.VisibilityTimeout and Policy.RetryDelay: a century, refused above it rather than clamped, because a deadline past 2262 cannot be written into the durable broker's file names and read back.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.MaxDeadlineOffset.

const MaxSQLTableLen int = 63

MaxSQLTableLen is the longest table name NewSQL and SQLMigration accept, in bytes: PostgreSQL's identifier limit.

Initialised with github.com/kitsunium/sdk/internal/service/data/queue.MaxSQLTableLen.

const PriorityHigh Priority = 1

PriorityHigh is above the normal level.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.PriorityHigh.

const PriorityHighest Priority = 2

PriorityHighest is the highest of the five levels.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.PriorityHighest.

const PriorityLow Priority = -1

PriorityLow is below the normal level.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.PriorityLow.

const PriorityLowest Priority = -2

PriorityLowest is the lowest of the five levels.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.PriorityLowest.

const PriorityNormal Priority = 0

PriorityNormal is the zero Priority, the level Broker.Publish publishes at.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.PriorityNormal.

Variables

var Acknowledged *errs.Error

Acknowledged is what a Handler returns once it has acknowledged its delivery itself, inside the transaction that holds its effects: Consume then neither acknowledges nor hands the message back.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.Acknowledged.

Code 0.2.23.12 · reason ACKNOWLEDGED · public "The handler acknowledged the message itself"

var ConsumerMisconfigured *errs.Error

ConsumerMisconfigured refuses a consumer with no Handler, or one whose author has not asserted HandlerIsIdempotent.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.ConsumerMisconfigured.

Code 0.3.53.3 · reason CONSUMER_MISCONFIGURED · public "The consumer configuration is not runnable and was refused"

var DeadLetterNotFound *errs.Error

DeadLetterNotFound refuses a replay or a deletion naming a dead letter the store does not hold.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.DeadLetterNotFound.

Code 0.2.23.7 · reason DEAD_LETTER_NOT_FOUND · public "No dead letter has that identifier"

var HandlerPanicked *errs.Error

HandlerPanicked reports a recovered handler panic. The message was nacked and the consumer kept running.

There is deliberately no HandlerFailed beside it: a handler that returns an error has its error recorded in the dead letter unmodified, because a verdict wrapped around it would displace the reason whoever reads that record actually needs.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.HandlerPanicked.

Code 0.3.53.4 · reason HANDLER_PANICKED · public "The queue handler panicked and was recovered"

var InvalidBatchSize *errs.Error

InvalidBatchSize refuses a non-positive batch size.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.InvalidBatchSize.

Code 0.2.23.5 · reason INVALID_BATCH_SIZE · public "A batch size must be positive"

var InvalidDeadLetter *errs.Error

InvalidDeadLetter refuses a dead letter DeadLetterWriter cannot write as it was given; the "field" field names which value.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.InvalidDeadLetter.

Code 0.2.23.8 · reason INVALID_DEAD_LETTER · public "The dead letter cannot be written as it was given"

var InvalidMessage *errs.Error

InvalidMessage refuses a live message MessageWriter cannot write as it was given; the "field" field names which value.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.InvalidMessage.

Code 0.2.23.9 · reason INVALID_MESSAGE · public "The message cannot be written as it was given"

var LeaseExpired *errs.Error

LeaseExpired reports an acknowledgement whose lease had already lapsed. Nothing happened, and the message is elsewhere.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.LeaseExpired.

Code 0.2.23.4 · reason LEASE_EXPIRED · public "The lease on that message has expired and the acknowledgement was ignored"

var MessageExpired *errs.Error

MessageExpired is what a dead letter records of a message that outlived Policy.MaxAge before any consumer was handed it.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.MessageExpired.

Code 0.2.23.11 · reason MESSAGE_EXPIRED · public "The message expired before it could be delivered"

var MessageTooLarge *errs.Error

MessageTooLarge refuses a payload above the policy's bound, at the producer, where it can still be made smaller.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.MessageTooLarge.

Code 0.2.23.2 · reason MESSAGE_TOO_LARGE · public "The message payload exceeds the queue's configured maximum size"

var NotRetryable *errs.Error

NotRetryable is the failure no retry can fix: what a dead letter records when DoNotRetry marked an error with no public words of its own, or none at all.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.NotRetryable.

Code 0.2.23.6 · reason NOT_RETRYABLE · public "The message cannot be processed and was dead-lettered without a retry"

var QueueBackendFailed *errs.Error

QueueBackendFailed reports a refusal from a durable broker's storage: the filesystem's, or a statement the SQL broker's database did not complete.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.QueueBackendFailed.

Code 0.3.53.1 · reason QUEUE_BACKEND_FAILED · public "The queue's storage refused an operation"

var QueueDirectoryUnusable *errs.Error

QueueDirectoryUnusable refuses a queue directory that is missing, is not a directory, or is writable by accounts that must not be able to inject or drain messages.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.QueueDirectoryUnusable.

Code 0.3.53.2 · reason QUEUE_DIRECTORY_UNUSABLE · public "The queue directory cannot be used safely"

var QueueMisconfigured *errs.Error

QueueMisconfigured refuses a Policy field the broker cannot honour.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.QueueMisconfigured.

Code 0.2.23.1 · reason QUEUE_MISCONFIGURED · public "The queue policy is not one this broker can honour"

var RetryRequested *errs.Error

RetryRequested is what a dead letter records when RetryAfter marked an error with no public words of its own, or none at all, and the message ran out of attempts.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.RetryRequested.

Code 0.2.23.10 · reason RETRY_REQUESTED · public "The message was handed back to be retried after a delay"

var SQLQueueMisconfigured *errs.Error

SQLQueueMisconfigured refuses an SQLConfig no SQL broker could run: no transactor, one it cannot join a transaction of, a dialect it cannot spell, or a table name it cannot use.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.SQLQueueMisconfigured.

Code 0.3.53.5 · reason SQL_QUEUE_MISCONFIGURED · public "The SQL queue's configuration cannot be used"

var UnknownReceipt *errs.Error

UnknownReceipt refuses an acknowledgement naming a lease this queue never issued.

Initialised with github.com/kitsunium/sdk/internal/core/data/queue.UnknownReceipt.

Code 0.2.23.3 · reason UNKNOWN_RECEIPT · public "The receipt does not name a lease this queue issued"

func Consume

func Consume(ctx context.Context, broker Broker, cfg ConsumerConfig) error

Consume runs cfg.Handler against broker until ctx is done, on cfg.Parallelism goroutines it owns.

It returns nil when ctx ends — a cancelled consumer is a stopped consumer, not a failed one — and returns the broker's error when the STORAGE fails. A handler's error is never returned: it is nacked, retried, and eventually dead-lettered, which is the entire point of the domain.

A stop hands back, uncounted, every message a handler did not finish, after cfg.DrainTimeout; a handler's context ends a margin before its lease does, unless cfg.AutoExtend renews it. See the package documentation.

cfg.HandlerIsIdempotent must be true; see the package documentation.

func DoNotRetry

func DoNotRetry(cause error) error

DoNotRetry marks cause as a failure no retry can fix: returned from a Handler, it makes Consume dead-letter the message at once, with cause, instead of handing it back for another attempt. An SDK error keeps its own reason, code and public words in the dead letter; any other error, or none, is recorded as NotRetryable.

func NewFile

func NewFile(cfg FileConfig) (broker Broker, err error)

NewFile returns the durable broker: its whole state is the directory in cfg.Dir, so it survives the process that published into it and two brokers over one directory are one queue.

It refuses at construction — never at first use — a policy it cannot honour, a directory it cannot use safely, and a platform with no atomic replace or no flushable directory handle. Windows is such a platform: after the directory checks — which read its access control lists there — NewFile returns the SDK's UNSUPPORTED_PLATFORM, matched with errors.Is(err, proc.UnsupportedPlatform).

The broker holds two directory descriptors and additionally implements io.Closer, which releases them — reached by type assertion, as the session file store's is, so Broker grows no method:

if closer, ok := broker.(io.Closer); ok { defer closer.Close() }

The messages stay on disk; every call after Close fails.

func NewMemory

func NewMemory(cfg MemoryConfig) (broker Broker, err error)

NewMemory returns the in-process test double: the port's full semantics, no directory, no flush, and nothing that survives the process.

It refuses the same policies NewFile refuses, through the same guard, which is what makes it an honest double.

func NewSQL

func NewSQL(cfg SQLConfig) (broker Broker, err error)

NewSQL returns a durable broker whose queue is one table of the caller's database, created by SQLMigration. Every call runs on the transaction its context carries for cfg.Transactor, so a message published inside that transaction exists if and only if it commits. It sends no statement.

It refuses at construction — never at first use — a policy it cannot honour and a transactor, dialect or table it cannot use (SQLQueueMisconfigured). The transactor must be the SDK's, or another core/data/sql Joiner and Deferrer.

func RetryAfter

func RetryAfter(cause error, delay time.Duration) error

RetryAfter marks cause as a failure to retry no sooner than delay from now: returned from a Handler, it makes Consume hand the message back through the broker's DelayedNacker with that delay instead of the policy's retry delay, its delivery counted. A negative delay is read as zero; DoNotRetry wins over it.

func SQLMigration

func SQLMigration(dialect sql.Dialect, table string, version uint64) (sql.Migration, error)

SQLMigration returns the migration that creates the one table an SQL queue named table keeps on dialect, numbered version for the caller's own version table. Its Down drops the table, and every message in it. Its statement does nothing when the table exists. Its Checksum fingerprints Up's statements, so a version table that recorded it refuses a build whose statements differ.

func SQLPriorityMigration

func SQLPriorityMigration(dialect sql.Dialect, table string, version uint64) (sql.Migration, error)

SQLPriorityMigration returns the migration that adds the priority column and its index a broker with SQLConfig.Priority reads to a table SQLMigration created, numbered version for the caller's own version table. It does nothing when the table has the column or does not exist; its Down drops the index and the column.

func SQLTrailMigration

func SQLTrailMigration(dialect sql.Dialect, table string, version uint64) (sql.Migration, error)

SQLTrailMigration returns the migration that adds the trail column a broker with SQLConfig.Trail writes to a table an earlier SQLMigration created, numbered version for the caller's own version table. It does nothing when the table has the column or does not exist; its Down drops the column.

type AckOutcome

AckOutcome is the public alias for what Acker.AckMany did with one receipt.

type AckOutcome = corequeue.AckOutcome

AckOutcome is an alias of github.com/kitsunium/sdk/internal/core/data/queue.AckOutcome, declared as:

type AckOutcome uint8

type AckResult

AckResult is the public alias for what Acker.AckMany reports.

type AckResult = corequeue.AckResultValue

AckResult is an alias of github.com/kitsunium/sdk/internal/core/data/queue.AckResultValue, declared as:

type AckResult struct {
	// Outcomes holds the outcome of each receipt, Outcomes[i] for the i-th;
	// it is as long as the batch, even beside an error.
	Outcomes []AckOutcome
}

type Acker

Acker is the public alias for the capability of acknowledging several leased messages in one call, with one outcome per receipt. Every broker here implements it.

type Acker = corequeue.Acker

Acker is an alias of github.com/kitsunium/sdk/internal/core/data/queue.Acker, declared as:

type Acker interface {
	// AckMany acknowledges every receipt as [Broker.Ack] acknowledges one,
	// and returns one outcome per receipt, in the same order: acknowledged,
	// a lapsed lease, or a receipt the broker never issued. A refused
	// receipt is an outcome, never the call's error: the others are still
	// acknowledged.
	//
	// The error is the broker's storage failing, or ctx ending; the result
	// beside it still holds an outcome per receipt, [AckNotAttempted] for
	// the ones the broker did not reach, which are as they were before the
	// call. An empty batch acknowledges nothing and returns no error. A
	// broker that joins the transaction ctx carries (ADR 0151) acknowledges
	// in that transaction.
	AckMany(ctx context.Context, receipts []ReceiptValue) (AckResultValue, error)
}

func (Acker) AckMany
func (Acker) AckMany(ctx context.Context, receipts []ReceiptValue) (AckResultValue, error)

AckMany acknowledges every receipt as Broker.Ack acknowledges one, and returns one outcome per receipt, in the same order: acknowledged, a lapsed lease, or a receipt the broker never issued. A refused receipt is an outcome, never the call's error: the others are still acknowledged.

The error is the broker's storage failing, or ctx ending; the result beside it still holds an outcome per receipt, AckNotAttempted for the ones the broker did not reach, which are as they were before the call. An empty batch acknowledges nothing and returns no error. A broker that joins the transaction ctx carries (ADR 0151) acknowledges in that transaction.

type BatchPublisher

BatchPublisher is the public alias for the capability of publishing several messages in one call. Every broker here implements it.

type BatchPublisher = corequeue.BatchPublisher

BatchPublisher is an alias of github.com/kitsunium/sdk/internal/core/data/queue.BatchPublisher, declared as:

type BatchPublisher interface {
	// PublishMany appends every payload to the queue, in order, as
	// [Broker.Publish] appends one, and returns the messages the broker
	// minted, one per payload and in the same order.
	//
	// Every payload is checked against [PolicyValue.MaxMessageBytes] before
	// anything is written: one above it is [MessageTooLarge], and nothing is
	// published. An empty batch publishes nothing and returns no error.
	//
	// A broker that joins the transaction ctx carries (ADR 0151) publishes
	// the batch in that transaction, so it exists entirely or not at all. A
	// broker that writes message by message says so: on a failure at the
	// k-th payload, the messages it returns beside the error — the first
	// k−1 — are durable, and the rest were not published.
	PublishMany(ctx context.Context, payloads [][]byte) ([]MessageValue, error)
}

func (BatchPublisher) PublishMany
func (BatchPublisher) PublishMany(ctx context.Context, payloads [][]byte) ([]MessageValue, error)

PublishMany appends every payload to the queue, in order, as Broker.Publish appends one, and returns the messages the broker minted, one per payload and in the same order.

Every payload is checked against [PolicyValue.MaxMessageBytes] before anything is written: one above it is MessageTooLarge, and nothing is published. An empty batch publishes nothing and returns no error.

A broker that joins the transaction ctx carries (ADR 0151) publishes the batch in that transaction, so it exists entirely or not at all. A broker that writes message by message says so: on a failure at the k-th payload, the messages it returns beside the error — the first k−1 — are durable, and the rest were not published.

type Broker

Broker is the public alias for the queue contract. It is FROZEN at four methods; capabilities arrive as siblings (DeadLetterReader, LeaseExtender, Waker, Rejecter, Releaser, DelayedNacker, DelayedPublisher, OncePublisher, DeadLetterManager, DeadLetterWriter, MessageReader, MessageWriter, MessageLister, Counter, Purger, DeadLetterPager, DeadLetterBulkManager, BatchPublisher, Acker) reached by type assertion.

type Broker = corequeue.Broker

Broker is an alias of github.com/kitsunium/sdk/internal/core/data/queue.Broker, declared as:

type Broker interface {
	// Ack removes the leased message permanently. It is the only call that
	// does.
	//
	// A receipt whose lease has already lapsed is [LeaseExpired] and removes
	// NOTHING: the message is back in the queue, or in the dead-letter
	// store, and another consumer may already hold it. Reporting that is the
	// same decision core/app/lock makes for a lapsed lease — a holder that lost
	// its claim must not be able to end somebody else's turn, and must find
	// out that it lost it.
	//
	// A receipt this broker never issued, or one it cannot parse, is
	// [UnknownReceipt].
	Ack(ctx context.Context, receipt ReceiptValue) error
	// Nack reports that processing failed and hands the message back.
	//
	// The broker decides what happens next and reports it: the message
	// becomes visible again after [PolicyValue.RetryDelay] — grown by
	// [PolicyValue.MaxRetryDelay] when the policy asks for it — or, when it
	// has now been delivered [PolicyValue.MaxDeliveries] times, it is moved to
	// the dead-letter store together with cause. See [NackValue]. A failure
	// no retry can fix is not nacked but rejected ([Rejecter]).
	//
	// cause is recorded. What is kept of it is the implementation's to
	// document, and the rule this domain follows is CLAUDE.md rule 4: the
	// wire-safe Public half and the dotted-quad code travel with the dead
	// letter, the Private half stays in the failing process's log, and the
	// message identifier is the join key between them. A nil cause is
	// accepted and recorded as an unexplained failure, because a consumer
	// that knows only "this did not work" must still be able to say so.
	//
	// A lapsed or unknown receipt is refused exactly as [Broker.Ack] refuses
	// it.
	Nack(ctx context.Context, receipt ReceiptValue, cause error) (NackValue, error)
	// Publish appends payload to the queue and returns the message the
	// broker minted for it.
	//
	// It takes BYTES and not a [MessageValue] on purpose: the identifier and
	// the enqueue instant are the broker's to assign, and a struct with
	// fields the caller must leave blank is a struct that will eventually be
	// filled in.
	//
	// When it returns nil, a durable implementation has flushed the message
	// to its device: a crash one instruction later does not lose it. That
	// flush is most of what the call costs — see the domain's BENCH.md. An
	// implementation that joins the transaction ctx carries (ADR 0151) has
	// instead made the message part of that transaction: it is durable, and
	// receivable, once the transaction commits, and it never existed if the
	// transaction rolls back.
	//
	// A payload larger than [PolicyValue.MaxMessageBytes] is
	// [MessageTooLarge]; an empty payload is legitimate, because a message
	// whose whole meaning is its arrival is a normal thing to send.
	Publish(ctx context.Context, payload []byte) (MessageValue, error)
	// Receive leases up to max messages and returns them. It does NOT block
	// waiting for work: an empty queue yields an empty slice and a nil
	// error, and polling is the caller's (see internal/service/data/queue.Consume,
	// which does it for you).
	//
	// Each returned message is invisible to every other consumer until its
	// [DeliveryValue.Lease] expires. A consumer that finishes acknowledges
	// with [Broker.Ack]; a consumer that fails reports it with [Broker.Nack];
	// a consumer that DIES does neither, and the lapsed lease is what makes
	// the message reappear.
	//
	// max must be positive: [InvalidBatchSize] otherwise, rather than an
	// empty slice forever from a consumer loop that spins and processes
	// nothing.
	Receive(ctx context.Context, max int) ([]DeliveryValue, error)
}

func (Broker) Ack
func (Broker) Ack(ctx context.Context, receipt ReceiptValue) error

Ack removes the leased message permanently. It is the only call that does.

A receipt whose lease has already lapsed is LeaseExpired and removes NOTHING: the message is back in the queue, or in the dead-letter store, and another consumer may already hold it. Reporting that is the same decision core/app/lock makes for a lapsed lease — a holder that lost its claim must not be able to end somebody else's turn, and must find out that it lost it.

A receipt this broker never issued, or one it cannot parse, is UnknownReceipt.

func (Broker) Nack
func (Broker) Nack(ctx context.Context, receipt ReceiptValue, cause error) (NackValue, error)

Nack reports that processing failed and hands the message back.

The broker decides what happens next and reports it: the message becomes visible again after [PolicyValue.RetryDelay] — grown by [PolicyValue.MaxRetryDelay] when the policy asks for it — or, when it has now been delivered [PolicyValue.MaxDeliveries] times, it is moved to the dead-letter store together with cause. See [NackValue]. A failure no retry can fix is not nacked but rejected (Rejecter).

cause is recorded. What is kept of it is the implementation's to document, and the rule this domain follows is CLAUDE.md rule 4: the wire-safe Public half and the dotted-quad code travel with the dead letter, the Private half stays in the failing process's log, and the message identifier is the join key between them. A nil cause is accepted and recorded as an unexplained failure, because a consumer that knows only "this did not work" must still be able to say so.

A lapsed or unknown receipt is refused exactly as Broker.Ack refuses it.

func (Broker) Publish
func (Broker) Publish(ctx context.Context, payload []byte) (MessageValue, error)

Publish appends payload to the queue and returns the message the broker minted for it.

It takes BYTES and not a [MessageValue] on purpose: the identifier and the enqueue instant are the broker's to assign, and a struct with fields the caller must leave blank is a struct that will eventually be filled in.

When it returns nil, a durable implementation has flushed the message to its device: a crash one instruction later does not lose it. That flush is most of what the call costs — see the domain's BENCH.md. An implementation that joins the transaction ctx carries (ADR 0151) has instead made the message part of that transaction: it is durable, and receivable, once the transaction commits, and it never existed if the transaction rolls back.

A payload larger than [PolicyValue.MaxMessageBytes] is MessageTooLarge; an empty payload is legitimate, because a message whose whole meaning is its arrival is a normal thing to send.

func (Broker) Receive
func (Broker) Receive(ctx context.Context, max int) ([]DeliveryValue, error)

Receive leases up to max messages and returns them. It does NOT block waiting for work: an empty queue yields an empty slice and a nil error, and polling is the caller's (see internal/service/data/queue.Consume, which does it for you).

Each returned message is invisible to every other consumer until its [DeliveryValue.Lease] expires. A consumer that finishes acknowledges with Broker.Ack; a consumer that fails reports it with Broker.Nack; a consumer that DIES does neither, and the lapsed lease is what makes the message reappear.

max must be positive: InvalidBatchSize otherwise, rather than an empty slice forever from a consumer loop that spins and processes nothing.

type ConsumerConfig

ConsumerConfig is the public alias for Consume's configuration.

type ConsumerConfig = svcqueue.ConsumerConfig

ConsumerConfig is an alias of github.com/kitsunium/sdk/internal/service/data/queue.ConsumerConfig, declared as:

type ConsumerConfig struct {
	// Handler processes one delivery. It must be non-nil.
	Handler             corequeue.Handler
	// Clock drives the idle wait, the handler's lease bound, the drain, the
	// bound on each outcome and the lease renewals. Nil means clock.System.
	//
	// Nothing in this package calls time.Sleep; an idle worker waits through
	// Waiter.After and a cancelled context wakes it immediately.
	//
	// A lease's end is the broker's instant and is read on this clock, so it
	// must be the broker's: a test drives both with one ManualClock. A lease
	// that already looks ended on this clock the moment it is received —
	// two clocks that disagree — is not bounded at all, as before the bound
	// existed, rather than never run.
	Clock               clock.Timed
	// PollInterval is how long a worker waits after finding the queue empty.
	// Zero means [DefaultPollInterval]; negative is read as zero.
	//
	// When the broker is a [corequeue.Waker] — every broker in this package
	// is — it is an upper bound and not a cadence: a Publish or a Nack in
	// this process wakes the worker at once, and a retry delay ending or a
	// lease lapsing wakes it at that instant. What is left for the poll to
	// find is a publication another process made into a durable queue, so set
	// it to how late such a message may be noticed — seconds, not
	// milliseconds, is the usual answer — and an idle consumer then costs
	// nothing between polls.
	PollInterval        time.Duration
	// Parallelism is how many independent workers run. Each one polls, leases
	// and processes on its OWN goroutine, which is what makes this the
	// consumer's goroutine rather than the publisher's.
	//
	// Non-positive is clamped to 1: one worker is what a caller who did not
	// think about concurrency wants, it is never wrong, and there is nothing
	// here for the SDK to invent.
	//
	// Above 1 there is NO ordering left, in any sense.
	Parallelism         int
	// BatchSize is how many messages one worker leases per poll. Non-positive
	// is clamped to 1.
	//
	// Above 1 a worker processes its batch SEQUENTIALLY, so the extra
	// messages are held under a lease they are not yet being worked on: a
	// batch of 50 with a 30-second visibility timeout gives the last message
	// 30 seconds minus the other 49's processing time. Batch for throughput
	// against a durable broker, where one poll is a directory read; do not
	// batch to get parallelism, which is what Parallelism is.
	//
	// A message whose lease has run down to the margin [Consume] keeps
	// before its turn comes is not started: it is handed back uncounted
	// and leased again by a later poll — or, with AutoExtend, renewed
	// before it starts.
	BatchSize           int
	// SettleTimeout bounds each outcome a worker reports once ctx has
	// ended — an acknowledgement, a nack, a rejection or a release — and
	// each lease renewal AutoExtend sends. Zero means
	// [DefaultSettleTimeout]; negative is read as zero.
	//
	// Every outcome is sent on the consumer's context WITHOUT its
	// cancellation, so a stop landing between a handler's return and its
	// acknowledgement does not abandon it (kitsunium/sdk#472) and a batch a
	// stop interrupts is still handed back. That is what needs a bound: a
	// medium that stopped answering would otherwise hold the stop forever.
	// It runs on Clock, from the instant ctx ends — while the consumer
	// runs an outcome is not timed, and nothing but the worker waits on
	// it. An outcome the bound cut short is QueueBackendFailed, which a
	// stopping consumer reads as its stop, not as a failure.
	//
	// It also caps the margin a handler is stopped by before its lease
	// ends: a tenth of the lease, at most this.
	SettleTimeout       time.Duration
	// DrainTimeout is how long the message a worker is handling when ctx
	// ends may still run before its handler's context is cancelled — River's
	// SoftStopTimeout. Zero cancels it at once, which is what a stop always
	// did; negative is read as zero.
	//
	// A handler that finishes inside it is acknowledged; one the drain cut
	// short is handed back uncounted, through the broker's
	// [corequeue.Releaser]. No new message is leased once ctx has ended,
	// and the lease bound still applies: a drain never keeps a handler past
	// the margin before its lease ends. Measured on Clock. A caller's stop
	// budget — a lifecycle component's — must cover it plus SettleTimeout.
	DrainTimeout        time.Duration
	// AckedWindow is how many message identifiers the engine remembers
	// having handled: the last AckedWindow messages whose handler returned
	// nil or [corequeue.Acknowledged] in this process. A delivery of one of
	// them — its lease lapsed before the acknowledgement landed, and the
	// broker delivered it again — is acknowledged without running the
	// handler a second time.
	//
	// It narrows the duplicates at-least-once delivery leaves, as the mail
	// spool's memory of what it delivered does, for every broker; it does
	// not remove them. The window is this process's and is lost when it
	// stops: a redelivery to another consumer, or after a restart, still
	// runs the handler, which is still idempotent. Zero, or negative,
	// remembers nothing — every delivery runs the handler, what Consume
	// always did.
	AckedWindow         int
	// AutoExtend renews the lease of the message a handler is running, a
	// third of the lease after it was granted or last renewed, for as long
	// as the handler runs — the heartbeat of SQS, JetStream's in-progress
	// acknowledgement, Temporal's activity heartbeat. A message of a batch
	// whose lease is a third spent when its turn comes is renewed before
	// its handler starts. Each renewal is as long as the lease the broker
	// first granted, and is bounded by SettleTimeout.
	//
	// It needs the broker's [corequeue.LeaseExtender]; without it, or when
	// false, the lease is never renewed and the handler is stopped a margin
	// before it ends. A renewal the broker refuses — the lease lapsed, the
	// medium failed — cancels the handler's context with that refusal as
	// its cause: its work is now a duplicate in the making.
	//
	// A renewal replaces the receipt, and the engine acknowledges with the
	// newest one. So a handler that extends its own lease — parking its
	// message, as the mail spool does — must leave AutoExtend false: the
	// receipt it holds is void after the first renewal.
	AutoExtend          bool
	// HandlerIsIdempotent asserts, IN CODE, that Handler is safe to run more
	// than once for the same message.
	//
	// It must be true. Its zero value is false and is REFUSED
	// ([corequeue.ConsumerMisconfigured]), and that refusal is the domain's
	// one piece of deliberate ceremony.
	//
	// The reason is that this queue delivers AT LEAST ONCE and cannot do
	// otherwise. A consumer that dies between finishing the work and
	// acknowledging it has done the work and not said so, and the message
	// comes back — that is not a bug to be fixed, it is the only honest
	// behaviour available to a queue whose consumers can die. Exactly-once
	// delivery does not exist over a transport; what exists is at-least-once
	// plus an idempotent consumer, and the second half is the caller's.
	//
	// The SDK cannot check it, so it makes the caller assert it where a
	// reviewer will see it. It is precisely
	// resilience.HedgeConfig.Idempotent's instrument, for precisely its
	// reason, and it has the same useful side effect:
	// `grep -rn HandlerIsIdempotent` enumerates every place in a codebase
	// where somebody promised this.
	HandlerIsIdempotent bool
}

type Counter

Counter is the public alias for the capability of counting a queue's backlog — waiting, leased, dead, the oldest and the next due — without reading a payload. Every broker here implements it.

type Counter = corequeue.Counter

Counter is an alias of github.com/kitsunium/sdk/internal/core/data/queue.Counter, declared as:

type Counter interface {
	// Stats counts the queue as the broker records it now: no payload is
	// read, nothing is leased and no lapsed lease is reclaimed, so a lease
	// that ran out and nobody reclaimed yet still counts as leased. It is a
	// snapshot that races the queue, as [MessageReader.Messages] is.
	Stats(ctx context.Context) (StatsValue, error)
}

func (Counter) Stats
func (Counter) Stats(ctx context.Context) (StatsValue, error)

Stats counts the queue as the broker records it now: no payload is read, nothing is leased and no lapsed lease is reclaimed, so a lease that ran out and nobody reclaimed yet still counts as leased. It is a snapshot that races the queue, as MessageReader.Messages is.

type DeadLetter

DeadLetter is the public alias for one abandoned message and its cause.

type DeadLetter = corequeue.DeadLetterValue

DeadLetter is an alias of github.com/kitsunium/sdk/internal/core/data/queue.DeadLetterValue, declared as:

type DeadLetter struct {
	// FailedAt is when the last attempt failed and the message was moved.
	FailedAt   time.Time
	// Reason is the SCREAMING_SNAKE identifier of the failure — the
	// errs.Error Reason of the cause, NOT_RETRYABLE for a handler's
	// [DoNotRetry] around an error that carried none, or LEASE_EXPIRED when
	// the message died by exhausting its deliveries without any consumer ever
	// reporting anything.
	Reason     string
	// Cause is the wire-safe Public half of the last failure, or an empty
	// string when the consumer reported no cause at all. It never carries
	// the message payload, which is a security property this domain shares
	// with the rest of the SDK: a queue moves bytes a consumer must not
	// leak, and an error message is the classic place they leak from.
	Cause      string
	// Message is the work, unmodified, with the identifier it was published
	// under.
	Message    MessageValue
	// Trail is the dotted-quad codes the last failure gathered as it was
	// wrapped, oldest first — errs.TrailOf of the cause, whose origin is
	// Code and not repeated here —, at most sixteen, the newest kept when
	// there were more. It is what tells a failure apart once its process
	// is gone: a mail spool's SEND_REFUSED is a rejected mail when the
	// spool added MAIL_REJECTED to its trail, and a dead letter keeps that
	// code where its Reason and Cause keep only the origin's. Nil when the
	// cause gathered none, for a dead letter written before the field
	// existed, and on an SQL broker whose SQLConfig.Trail is not set.
	// Like Code, each entry is a uint32: errs.Code(dl.Trail[i]) converts.
	Trail      []uint32
	// Code is the dotted-quad errs.Code of the last failure, or 0 when the
	// cause carried none. It is a uint32 rather than an errs.Code so that
	// this package's values stay free of the kernel's error type in a shape
	// a consumer may serialise; errs.Code(dl.Code) converts.
	Code       uint32
	// Deliveries is how many times the message was delivered before it was
	// abandoned. It equals [PolicyValue.MaxDeliveries] for every dead letter
	// this SDK's brokers produce by running out of attempts, and is the count
	// the message had for one a handler rejected ([Rejecter]).
	Deliveries int
	// Priority is the level the message was published at
	// ([PriorityPublisher]); a replay queues it at that level again.
	// [PriorityNormal] for one written before priorities existed, and on
	// an SQL broker whose SQLConfig.Priority is not set.
	Priority   Priority
}

type DeadLetterBulkManager

DeadLetterBulkManager is the public alias for the capability of replaying the oldest dead letters, or deleting those that failed before an instant, in one call. Every broker here implements it.

type DeadLetterBulkManager = corequeue.DeadLetterBulkManager

DeadLetterBulkManager is an alias of github.com/kitsunium/sdk/internal/core/data/queue.DeadLetterBulkManager, declared as:

type DeadLetterBulkManager interface {
	// PurgeDeadLetters deletes up to max dead letters whose
	// [DeadLetterValue.FailedAt] lies before before, the oldest first in
	// [DeadLetterPager]'s order, and returns how many it deleted. It never
	// touches a live message.
	//
	// max must be positive ([InvalidBatchSize]). A failure part-way returns
	// the number deleted so far with the error, as ReplayDeadLetters does.
	PurgeDeadLetters(ctx context.Context, before time.Time, max int) (removed int, err error)
	// ReplayDeadLetters replays up to max dead letters, the oldest first in
	// [DeadLetterPager]'s order, each as [DeadLetterManager.ReplayDeadLetter]
	// replays one, and returns how many it replayed. Called again, it
	// replays the next ones; it returns 0 once the store is empty.
	//
	// max must be positive ([InvalidBatchSize]). A failure part-way returns
	// the number replayed so far with the error, and every dead letter not
	// counted is still a dead letter, so the caller resumes by calling
	// again. Idle consumers are woken.
	ReplayDeadLetters(ctx context.Context, max int) (replayed int, err error)
}

func (DeadLetterBulkManager) PurgeDeadLetters
func (DeadLetterBulkManager) PurgeDeadLetters(ctx context.Context, before time.Time, max int) (removed int, err error)

PurgeDeadLetters deletes up to max dead letters whose [DeadLetterValue.FailedAt] lies before before, the oldest first in DeadLetterPager's order, and returns how many it deleted. It never touches a live message.

max must be positive (InvalidBatchSize). A failure part-way returns the number deleted so far with the error, as ReplayDeadLetters does.

func (DeadLetterBulkManager) ReplayDeadLetters
func (DeadLetterBulkManager) ReplayDeadLetters(ctx context.Context, max int) (replayed int, err error)

ReplayDeadLetters replays up to max dead letters, the oldest first in DeadLetterPager's order, each as DeadLetterManager.ReplayDeadLetter replays one, and returns how many it replayed. Called again, it replays the next ones; it returns 0 once the store is empty.

max must be positive (InvalidBatchSize). A failure part-way returns the number replayed so far with the error, and every dead letter not counted is still a dead letter, so the caller resumes by calling again. Idle consumers are woken.

type DeadLetterManager

DeadLetterManager is the public alias for the capability of replaying a dead letter into its queue, its count reset, or deleting it. Every broker here implements it.

type DeadLetterManager = corequeue.DeadLetterManager

DeadLetterManager is an alias of github.com/kitsunium/sdk/internal/core/data/queue.DeadLetterManager, declared as:

type DeadLetterManager interface {
	// DeleteDeadLetter removes the dead letter id for good, payload, cause
	// and all. It is the only call in the domain that destroys a message
	// without its consumer's acknowledgement, and it is never made by the
	// SDK itself. An id the store does not hold is [DeadLetterNotFound].
	DeleteDeadLetter(ctx context.Context, id string) error
	// ReplayDeadLetter moves the dead letter id back into the queue, visible
	// at once, with its delivery count reset: its next delivery counts 1 and
	// it gets [PolicyValue.MaxDeliveries] attempts again. It keeps its ID, its
	// payload and its EnqueuedAt, so the log lines of both its lives join on
	// one identifier; the dead-letter record, its cause included, is gone once
	// the message is queued again — log it first if it matters.
	//
	// An id the store does not hold is [DeadLetterNotFound], and nothing is
	// queued. Two replays of one dead letter racing each other may both
	// queue it on a broker that cannot tell them apart, which is a duplicate
	// the at-least-once guarantee already permits; the brokers say which
	// they are.
	ReplayDeadLetter(ctx context.Context, id string) error
}

func (DeadLetterManager) DeleteDeadLetter
func (DeadLetterManager) DeleteDeadLetter(ctx context.Context, id string) error

DeleteDeadLetter removes the dead letter id for good, payload, cause and all. It is the only call in the domain that destroys a message without its consumer's acknowledgement, and it is never made by the SDK itself. An id the store does not hold is DeadLetterNotFound.

func (DeadLetterManager) ReplayDeadLetter
func (DeadLetterManager) ReplayDeadLetter(ctx context.Context, id string) error

ReplayDeadLetter moves the dead letter id back into the queue, visible at once, with its delivery count reset: its next delivery counts 1 and it gets [PolicyValue.MaxDeliveries] attempts again. It keeps its ID, its payload and its EnqueuedAt, so the log lines of both its lives join on one identifier; the dead-letter record, its cause included, is gone once the message is queued again — log it first if it matters.

An id the store does not hold is DeadLetterNotFound, and nothing is queued. Two replays of one dead letter racing each other may both queue it on a broker that cannot tell them apart, which is a duplicate the at-least-once guarantee already permits; the brokers say which they are.

type DeadLetterPager

DeadLetterPager is the public alias for the capability of paging through the dead letters after an identifier, oldest enqueued first. Every broker here implements it.

type DeadLetterPager = corequeue.DeadLetterPager

DeadLetterPager is an alias of github.com/kitsunium/sdk/internal/core/data/queue.DeadLetterPager, declared as:

type DeadLetterPager interface {
	// DeadLettersAfter returns up to max dead letters whose identifier sorts
	// after after — from the first when after is empty —, in the byte order
	// of their identifiers. Like [DeadLetterReader.DeadLetters] it removes
	// nothing. A dead letter written, replayed or deleted meanwhile may be
	// listed or not.
	//
	// max must be positive ([InvalidBatchSize]).
	DeadLettersAfter(ctx context.Context, after string, max int) ([]DeadLetterValue, error)
}

func (DeadLetterPager) DeadLettersAfter
func (DeadLetterPager) DeadLettersAfter(ctx context.Context, after string, max int) ([]DeadLetterValue, error)

DeadLettersAfter returns up to max dead letters whose identifier sorts after after — from the first when after is empty —, in the byte order of their identifiers. Like DeadLetterReader.DeadLetters it removes nothing. A dead letter written, replayed or deleted meanwhile may be listed or not.

max must be positive (InvalidBatchSize).

type DeadLetterReader

DeadLetterReader is the public alias for the capability of reading the dead-letter store back. Every broker here implements it.

type DeadLetterReader = corequeue.DeadLetterReader

DeadLetterReader is an alias of github.com/kitsunium/sdk/internal/core/data/queue.DeadLetterReader, declared as:

type DeadLetterReader interface {
	// DeadLetters returns up to max dead letters, oldest first. It does NOT
	// remove them: a dead letter is evidence, and a read that consumed it
	// would make the second investigator's job impossible. Removing one is a
	// decision, and [DeadLetterManager] is where it is made.
	//
	// max must be positive ([InvalidBatchSize]).
	DeadLetters(ctx context.Context, max int) ([]DeadLetterValue, error)
}

func (DeadLetterReader) DeadLetters
func (DeadLetterReader) DeadLetters(ctx context.Context, max int) ([]DeadLetterValue, error)

DeadLetters returns up to max dead letters, oldest first. It does NOT remove them: a dead letter is evidence, and a read that consumed it would make the second investigator's job impossible. Removing one is a decision, and DeadLetterManager is where it is made.

max must be positive (InvalidBatchSize).

type DeadLetterWriter

DeadLetterWriter is the public alias for the capability of writing a dead letter into the store as it was recorded elsewhere — what a copy of a queue from one broker to another needs. Every broker here implements it.

type DeadLetterWriter = corequeue.DeadLetterWriter

DeadLetterWriter is an alias of github.com/kitsunium/sdk/internal/core/data/queue.DeadLetterWriter, declared as:

type DeadLetterWriter interface {
	// PutDeadLetter writes dead into the dead-letter store and returns the
	// identifier the broker minted for it. Everything else is kept as given:
	// the payload, the enqueue instant, the failure instant, the delivery
	// count, the reason, the cause, the code — 0 included, the code of a
	// failure that carried none — and the trail, which an SQL broker
	// without SQLConfig.Trail does not keep. What
	// [DeadLetterReader.DeadLetters] then returns of it equals dead but for
	// the identifier — an empty trail read back nil —, and
	// [DeadLetterManager] acts on it as on any other dead letter.
	//
	// dead.Message.ID must be empty: the identifier is the broker's to mint,
	// as [Broker.Publish] mints it, and one a caller set would not be kept.
	// dead is refused with [InvalidDeadLetter] — the "field" field naming
	// which — when its Deliveries is below one or above 2147483647, when an
	// instant lies outside what a durable broker records (1970 to 2262),
	// when its Reason or Cause is longer than 120 runes or holds a line
	// break, or when its Trail holds more than sixteen codes; a payload
	// above [PolicyValue.MaxMessageBytes] is [MessageTooLarge]. A refused
	// dead letter writes nothing.
	//
	// It is never delivered and wakes no consumer. A broker that joins the
	// transaction ctx carries (ADR 0151) writes it in that transaction, so
	// it exists if and only if the transaction commits.
	PutDeadLetter(ctx context.Context, dead DeadLetterValue) (string, error)
}

func (DeadLetterWriter) PutDeadLetter
func (DeadLetterWriter) PutDeadLetter(ctx context.Context, dead DeadLetterValue) (string, error)

PutDeadLetter writes dead into the dead-letter store and returns the identifier the broker minted for it. Everything else is kept as given: the payload, the enqueue instant, the failure instant, the delivery count, the reason, the cause, the code — 0 included, the code of a failure that carried none — and the trail, which an SQL broker without SQLConfig.Trail does not keep. What DeadLetterReader.DeadLetters then returns of it equals dead but for the identifier — an empty trail read back nil —, and DeadLetterManager acts on it as on any other dead letter.

dead.Message.ID must be empty: the identifier is the broker's to mint, as Broker.Publish mints it, and one a caller set would not be kept. dead is refused with InvalidDeadLetter — the "field" field naming which — when its Deliveries is below one or above 2147483647, when an instant lies outside what a durable broker records (1970 to 2262), when its Reason or Cause is longer than 120 runes or holds a line break, or when its Trail holds more than sixteen codes; a payload above [PolicyValue.MaxMessageBytes] is MessageTooLarge. A refused dead letter writes nothing.

It is never delivered and wakes no consumer. A broker that joins the transaction ctx carries (ADR 0151) writes it in that transaction, so it exists if and only if the transaction commits.

type DelayedNacker

DelayedNacker is the public alias for the capability of handing a leased message back after a delay its caller names. Every broker here implements it, and Consume uses it for a RetryAfter failure.

type DelayedNacker = corequeue.DelayedNacker

DelayedNacker is an alias of github.com/kitsunium/sdk/internal/core/data/queue.DelayedNacker, declared as:

type DelayedNacker interface {
	// NackAfter reports that processing failed and hands the message back,
	// receivable again delay after now on the broker's clock — whatever
	// [PolicyValue.RetryDelay] and [PolicyValue.MaxRetryDelay] would have
	// made it wait. Everything else is [Broker.Nack]'s: the delivery is
	// counted, and a message delivered [PolicyValue.MaxDeliveries] times is
	// moved to the dead-letter store with cause instead, the delay unused.
	// See [NackValue].
	//
	// A zero delay makes the message receivable at once, and a negative one
	// is read as zero: both have the one reading "now". A delay whose instant
	// a durable broker cannot record (past 2262, Unix nanoseconds) is
	// [QueueMisconfigured], refused before anything moves, and the lease the
	// caller holds stays valid. A lapsed or unknown receipt is refused
	// exactly as [Broker.Ack] refuses it.
	NackAfter(ctx context.Context, receipt ReceiptValue, cause error, delay time.Duration) (NackValue, error)
}

func (DelayedNacker) NackAfter
func (DelayedNacker) NackAfter(ctx context.Context, receipt ReceiptValue, cause error, delay time.Duration) (NackValue, error)

NackAfter reports that processing failed and hands the message back, receivable again delay after now on the broker's clock — whatever [PolicyValue.RetryDelay] and [PolicyValue.MaxRetryDelay] would have made it wait. Everything else is Broker.Nack's: the delivery is counted, and a message delivered [PolicyValue.MaxDeliveries] times is moved to the dead-letter store with cause instead, the delay unused. See [NackValue].

A zero delay makes the message receivable at once, and a negative one is read as zero: both have the one reading "now". A delay whose instant a durable broker cannot record (past 2262, Unix nanoseconds) is QueueMisconfigured, refused before anything moves, and the lease the caller holds stays valid. A lapsed or unknown receipt is refused exactly as Broker.Ack refuses it.

type DelayedPublisher

DelayedPublisher is the public alias for the capability of publishing a message no consumer receives before an instant. Every broker here implements it.

type DelayedPublisher = corequeue.DelayedPublisher

DelayedPublisher is an alias of github.com/kitsunium/sdk/internal/core/data/queue.DelayedPublisher, declared as:

type DelayedPublisher interface {
	// PublishAt appends payload to the queue as [Broker.Publish] does, and
	// hands it to no consumer before at. Its EnqueuedAt is the broker's now,
	// as every publication's is; at is when it becomes visible, which
	// [MessageReader.Messages] reports as its VisibleAt.
	//
	// An at already past — the zero time.Time included — means now: the
	// message is receivable at once, as a published one is. An at a durable
	// broker cannot record (past 2262, Unix nanoseconds) is
	// [QueueMisconfigured], and nothing is published.
	//
	// Everything else is Publish's: durable when it returns nil, part of the
	// transaction ctx carries on a broker that joins one (ADR 0151), a
	// payload above [PolicyValue.MaxMessageBytes] refused with
	// [MessageTooLarge]. Idle consumers are woken, and from then on wait
	// until at rather than polling for it. A queue delivers by visibility,
	// so a deferred message comes behind every message visible before it.
	PublishAt(ctx context.Context, payload []byte, at time.Time) (MessageValue, error)
}

func (DelayedPublisher) PublishAt
func (DelayedPublisher) PublishAt(ctx context.Context, payload []byte, at time.Time) (MessageValue, error)

PublishAt appends payload to the queue as Broker.Publish does, and hands it to no consumer before at. Its EnqueuedAt is the broker's now, as every publication's is; at is when it becomes visible, which MessageReader.Messages reports as its VisibleAt.

An at already past — the zero time.Time included — means now: the message is receivable at once, as a published one is. An at a durable broker cannot record (past 2262, Unix nanoseconds) is QueueMisconfigured, and nothing is published.

Everything else is Publish's: durable when it returns nil, part of the transaction ctx carries on a broker that joins one (ADR 0151), a payload above [PolicyValue.MaxMessageBytes] refused with MessageTooLarge. Idle consumers are woken, and from then on wait until at rather than polling for it. A queue delivers by visibility, so a deferred message comes behind every message visible before it.

type Delivery

Delivery is the public alias for one message handed to one consumer, carrying the lease that proves the claim and the count that says whether this is a retry.

type Delivery = corequeue.DeliveryValue

Delivery is an alias of github.com/kitsunium/sdk/internal/core/data/queue.DeliveryValue, declared as:

type Delivery struct {
	// Lease is the claim this delivery grants and the receipt that ends it.
	Lease      LeaseValue
	// Message is the work.
	Message    MessageValue
	// Deliveries counts how many times this message has been handed to a
	// consumer, INCLUDING this one. It is 1 on the first delivery.
	//
	// It is a field of the port rather than a note in the documentation
	// because it is the at-least-once guarantee made unignorable: a consumer
	// cannot read a delivery without reading the number that says this may
	// not be the first time. A value above 1 means a previous consumer
	// failed, timed out, or died — the three are indistinguishable from
	// here, deliberately, since none of them changes what the handler must
	// do.
	//
	// It is also the number [PolicyValue.MaxDeliveries] is compared against:
	// a delivery whose count already equals the maximum is the last one this
	// message gets before the dead-letter store.
	Deliveries int
}

type FileConfig

FileConfig is the public alias for NewFile's configuration.

type FileConfig = svcqueue.FileConfig

FileConfig is an alias of github.com/kitsunium/sdk/internal/service/data/queue.FileConfig, declared as:

type FileConfig struct {
	// Clock reads time. It is the narrow half of kernel/clock because this
	// broker never waits: a lapsed lease is noticed by the next Receive, not
	// by a timer.
	//
	// Injecting a ManualClock makes every deadline in a SINGLE-process test
	// instant. It cannot make a cross-process test instant, and that
	// asymmetry is not an oversight: a lease deadline is a wall-clock instant
	// written into a filename precisely so that another process can read it,
	// and another process does not share this one's fake clock. The
	// consumer-death test therefore waits on the real clock, and says so.
	//
	// Nil means clock.System.
	Clock  clock.Clock
	// Random draws the retry jitter [corequeue.PolicyValue].RetryJitter asks
	// for: a number in [0, 1), as math/rand/v2.Float64 returns. Nil means
	// math/rand/v2.Float64. A test injects a fixed sequence and asserts every
	// retry instant to the nanosecond, as it does without a jitter; the
	// three brokers draw from it identically, so the memory broker stays an
	// exact double. It may be called by several goroutines at once.
	Random func() float64
	// Dir is the queue directory. It is created owner-only when absent, and
	// refused when it exists and any account outside the owner and group
	// could replace its entries. Its three state directories are held to the
	// same rule — created owner-only, checked when somebody else made them —
	// and refused besides when one is a symlink or not a directory at all.
	// On Windows "owner-only" is not something a mode can say: a directory
	// inherits its parent's access control list, and the rule reads that list
	// instead (dirtrust_windows.go).
	//
	// Two brokers over one Dir — in one process or in two — are ONE queue.
	// That is the domain's inter-process claim, and it is the whole reason
	// the directory is the configuration rather than an implementation
	// detail.
	Dir    string
	// Policy is the delivery discipline. Its zero value is refused; see
	// core/data/queue.PolicyValue.
	Policy corequeue.PolicyValue
}

type Handler

Handler is the public alias for the function that processes one delivery.

type Handler = corequeue.Handler

Handler is an alias of github.com/kitsunium/sdk/internal/core/data/queue.Handler, declared as:

type Handler func(ctx context.Context, delivery DeliveryValue) error

type Lease

Lease is the public alias for a consumer's exclusive claim on one message.

type Lease = corequeue.LeaseValue

Lease is an alias of github.com/kitsunium/sdk/internal/core/data/queue.LeaseValue, declared as:

type Lease struct {
	// ExpiresAt is when the claim lapses and the message becomes eligible for
	// redelivery. It is the broker's clock.
	//
	// It is a promise about VISIBILITY and not about mutual exclusion: after
	// this instant another consumer may hold the same message while this one
	// is still running. A handler that is still working past it is producing
	// a duplicate, which the at-least-once guarantee already permits — see
	// [LeaseExtender] for the way to avoid producing one on purpose.
	ExpiresAt time.Time
	// Receipt names this lease to [Broker.Ack] and [Broker.Nack].
	Receipt   ReceiptValue
}

type LeaseExtender

LeaseExtender is the public alias for the capability of renewing a lease a handler is still working under. Every broker here implements it.

type LeaseExtender = corequeue.LeaseExtender

LeaseExtender is an alias of github.com/kitsunium/sdk/internal/core/data/queue.LeaseExtender, declared as:

type LeaseExtender interface {
	// Extend renews the lease named by receipt for a further by, measured
	// from now, and returns the lease that replaces it.
	//
	// It returns a NEW [LeaseValue] — a new receipt included — and the old
	// receipt is void from that moment. That is not ceremony: an
	// implementation whose lease deadline lives in a durable name can only
	// change it by replacing the name, and a receipt that silently kept
	// meaning something after the thing it named had moved would be the one
	// bug this whole domain is built to avoid.
	//
	// A lapsed lease is [LeaseExpired] and is NOT renewed, even when no
	// other consumer has taken the message yet: expiry is a deadline, not
	// "until somebody else wants it". The handler that asks and is refused
	// has learnt that its work is now a duplicate, which is exactly when it
	// needed to know.
	//
	// A non-positive by is [QueueMisconfigured]: the two readings of zero —
	// "lapse now" and "never lapse" — are opposites and both harmful. So is a
	// by whose deadline reaches past what a durable implementation can record
	// (2262, Unix nanoseconds): it is refused before anything moves, and the
	// lease the caller holds stays valid. Validate never sees this duration,
	// so [MaxDeadlineOffset] cannot bound it; the instant itself is checked.
	Extend(ctx context.Context, receipt ReceiptValue, by time.Duration) (LeaseValue, error)
}

func (LeaseExtender) Extend
func (LeaseExtender) Extend(ctx context.Context, receipt ReceiptValue, by time.Duration) (LeaseValue, error)

Extend renews the lease named by receipt for a further by, measured from now, and returns the lease that replaces it.

It returns a NEW [LeaseValue] — a new receipt included — and the old receipt is void from that moment. That is not ceremony: an implementation whose lease deadline lives in a durable name can only change it by replacing the name, and a receipt that silently kept meaning something after the thing it named had moved would be the one bug this whole domain is built to avoid.

A lapsed lease is LeaseExpired and is NOT renewed, even when no other consumer has taken the message yet: expiry is a deadline, not "until somebody else wants it". The handler that asks and is refused has learnt that its work is now a duplicate, which is exactly when it needed to know.

A non-positive by is QueueMisconfigured: the two readings of zero — "lapse now" and "never lapse" — are opposites and both harmful. So is a by whose deadline reaches past what a durable implementation can record (2262, Unix nanoseconds): it is refused before anything moves, and the lease the caller holds stays valid. Validate never sees this duration, so MaxDeadlineOffset cannot bound it; the instant itself is checked.

type LiveMessage

LiveMessage is the public alias for one live message as a broker records it.

type LiveMessage = corequeue.LiveMessageValue

LiveMessage is an alias of github.com/kitsunium/sdk/internal/core/data/queue.LiveMessageValue, declared as:

type LiveMessage struct {
	// VisibleAt is when the message may next be delivered: its visibility
	// for a waiting one — now, or later after a retry delay —, the deadline
	// of its lease for a leased one. An instant already past means
	// receivable now.
	VisibleAt  time.Time
	// Message is the work, with the identifier it was published under.
	Message    MessageValue
	// Deliveries is how many times the message has been delivered: 0 for
	// one never leased, the count a retry or a lapsed lease left.
	Deliveries int
	// Leased reports that a consumer holds the message's lease. Such a
	// message is still alive: it comes back at VisibleAt unless it is
	// acknowledged.
	Leased     bool
	// Priority is the level the message is delivered at
	// ([PriorityPublisher]), kept across a retry and a copy.
	Priority   Priority
}

type MemoryConfig

MemoryConfig is the public alias for NewMemory's configuration.

type MemoryConfig = svcqueue.MemoryConfig

MemoryConfig is an alias of github.com/kitsunium/sdk/internal/service/data/queue.MemoryConfig, declared as:

type MemoryConfig struct {
	// Clock reads time. A memory broker never WAITS — expiry is noticed on
	// the next Receive — so it asks for the narrow half of kernel/clock and
	// a test can drive every deadline in this package with a ManualClock and
	// no sleeping at all.
	//
	// Nil means clock.System.
	Clock  clock.Clock
	// Random draws the retry jitter [corequeue.PolicyValue].RetryJitter asks
	// for: a number in [0, 1), as math/rand/v2.Float64 returns. Nil means
	// math/rand/v2.Float64. A test injects a fixed sequence and asserts every
	// retry instant to the nanosecond, as it does without a jitter; the
	// three brokers draw from it identically, so the memory broker stays an
	// exact double. It may be called by several goroutines at once.
	Random func() float64
	// Policy is the delivery discipline. Its zero value is refused; see
	// core/data/queue.PolicyValue.
	Policy corequeue.PolicyValue
}

type Message

Message is the public alias for one unit of work as the broker minted it.

type Message = corequeue.MessageValue

Message is an alias of github.com/kitsunium/sdk/internal/core/data/queue.MessageValue, declared as:

type Message struct {
	// EnqueuedAt is when the broker accepted the message. It is the ordering
	// key a FIFO implementation sorts on, and it is the broker's clock, not
	// the producer's.
	EnqueuedAt time.Time
	// ID identifies the message for the life of the queue and beyond it: a
	// dead letter carries the same value, so the log line the failing
	// consumer wrote and the record left on disk can be joined without either
	// one carrying the other's half of the story (CLAUDE.md rule 4).
	//
	// It is opaque. Its shape is the implementation's, it is unique within
	// one queue, and nothing outside the broker that minted it may parse it.
	ID         string
	// Payload is the caller's bytes, unmodified and uninterpreted. This
	// domain does not encode: pick a codec, marshal, publish the bytes.
	//
	// The slice belongs to the receiver — a durable broker read it from a
	// device, and an in-memory one copies before handing it over, so a
	// handler may keep it, mutate it, or hold it past the acknowledgement
	// without corrupting a redelivery.
	Payload    []byte
}

type MessageLister

MessageLister is the public alias for the capability of ranging over the live messages through an iterator, as MessageReader lists them a page at a time. Every broker here implements it.

type MessageLister = corequeue.MessageLister

MessageLister is an alias of github.com/kitsunium/sdk/internal/core/data/queue.MessageLister, declared as:

type MessageLister interface {
	// AllMessages yields every live message whose identifier sorts after
	// after — from the first when after is empty —, in the byte order of
	// their identifiers, as [MessageReader.Messages] lists them.
	//
	// A failure of the broker's storage, or ctx ending, is yielded once as
	// the error of a zero [LiveMessageValue], and the iteration stops there;
	// the identifier of the last message yielded before it is where a later
	// call resumes. A loop that breaks early stops the reading: nothing is
	// held open past the break.
	AllMessages(ctx context.Context, after string) iter.Seq2[LiveMessageValue, error]
}

func (MessageLister) AllMessages
func (MessageLister) AllMessages(ctx context.Context, after string) iter.Seq2[LiveMessageValue, error]

AllMessages yields every live message whose identifier sorts after after — from the first when after is empty —, in the byte order of their identifiers, as MessageReader.Messages lists them.

A failure of the broker's storage, or ctx ending, is yielded once as the error of a zero [LiveMessageValue], and the iteration stops there; the identifier of the last message yielded before it is where a later call resumes. A loop that breaks early stops the reading: nothing is held open past the break.

type MessageReader

MessageReader is the public alias for the capability of listing the live messages — waiting and leased — without leasing one. Every broker here implements it.

type MessageReader = corequeue.MessageReader

MessageReader is an alias of github.com/kitsunium/sdk/internal/core/data/queue.MessageReader, declared as:

type MessageReader interface {
	// Messages returns up to max live messages whose identifier sorts after
	// after — from the first when after is empty —, in the byte order of
	// their identifiers, each with its payload, its enqueue instant, its
	// delivery count, whether a consumer leases it and when it may next be
	// delivered. It leases nothing, counts no delivery and reclaims no lapsed
	// lease: what it reads is what the broker records, and the last
	// identifier it returns is where a later call resumes. A message
	// published, acknowledged or dead-lettered meanwhile may be listed or
	// not.
	//
	// max must be positive ([InvalidBatchSize]).
	Messages(ctx context.Context, after string, max int) ([]LiveMessageValue, error)
}

func (MessageReader) Messages
func (MessageReader) Messages(ctx context.Context, after string, max int) ([]LiveMessageValue, error)

Messages returns up to max live messages whose identifier sorts after after — from the first when after is empty —, in the byte order of their identifiers, each with its payload, its enqueue instant, its delivery count, whether a consumer leases it and when it may next be delivered. It leases nothing, counts no delivery and reclaims no lapsed lease: what it reads is what the broker records, and the last identifier it returns is where a later call resumes. A message published, acknowledged or dead-lettered meanwhile may be listed or not.

max must be positive (InvalidBatchSize).

type MessageWriter

MessageWriter is the public alias for the capability of writing a live message as another broker recorded it — its count, its instants. Every broker here implements it.

type MessageWriter = corequeue.MessageWriter

MessageWriter is an alias of github.com/kitsunium/sdk/internal/core/data/queue.MessageWriter, declared as:

type MessageWriter interface {
	// PutMessage writes live into the queue as a waiting message and returns
	// the identifier the broker minted for it. Everything else is kept as
	// given: the payload, the enqueue instant, the delivery count — 0
	// included — and the instant it becomes visible, before which it is
	// never delivered. What [MessageReader.Messages] then returns of it
	// equals live but for the identifier, and its next delivery counts one
	// more than live.Deliveries.
	//
	// live.Message.ID must be empty: the identifier is the broker's to mint,
	// as [Broker.Publish] mints it. live.Leased must be false: a lease is
	// the broker's that granted it, and a written message is waiting. live is
	// refused with [InvalidMessage] — the "field" field naming which — for
	// either, for a delivery count below 0 or above 2147483647, and for an
	// instant outside what a durable broker records (1970 to 2262); a
	// payload above [PolicyValue.MaxMessageBytes] is [MessageTooLarge]. A
	// refused message writes nothing.
	//
	// A broker that joins the transaction ctx carries (ADR 0151) writes it in
	// that transaction, so it exists if and only if the transaction commits,
	// and wakes its consumers after the commit, as a publication does.
	PutMessage(ctx context.Context, live LiveMessageValue) (string, error)
}

func (MessageWriter) PutMessage
func (MessageWriter) PutMessage(ctx context.Context, live LiveMessageValue) (string, error)

PutMessage writes live into the queue as a waiting message and returns the identifier the broker minted for it. Everything else is kept as given: the payload, the enqueue instant, the delivery count — 0 included — and the instant it becomes visible, before which it is never delivered. What MessageReader.Messages then returns of it equals live but for the identifier, and its next delivery counts one more than live.Deliveries.

live.Message.ID must be empty: the identifier is the broker's to mint, as Broker.Publish mints it. live.Leased must be false: a lease is the broker's that granted it, and a written message is waiting. live is refused with InvalidMessage — the "field" field naming which — for either, for a delivery count below 0 or above 2147483647, and for an instant outside what a durable broker records (1970 to 2262); a payload above [PolicyValue.MaxMessageBytes] is MessageTooLarge. A refused message writes nothing.

A broker that joins the transaction ctx carries (ADR 0151) writes it in that transaction, so it exists if and only if the transaction commits, and wakes its consumers after the commit, as a publication does.

type Nack

Nack is the public alias for what Broker.Nack decided: retried, or dead-lettered.

type Nack = corequeue.NackValue

Nack is an alias of github.com/kitsunium/sdk/internal/core/data/queue.NackValue, declared as:

type Nack struct {
	// Deliveries is how many times the message had been delivered when it
	// failed, counting from 1.
	Deliveries   int
	// VisibleAt is when the message becomes available again. It is the zero
	// Time when DeadLettered is true.
	VisibleAt    time.Time
	// DeadLettered reports that the message exhausted
	// [PolicyValue.MaxDeliveries] and was moved to the dead-letter store
	// with its cause. It will not be delivered again, unless somebody replays
	// it ([DeadLetterManager]).
	DeadLettered bool
}

type OncePublisher

OncePublisher is the public alias for the capability of publishing at most once per key over a window. Every broker here implements it.

type OncePublisher = corequeue.OncePublisher

OncePublisher is an alias of github.com/kitsunium/sdk/internal/core/data/queue.OncePublisher, declared as:

type OncePublisher interface {
	// PublishOnce publishes payload as [Broker.Publish] does and reserves key
	// for window from now — unless key is still reserved, and then it
	// publishes nothing. It reports which: true with the message it minted,
	// or false with the zero [MessageValue] when an earlier publication's
	// window still holds key.
	//
	// A duplicate is an answer and not an error, on purpose: it is the
	// expected outcome of the retry this call exists for, and a caller
	// asking "did it go out?" must not have to tell a sentinel from a
	// failure. The message the earlier publication minted is not returned:
	// the key is what the caller holds.
	//
	// The window runs on the broker's clock from the publication that
	// reserved key, and ends there: a publication after it publishes again
	// and reserves the key anew. A reservation past its window is removed
	// lazily, by a later publication through this sibling. Keys are compared
	// byte for byte; the durable brokers keep only their SHA-256.
	//
	// An empty key is refused, and so is a non-positive window — "never
	// deduplicate" and "forever" are opposite readings of zero, and either
	// would be a publication that reads as deduplicated and is not — or one
	// whose end a durable broker cannot record (2262): [QueueMisconfigured],
	// naming the argument, and nothing is published. A payload above
	// [PolicyValue.MaxMessageBytes] is [MessageTooLarge].
	//
	// A broker that joins the transaction ctx carries (ADR 0151) reserves key
	// in that transaction: the reservation and the message exist if and only
	// if it commits. What a crash between the two costs is the broker's to
	// say, and it is at most a duplicate, never a lost message.
	PublishOnce(ctx context.Context, key string, payload []byte, window time.Duration) (message MessageValue, published bool, err error)
}

func (OncePublisher) PublishOnce
func (OncePublisher) PublishOnce(ctx context.Context, key string, payload []byte, window time.Duration) (message MessageValue, published bool, err error)

PublishOnce publishes payload as Broker.Publish does and reserves key for window from now — unless key is still reserved, and then it publishes nothing. It reports which: true with the message it minted, or false with the zero [MessageValue] when an earlier publication's window still holds key.

A duplicate is an answer and not an error, on purpose: it is the expected outcome of the retry this call exists for, and a caller asking "did it go out?" must not have to tell a sentinel from a failure. The message the earlier publication minted is not returned: the key is what the caller holds.

The window runs on the broker's clock from the publication that reserved key, and ends there: a publication after it publishes again and reserves the key anew. A reservation past its window is removed lazily, by a later publication through this sibling. Keys are compared byte for byte; the durable brokers keep only their SHA-256.

An empty key is refused, and so is a non-positive window — "never deduplicate" and "forever" are opposite readings of zero, and either would be a publication that reads as deduplicated and is not — or one whose end a durable broker cannot record (2262): QueueMisconfigured, naming the argument, and nothing is published. A payload above [PolicyValue.MaxMessageBytes] is MessageTooLarge.

A broker that joins the transaction ctx carries (ADR 0151) reserves key in that transaction: the reservation and the message exist if and only if it commits. What a crash between the two costs is the broker's to say, and it is at most a duplicate, never a lost message.

type Policy

Policy is the public alias for the delivery discipline a broker enforces.

type Policy = corequeue.PolicyValue

Policy is an alias of github.com/kitsunium/sdk/internal/core/data/queue.PolicyValue, declared as:

type Policy struct {
	// VisibilityTimeout is how long a [Broker.Receive] hides a message from
	// every other consumer. It is the deadline the consumer is racing, and
	// it must be longer than the slowest run of the handler — or the handler
	// must renew it through [LeaseExtender].
	//
	// It is the single knob that sets how quickly a dead consumer's work is
	// picked up: a message held by a process that has been SIGKILLed
	// reappears exactly this long after it was leased, and not before.
	//
	// A non-positive value is refused ([QueueMisconfigured]), and so is one
	// above [MaxDeadlineOffset].
	VisibilityTimeout   time.Duration
	// RetryDelay is how long a NACKED message stays invisible before it
	// becomes eligible again, measured from the nack.
	//
	// It does NOT apply to a lease that simply lapsed: that message has
	// already been invisible for a whole visibility timeout, and charging it
	// a second wait would punish a crashed consumer more than a failing one.
	//
	// Zero is a working value and means "eligible as soon as the nack
	// returns". Negative is read as zero. Above [MaxDeadlineOffset] it is
	// refused ([QueueMisconfigured]).
	//
	// It is the same wait after every failure unless [PolicyValue]
	// .MaxRetryDelay makes it grow.
	RetryDelay          time.Duration
	// MaxRetryDelay makes the retry delay GROW, and is where the growth
	// stops. When it is positive, a message nacked on its n-th delivery waits
	// RetryDelay × 2^(n−1), held at MaxRetryDelay — the SDK's one backoff
	// curve (kernel/backoff, ADR 0103), keyed on the delivery that failed,
	// so a downstream that is down is asked less and less often instead of at
	// a fixed cadence it is already failing to keep up with (ADR 0151).
	//
	// Zero keeps RetryDelay constant, which is what every policy did before
	// this field existed: a policy that does not name it behaves exactly as
	// it did. That is the zero's one reading — a ceiling nobody set is not a
	// request for unbounded growth, which could not be recorded anyway.
	//
	// A negative value is refused, because it is not a ceiling and the caller
	// who wrote it meant something; so is one above [MaxDeadlineOffset],
	// whose deadline a durable broker could not record; so is a ceiling with
	// no RetryDelay to grow from, which would leave every retry at zero while
	// reading like a backoff; and so is a ceiling below RetryDelay, which
	// would silently shorten every wait the caller asked for
	// ([QueueMisconfigured], naming the field).
	MaxRetryDelay       time.Duration
	// MaxDeliveries is how many times one message may be handed to a
	// consumer before the broker gives up and dead-letters it. It is
	// compared against [DeliveryValue.Deliveries], which counts from 1, so a
	// value of 1 means "one attempt, then the dead-letter store" and is a
	// legitimate configuration rather than an off-by-one.
	//
	// A non-positive value is refused ([QueueMisconfigured]).
	MaxDeliveries       int
	// MaxMessageBytes bounds one payload. A larger payload is refused at
	// [Broker.Publish] with [MessageTooLarge], at the call site that built
	// it, rather than accepted and discovered by a consumer that cannot read
	// it back.
	//
	// Zero means [DefaultMaxMessageBytes]. Negative is refused, because it
	// is not a bound at all and the caller who wrote it meant something.
	MaxMessageBytes     int
	// MaxAge is how long a message may wait for a consumer, measured from
	// its [MessageValue.EnqueuedAt]. A message whose age has reached it is
	// never handed to a consumer: the next [Broker.Receive] that reaches it
	// moves it to the dead-letter store with [MessageExpired] as its reason
	// — evidence of what expired, never a message silently gone. A message
	// a consumer already leases is not taken from it: its acknowledgement
	// still lands, and only a delivery after its age is refused.
	//
	// It is for work that is worthless late — a notification an hour
	// stale, a cache warm-up for a deployment already replaced —, the
	// lifetime SQS and RabbitMQ call a message's retention and TTL.
	//
	// Zero means a message never expires, which is what every policy did
	// before the field existed: that is its one reading. A negative value
	// is refused, and so is one above [MaxDeadlineOffset]
	// ([QueueMisconfigured], naming the field).
	MaxAge              time.Duration
	// DeadLetterRetention is how long a dead letter is kept, measured from
	// its [DeadLetterValue.FailedAt]. A dead letter older than it is removed
	// by a later [Broker.Receive] — at most one sweep a minute on the
	// broker's clock —, so a queue that dead-letters steadily stops growing
	// its store until somebody deletes by hand.
	//
	// Zero keeps every dead letter for good, which is what every policy did
	// before the field existed. A negative value is refused, and so is one
	// above [MaxDeadlineOffset] ([QueueMisconfigured], naming the field).
	DeadLetterRetention time.Duration
	// RetryJitter widens each policy retry delay by a random fraction of
	// itself, drawn from [0, RetryJitter) and added, so messages nacked in
	// the same instant — every message of every consumer during a
	// downstream's outage — come back spread over a window instead of in
	// one burst aimed at a dependency already failing to keep up. A delay
	// held at MaxRetryDelay may reach MaxRetryDelay × (1 + RetryJitter).
	// It is read only under the zero [PolicyValue.RetryJitterMode].
	//
	// Zero is no jitter: the exact instant every retry waited before the
	// field existed. A value outside [0, 1], NaN included, is refused, and
	// so is a jitter with no MaxRetryDelay — a growing delay is what it
	// spreads ([QueueMisconfigured], naming the field). A delay a caller
	// names through [DelayedNacker] is the caller's, and never jittered.
	RetryJitter         float64
	// RetryJitterMode is how the retry delay is randomised, as
	// kernel/backoff's curve is (ADR 0184): the zero, additive, widens it
	// by [PolicyValue.RetryJitter]; full draws the whole delay from
	// [0, grown]; decorrelated draws it from [RetryDelay, 3 × the delay
	// before], held at MaxRetryDelay. The last two ignore RetryJitter, and
	// setting one beside them is refused rather than ignored.
	//
	// A mode other than the zero needs a MaxRetryDelay, as RetryJitter
	// does, and a mode the package does not define is refused
	// ([QueueMisconfigured], naming the field).
	RetryJitterMode     kbackoff.JitterMode
}

func (Policy) Normalized
func (Policy) Normalized() PolicyValue

Normalized returns the policy with its clamped fields resolved, so that every implementation reads the same numbers rather than each remembering to apply the same default.

It does NOT validate: call [PolicyValue.Validate] first, at construction, where the refusal reaches the caller who wrote the mistake.

func (Policy) Validate
func (Policy) Validate() error

Validate reports whether the policy is one a broker can honour, and refuses it BY FIELD when it is not.

It lives in core so that the memory, file and SQL brokers refuse identical inputs identically — the property that makes the first usable as a test double for the other two. It is the same instrument core/data/vfs's ValidatePath and ValidatePerm are, for the same reason.

type Priority

Priority is the public alias for the level a message is delivered at.

type Priority = corequeue.Priority

Priority is an alias of github.com/kitsunium/sdk/internal/core/data/queue.Priority, declared as:

type Priority int8

type PriorityPublisher

PriorityPublisher is the public alias for the capability of publishing a message at a level, the visible messages of the highest level received first, strictly. Every broker here implements it; the SQL broker keeps a level once SQLPriorityMigration has run and SQLConfig.Priority is set.

type PriorityPublisher = corequeue.PriorityPublisher

PriorityPublisher is an alias of github.com/kitsunium/sdk/internal/core/data/queue.PriorityPublisher, declared as:

type PriorityPublisher interface {
	// PublishPriority appends payload to the queue at level, as
	// [Broker.Publish] appends one at [PriorityNormal], and returns the
	// message the broker minted.
	//
	// A level outside the five [PriorityLowest] to [PriorityHighest] is
	// [QueueMisconfigured], and nothing is published — so is one a broker
	// cannot keep. A payload above [PolicyValue.MaxMessageBytes] is
	// [MessageTooLarge].
	PublishPriority(ctx context.Context, payload []byte, level Priority) (MessageValue, error)
}

func (PriorityPublisher) PublishPriority
func (PriorityPublisher) PublishPriority(ctx context.Context, payload []byte, level Priority) (MessageValue, error)

PublishPriority appends payload to the queue at level, as Broker.Publish appends one at PriorityNormal, and returns the message the broker minted.

A level outside the five PriorityLowest to PriorityHighest is QueueMisconfigured, and nothing is published — so is one a broker cannot keep. A payload above [PolicyValue.MaxMessageBytes] is MessageTooLarge.

type Purged

Purged is the public alias for what Purger.Purge removed.

type Purged = corequeue.PurgeValue

Purged is an alias of github.com/kitsunium/sdk/internal/core/data/queue.PurgeValue, declared as:

type Purged struct {
	// Queued is how many waiting messages were removed, visible or not yet.
	Queued int
	// Leased is how many leased messages were removed — a lapsed lease
	// nobody had reclaimed yet among them.
	Leased int
}

type Purger

Purger is the public alias for the capability of removing every live message of a queue, never a dead letter. Every broker here implements it.

type Purger = corequeue.Purger

Purger is an alias of github.com/kitsunium/sdk/internal/core/data/queue.Purger, declared as:

type Purger interface {
	// Purge removes every live message and reports how many were waiting
	// and how many were leased. It is, beside
	// [DeadLetterManager.DeleteDeadLetter] and
	// [DeadLetterBulkManager.PurgeDeadLetters], a call that destroys messages
	// without their consumers' acknowledgement, and it is never made by the
	// SDK itself.
	//
	// A consumer still working under a lease it held finds the message gone
	// at its acknowledgement, which then answers [LeaseExpired]. A message
	// published while Purge runs may survive it. Idle consumers are woken,
	// so they read the emptied queue at once. A broker that joins the
	// transaction ctx carries (ADR 0151) purges in that transaction.
	Purge(ctx context.Context) (PurgeValue, error)
}

func (Purger) Purge
func (Purger) Purge(ctx context.Context) (PurgeValue, error)

Purge removes every live message and reports how many were waiting and how many were leased. It is, beside DeadLetterManager.DeleteDeadLetter and DeadLetterBulkManager.PurgeDeadLetters, a call that destroys messages without their consumers' acknowledgement, and it is never made by the SDK itself.

A consumer still working under a lease it held finds the message gone at its acknowledgement, which then answers LeaseExpired. A message published while Purge runs may survive it. Idle consumers are woken, so they read the emptied queue at once. A broker that joins the transaction ctx carries (ADR 0151) purges in that transaction.

type Receipt

Receipt is the public alias for the opaque handle to one lease. Do not parse it and do not construct one.

type Receipt = corequeue.ReceiptValue

Receipt is an alias of github.com/kitsunium/sdk/internal/core/data/queue.ReceiptValue, declared as:

type Receipt string

type Rejecter

Rejecter is the public alias for the capability of dead-lettering a leased message at once, with the cause that condemned it. Every broker here implements it, and Consume uses it for a DoNotRetry failure.

type Rejecter = corequeue.Rejecter

Rejecter is an alias of github.com/kitsunium/sdk/internal/core/data/queue.Rejecter, declared as:

type Rejecter interface {
	// Reject moves the leased message named by receipt to the dead-letter
	// store with cause, recorded as [Broker.Nack] records the cause of a
	// message that ran out of attempts: its Reason, its dotted-quad Code and
	// its wire-safe Public half, never the Private one. The dead letter's
	// Deliveries is the count the message had, which may be below
	// [PolicyValue.MaxDeliveries].
	//
	// A lapsed receipt is [LeaseExpired] and one this broker cannot read is
	// [UnknownReceipt], exactly as [Broker.Ack] refuses them; neither moves
	// anything.
	Reject(ctx context.Context, receipt ReceiptValue, cause error) error
}

func (Rejecter) Reject
func (Rejecter) Reject(ctx context.Context, receipt ReceiptValue, cause error) error

Reject moves the leased message named by receipt to the dead-letter store with cause, recorded as Broker.Nack records the cause of a message that ran out of attempts: its Reason, its dotted-quad Code and its wire-safe Public half, never the Private one. The dead letter's Deliveries is the count the message had, which may be below [PolicyValue.MaxDeliveries].

A lapsed receipt is LeaseExpired and one this broker cannot read is UnknownReceipt, exactly as Broker.Ack refuses them; neither moves anything.

type Releaser

Releaser is the public alias for the capability of handing a leased message back at once without counting its delivery. Every broker here implements it, and Consume uses it for the messages a stop or a spent lease kept the handler from starting, or finishing.

type Releaser = corequeue.Releaser

Releaser is an alias of github.com/kitsunium/sdk/internal/core/data/queue.Releaser, declared as:

type Releaser interface {
	// Release makes the leased message named by receipt receivable again
	// now, by any consumer, with its delivery count put back to what it was
	// before this lease: its next delivery carries the [DeliveryValue]
	// .Deliveries this one did. It is never dead-lettered, whatever that
	// count, and no retry delay applies. It wakes an idle consumer, as a
	// Nack that hands a message back does.
	//
	// A lapsed receipt is [LeaseExpired] and one this broker cannot read is
	// [UnknownReceipt], exactly as [Broker.Ack] refuses them; neither moves
	// anything. The receipt is void once Release returns nil.
	Release(ctx context.Context, receipt ReceiptValue) error
}

func (Releaser) Release
func (Releaser) Release(ctx context.Context, receipt ReceiptValue) error

Release makes the leased message named by receipt receivable again now, by any consumer, with its delivery count put back to what it was before this lease: its next delivery carries the [DeliveryValue] .Deliveries this one did. It is never dead-lettered, whatever that count, and no retry delay applies. It wakes an idle consumer, as a Nack that hands a message back does.

A lapsed receipt is LeaseExpired and one this broker cannot read is UnknownReceipt, exactly as Broker.Ack refuses them; neither moves anything. The receipt is void once Release returns nil.

type SQLConfig

SQLConfig is the public alias for NewSQL's configuration: Transactor, Dialect and Table (required), Policy, Clock, Trail, which keeps a dead letter's trail once SQLTrailMigration has run, and Priority, which keeps a message's level once SQLPriorityMigration has run.

type SQLConfig = svcqueue.SQLConfig

SQLConfig is an alias of github.com/kitsunium/sdk/internal/service/data/queue.SQLConfig, declared as:

type SQLConfig struct {
	// Transactor is the transaction manager of the database the queue lives
	// in — the one the caller's own transactions are opened with, so a call
	// made under one of them joins it. It must also be a core/data/sql Joiner and
	// Deferrer, which the SDK's own transactor is: the first is how a
	// publication joins the caller's transaction, the second how it wakes a
	// consumer only once that transaction has committed.
	Transactor coresql.Transactor
	// Clock stamps every instant the broker writes — a message's visibility,
	// a lease's deadline, a dead letter's failure — and reads the instant a
	// lease is judged against. Nil means clock.System.
	//
	// The instants are the BROKER's, never the database's NOW(): a test
	// drives every deadline with a ManualClock, as it does the other two
	// brokers'. Brokers in several processes over one table compare
	// instants their own clocks wrote, so those clocks must agree to within
	// the precision a lease needs — as they must for the file broker.
	Clock      clock.Clock
	// Random draws the retry jitter [corequeue.PolicyValue].RetryJitter asks
	// for: a number in [0, 1), as math/rand/v2.Float64 returns. Nil means
	// math/rand/v2.Float64. A test injects a fixed sequence and asserts every
	// retry instant to the nanosecond, as it does without a jitter; the
	// three brokers draw from it identically, so the memory broker stays an
	// exact double. It may be called by several goroutines at once.
	Random     func() float64
	// Table names the queue's one table, which [SQLMigration] creates. It is
	// a lower-case SQL identifier of at most [MaxSQLTableLen] bytes, holds no
	// three underscores in a row — docstore's derived tables do — and does
	// not start with "sqlite_".
	//
	// Two brokers over one database and one Table — in one process or in
	// twenty — are ONE queue, as two file brokers over one directory are.
	Table      string
	// Policy is the delivery discipline. Its zero value is refused; see
	// core/data/queue.PolicyValue.
	Policy     corequeue.PolicyValue
	// Dialect is the engine's, and must be the Transactor's own: the SQL the
	// broker speaks to it.
	Dialect    coresql.Dialect
	// Trail makes the broker keep a dead letter's trail
	// (core/data/queue.DeadLetterValue.Trail) in the table's trail column,
	// which [SQLMigration] creates and [SQLTrailMigration] adds to a table an
	// earlier SQLMigration created: run that migration before setting it.
	// Unset, the broker sends the statements it sent before the column
	// existed, and a dead letter it buries or reads has no trail.
	Trail      bool
	// Priority makes the broker keep a message's level
	// (core/data/queue.PriorityPublisher) in the table's priority column,
	// which [SQLPriorityMigration] adds, and hand out the visible rows of
	// the highest level first: run that migration before setting it.
	// Unset, the broker sends the statements it sent before the column
	// existed, reads every message as core/data/queue.PriorityNormal, and
	// refuses to publish or write one at any other level.
	Priority   bool
}

type Stats

Stats is the public alias for a queue's backlog as Counter reads it.

type Stats = corequeue.StatsValue

Stats is an alias of github.com/kitsunium/sdk/internal/core/data/queue.StatsValue, declared as:

type Stats struct {
	// OldestEnqueuedAt is the enqueue instant of the oldest live message —
	// waiting or leased —, the zero time.Time when there is none. Its age
	// is how long the queue has been behind.
	OldestEnqueuedAt time.Time
	// NextDueAt is the earliest instant a waiting message becomes visible —
	// an instant already past when one is visible now —, the zero
	// time.Time when nothing waits.
	NextDueAt        time.Time
	// Queued counts the messages waiting for a consumer, visible now or
	// held until a later instant.
	Queued           int64
	// Leased counts the messages a consumer leases, a lapsed lease nobody
	// reclaimed yet among them.
	Leased           int64
	// Dead counts the dead letters.
	Dead             int64
}

type Wake

Wake is the public alias for what an idle consumer waits on: a signal the next Publish or Nack in this process closes, and how long until something the broker holds becomes receivable on its own.

type Wake = corequeue.WakeValue

Wake is an alias of github.com/kitsunium/sdk/internal/core/data/queue.WakeValue, declared as:

type Wake struct {
	// Signal is closed by the next event that may make a message receivable
	// through this broker in this process: a Publish, or a Nack that hands a
	// message back. It is never nil, and it is never closed by a publication
	// made in another process.
	Signal    <-chan struct{}
	// In is how long, measured on the broker's own clock at the moment Wake
	// was called, until a message the broker already holds becomes
	// receivable with no further event: a retry delay ending, or a lease
	// lapsing because its consumer died. It is meaningful only when
	// Scheduled is true; zero or less means now.
	//
	// It is a duration and not an instant so that a consumer whose clock is
	// not the broker's still waits the right length of time rather than
	// comparing two clocks that disagree.
	In        time.Duration
	// Scheduled reports that In names such a moment. It is false when the
	// broker holds nothing that will become receivable on its own.
	Scheduled bool
}

type Waker

Waker is the public alias for the capability of telling an idle consumer when to look again. Every broker here implements it, and Consume uses it; a broker of your own that omits it is simply polled.

type Waker = corequeue.Waker

Waker is an alias of github.com/kitsunium/sdk/internal/core/data/queue.Waker, declared as:

type Waker interface {
	// Wake returns what an idle consumer should wait on. Call it BEFORE the
	// Receive that may come back empty: a publication landing between that
	// Receive and the wait closes the channel already handed out, so it is
	// not missed.
	Wake() WakeValue
}

func (Waker) Wake
func (Waker) Wake() WakeValue

Wake returns what an idle consumer should wait on. Call it BEFORE the Receive that may come back empty: a publication landing between that Receive and the wait closes the channel already handed out, so it is not missed.

Generated from docs/api by tools/genindex.

Documentation

Overview

Package queue is the public facade for the SDK's ASYNCHRONOUS, DURABLE message queue: one process publishes work, another process picks it up later, and a crash in between loses nothing. 

broker, err := queue.NewFile(queue.FileConfig{
	Dir: "/var/lib/myapp/jobs",
	Policy: queue.Policy{
		VisibilityTimeout: 30 * time.Second, // longer than the slowest handler
		MaxDeliveries:     5,                // then the dead-letter store
		RetryDelay:        2 * time.Second,
	},
})

// …in the producer, possibly in another process entirely:
msg, err := broker.Publish(ctx, payload)

// …in the consumer:
err = queue.Consume(ctx, broker, queue.ConsumerConfig{
	Handler: func(ctx context.Context, d queue.Delivery) error {
		return process(ctx, d.Message.Payload) // d.Deliveries says if this is a retry
	},
	HandlerIsIdempotent: true, // required; see below
	Parallelism:         4,
})

This is not an event bus, and the difference is the whole design

Read this before anything else in the package. pkg/v1/app/events and this package are described with the same words — publish, handler, message — and they guarantee opposite things: 

                | events                     | queue (this package)
----------------|----------------------------|---------------------------
Scope           | one process                | many processes
Timing          | synchronous                | asynchronous
Goroutine       | the publisher's            | a consumer's
Transaction     | the publisher's            | its own
Durability      | none                       | the point of it
Retries / DLQ   | none                       | the point of it
If the process  | the event never happened   | the message is still
dies mid-flight |                            | there

If you want several things to happen right now, on your goroutine, inside your transaction, you want events and this package will be slower and harder for no benefit: Broker.Publish on a durable broker costs a disk flush, because that flush IS the guarantee.

The Transaction row is about where the WORK runs, and a handler always runs in its own. What NewSQL adds is that the MESSAGE can be part of yours: see "A queue in your own database" below.

Delivery is AT LEAST ONCE, and you must handle duplicates

Exactly-once delivery does not exist over a transport. What exists is at-least-once delivery plus an idempotent consumer, and any library that advertises the first is selling you the second with your half left as an exercise. This one says so instead — in three places you cannot miss:

  • Delivery.Deliveries is a field of every delivery. It counts from 1, and a value above 1 means these bytes have been handled before, by you or by a consumer in another process.
  • ConsumerConfig.HandlerIsIdempotent must be set to true. Its zero value is refused, so a handler cannot acquire the promise by omission, and `grep -rn HandlerIsIdempotent` enumerates every place in your codebase where somebody made it.
  • A message is removed at ACKNOWLEDGEMENT, never at read. That is what makes redelivery possible, and it is not negotiable: a consumer that dies after doing the work and before acknowledging it has done the work and not said so, and the queue has no way to tell that from a consumer that died before doing anything.

Ordering is NOT guaranteed, and a single retry is enough to lose it

One consumer, no failures, one durable broker: messages arrive in publication order. Change any of the three and they do not.

The one people discover in production is the retry. A message that fails is made visible again LATER, so messages published after it are delivered before it is retried. There is no per-key ordering here and no ordered mode; if two messages must be processed in order, they are one message.

What a crash actually costs

A consumer that is SIGKILLed mid-handler holds a lease it will never release. Nothing is lost: the lease lapses after Policy.VisibilityTimeout and the next consumer to look — in any process on that machine — picks the message up with Delivery.Deliveries incremented. That timeout is therefore how long a dead consumer's work sits idle, and it is the one number worth thinking about when you wire this.

A message whose consumers keep dying is dead-lettered like any other, by the same Policy.MaxDeliveries count, so a payload that crashes the process cannot loop forever taking every consumer with it.

The dead-letter store keeps the cause

After Policy.MaxDeliveries attempts the message is moved to the dead-letter store together with why: the failure's reason, its dotted-quad code, and its wire-safe Public half. The log-only Private half is NOT kept — a dead-letter store is read by whoever is investigating, often not the process or even the trust domain that failed — and Message.ID is the join key back to the log line that has it.

Read them with the DeadLetterReader capability, which every broker here implements. Reading does not remove them; evidence that a read consumes is evidence the second investigator does not get. 

if reader, ok := broker.(queue.DeadLetterReader); ok {
	dead, err := reader.DeadLetters(ctx, 100)
}

What becomes of one is then a decision, and DeadLetterManager is where it is taken: ReplayDeadLetter queues it again — same ID, same payload, its count reset, so it has every attempt again — once the downstream is fixed, and DeleteDeadLetter removes it for good. An ID the store does not hold is DeadLetterNotFound. 

if manager, ok := broker.(queue.DeadLetterManager); ok {
	err := manager.ReplayDeadLetter(ctx, dead[0].Message.ID)
}

A dead letter moves between brokers with DeadLetterWriter: PutDeadLetter writes one into the store as another broker recorded it — its payload, its count, its instants, its reason, cause and code — under an identifier the target mints. It is what a copy of a queue from a directory into a database needs, and nothing else does; a value no store keeps as given is InvalidDeadLetter. The live messages move the same way: MessageReader lists them — waiting and leased — without leasing one, and MessageWriter writes one as another broker recorded it, its delivery count and its instants kept; a value no broker keeps as given is InvalidMessage.

A failure no retry can fix is dead-lettered at once

A payload that does not decode will not decode on the fifth attempt either. Return DoNotRetry around the error and Consume dead-letters the message on its FIRST failure, through the broker's Rejecter, with your error as the cause the dead letter records: 

var order Order
if err := json.Unmarshal(d.Message.Payload, &order); err != nil {
	return queue.DoNotRetry(err)
}

Keep it for failures that belong to the message. A downstream that is down, a lock that is held, a deadline that passed are what retries are for. The mark is recognised with errs.HasCode(err, queue.CodeNotRetryable); a broker of your own without Rejecter is nacked as before, and the message then reaches the dead-letter store after its last attempt with the same cause.

A retry when the handler says, a message for later, a publication once

Three more siblings, every broker here implementing each (ADR 0183).

A handler that knows when to try again — a downstream that answered with a Retry-After, a server that greylisted the sender — returns RetryAfter around its error, and Consume hands the message back through DelayedNacker with that delay instead of the policy's; the delivery is counted, as a nack's always is: 

if resp.StatusCode == http.StatusTooManyRequests {
	return queue.RetryAfter(errThrottled, 30*time.Second)
}

Work that falls due once, later — a reminder, a trial that ends, an order cancelled if still unpaid — is published with DelayedPublisher: no consumer receives it before its instant, and it is as durable as any message. It does not recur: work that repeats on a calendar is pkg/v1/app/scheduler's, whose job may publish here. 

if deferred, ok := broker.(queue.DelayedPublisher); ok {
	_, err := deferred.PublishAt(ctx, payload, trialEnds)
}

A producer that may retry a publication whose answer it never got publishes under a key with OncePublisher: within the window, a second publication under the same key publishes nothing and says so — a boolean, not an error, since it is the answer the retry was looking for. The queue still delivers at least once; this only stops the producer adding a copy. 

if once, ok := broker.(queue.OncePublisher); ok {
	_, published, err := once.PublishOnce(ctx, "order-"+id+"-paid", payload, 5*time.Minute)
}

A retry delay that grows

Policy.RetryDelay is the same wait after every failure. Set Policy.MaxRetryDelay and it grows instead — RetryDelay after the first failure, doubling, never more than the ceiling — so a downstream that is down is asked less and less often: 

queue.Policy{
	VisibilityTimeout: 30 * time.Second,
	MaxDeliveries:     10,
	RetryDelay:        time.Second,     // 1 s, 2 s, 4 s, …
	MaxRetryDelay:     5 * time.Minute, // … never more than five minutes
}

Zero keeps the constant delay every policy had before the field existed. A ceiling with no RetryDelay to grow from, or one below it, is refused.

Zero values are safe or refused, never inert

Policy.VisibilityTimeout and Policy.MaxDeliveries are REFUSED at zero, because each has two natural readings that are opposites and one of each pair silently destroys the guarantee — "redeliver instantly" against "never redeliver", "unlimited attempts" against "no attempts". Any value the SDK invented would be arbitrary, and a lease lifetime belongs to the work being protected.

Policy.RetryDelay and Policy.MaxMessageBytes are CLAMPED, because their zeros have one reading each and it is harmless: no extra delay, and DefaultMaxMessageBytes.

The three durations are also bounded from ABOVE by MaxDeadlineOffset, a century, and refused past it — not as a judgement about leases but because the durable brokers write every deadline as Unix nanoseconds, into a file name or a 64-bit column, which end in 2262. A deadline past that could not be read back, and the message it names would never be delivered again. The math.MaxInt64 somebody reaches for to mean "never" is 292 years, so it is refused rather than stranding the message; an extension that would reach past 2262 is refused by every broker for the same reason.

An idle consumer sleeps until there is work

Every broker here implements Waker, and Consume waits on it: a Publish, a Nack or a replay in this process wakes an idle worker at once, and a retry delay ending or a lease lapsing wakes it at that instant. ConsumerConfig .PollInterval is then an upper bound rather than a cadence — what is left for it to find is a message ANOTHER process published into a durable queue — so an idle consumer costs nothing between polls and a poll of a few seconds loses no latency inside one process: 

queue.Consume(ctx, broker, queue.ConsumerConfig{
	Handler:             handle,
	HandlerIsIdempotent: true,
	PollInterval:        5 * time.Second, // bounds only a publication from another process
})

Two durable brokers over one directory — or one database and table — in one process share their wake, as they share everything else: they are one queue.

A stop hands back what it did not do, and a handler ends before its lease

When the consumer's context ends, Consume leases nothing more. The message a handler is running gets ConsumerConfig.DrainTimeout to finish; what the stop cut short, and every message of the batch never started, is handed back through Releaser — receivable at once, its delivery not counted — so a deployment no longer moves work toward the dead-letter store. Every outcome is sent past the stop, bounded by ConsumerConfig.SettleTimeout, so give the consumer a stop budget that covers both.

A handler's context ends a margin before its lease does — a tenth of it, at most SettleTimeout — with LEASE_EXPIRED as its cause, since past that instant another consumer may hold the same message. A handler that legitimately runs longer than a lease sets ConsumerConfig.AutoExtend, and its lease is renewed every third of it while it runs: 

queue.Consume(ctx, broker, queue.ConsumerConfig{
	Handler:             handle,
	HandlerIsIdempotent: true,
	DrainTimeout:        10 * time.Second, // let the message in hand finish
	AutoExtend:          true,             // renew the lease while it runs
})

A queue in your own database, joined to your transaction

NewSQL keeps the queue in one table of the PostgreSQL, MySQL or SQLite database your application already writes, handed over as the sql.Transactor you open your own transactions with. Every call runs on the transaction its context carries — a savepoint of it — so a message published inside yours exists if and only if yours commits. That is the transactional outbox: the write and the message that announces it are never one without the other, and a process that dies between your commit and a publication loses nothing, because there is no between. 

create, err := queue.SQLMigration(sql.DialectPostgres, "app__jobs", 20260930120000) // run by your Migrator
jobs, err := queue.NewSQL(queue.SQLConfig{
	Transactor: tm, Dialect: sql.DialectPostgres, Table: "app__jobs",
	Policy: queue.Policy{VisibilityTimeout: 30 * time.Second, MaxDeliveries: 5},
})
err = sql.Transact(ctx, tm, func(ctx context.Context, _ sql.Executor) error {
	if err := orders.Insert(ctx, order); err != nil {
		return err // no message either
	}
	_, err := jobs.Publish(ctx, payload) // joins this transaction
	return err
})

Idle consumers are woken once the publication's transaction commits, never for one rolled back. An idle Receive is one indexed read and takes no lock; a Receive that leases locks the rows it takes and skips those another consumer holds (FOR UPDATE SKIP LOCKED — MySQL 8.0.1, MariaDB 10.6) or, on SQLite, takes the one write lock. A failure of the database is QueueBackendFailed, with the driver's error reachable through errors.As and its text — which quotes rows — withheld; it ends Consume, as a storage failure always has, and a database fails transiently where a directory rarely does — so run a consumer over it under lifecycle's supervisor, and open SQLite with a busy timeout.

Three brokers

NewFile and NewSQL are the real ones. NewFile's state is a directory, it survives the process, and two brokers over one directory — in one process or in twenty — are one queue. NewSQL's state is a table, with the same guarantees, and the one property a directory cannot give: a publication inside your database transaction.

NewMemory is the test double. It gives your own tests the port's full semantics with no directory and no flush, which is what makes them fast enough to run on every save. It is one process and it holds nothing on a device, so a restart is a total loss: using it in production buys the asynchrony and throws away the durability, which is the exact confusion the frontier table above exists to prevent.

Index

Constants

View Source 
const (
	// CodeNotRetryable is the code [DoNotRetry] adds to a failure's wrap
	// trail: errs.HasCode(err, CodeNotRetryable) recognises the mark, where
	// errors.Is(err, NotRetryable) sees only an origin.
	CodeNotRetryable errs.Code = corequeue.CodeNotRetryable
	// CodeDeadLetterNotFound is [DeadLetterNotFound]'s code.
	CodeDeadLetterNotFound errs.Code = corequeue.CodeDeadLetterNotFound
	// CodeSQLQueueMisconfigured is [SQLQueueMisconfigured]'s code.
	CodeSQLQueueMisconfigured errs.Code = corequeue.CodeSQLQueueMisconfigured
)

The codes of the sentinels ADR 0151 added, for errs.HasCode.

View Source 
const (
	// CodeMessageExpired is [MessageExpired]'s code: the reason and code a
	// dead letter records for a message that outlived [Policy].MaxAge.
	CodeMessageExpired errs.Code = corequeue.CodeMessageExpired
	// CodeAcknowledged is [Acknowledged]'s code: errs.HasCode(err,
	// CodeAcknowledged) recognises a handler's report that it acknowledged
	// its delivery itself.
	CodeAcknowledged errs.Code = corequeue.CodeAcknowledged
)

The codes of the two sentinels ADR 0054's 2026-11-02 amendment added, for errs.HasCode.

View Source 
const CodeInvalidDeadLetter errs.Code = corequeue.CodeInvalidDeadLetter

CodeInvalidDeadLetter is InvalidDeadLetter's code, the one sentinel ADR 0178 added, for errs.HasCode.

View Source 
const CodeInvalidMessage errs.Code = corequeue.CodeInvalidMessage

CodeInvalidMessage is InvalidMessage's code, for errs.HasCode.

View Source 
const CodeRetryRequested errs.Code = corequeue.CodeRetryRequested

CodeRetryRequested is the code RetryAfter adds to a failure's wrap trail: errs.HasCode(err, CodeRetryRequested) recognises the mark.

View Source 
const DefaultMaxMessageBytes int = corequeue.DefaultMaxMessageBytes

DefaultMaxMessageBytes is the payload bound applied when Policy.MaxMessageBytes is left at zero: one mebibyte.

View Source 
const DefaultPollInterval time.Duration = svcqueue.DefaultPollInterval

DefaultPollInterval is how long an idle worker waits before asking again when ConsumerConfig.PollInterval is left at zero. With a Waker broker it bounds only how late another process's publication is noticed.

View Source 
const DefaultSettleTimeout time.Duration = svcqueue.DefaultSettleTimeout

DefaultSettleTimeout bounds each acknowledgement, nack, rejection or release Consume sends past a stop, when ConsumerConfig.SettleTimeout is left at zero: five seconds.

View Source 
const MaxDeadlineOffset time.Duration = corequeue.MaxDeadlineOffset

MaxDeadlineOffset is the ceiling on Policy.VisibilityTimeout and Policy.RetryDelay: a century, refused above it rather than clamped, because a deadline past 2262 cannot be written into the durable broker's file names and read back.

View Source 
const MaxSQLTableLen int = svcqueue.MaxSQLTableLen

MaxSQLTableLen is the longest table name NewSQL and SQLMigration accept, in bytes: PostgreSQL's identifier limit.

Variables

View Source 
var (
	// QueueMisconfigured refuses a Policy field the broker cannot honour.
	QueueMisconfigured = corequeue.QueueMisconfigured
	// MessageTooLarge refuses a payload above the policy's bound, at the
	// producer, where it can still be made smaller.
	MessageTooLarge = corequeue.MessageTooLarge
	// UnknownReceipt refuses an acknowledgement naming a lease this queue
	// never issued.
	UnknownReceipt = corequeue.UnknownReceipt
	// LeaseExpired reports an acknowledgement whose lease had already lapsed.
	// Nothing happened, and the message is elsewhere.
	LeaseExpired = corequeue.LeaseExpired
	// InvalidBatchSize refuses a non-positive batch size.
	InvalidBatchSize = corequeue.InvalidBatchSize
	// NotRetryable is the failure no retry can fix: what a dead letter
	// records when [DoNotRetry] marked an error with no public words of its
	// own, or none at all.
	NotRetryable = corequeue.NotRetryable
	// DeadLetterNotFound refuses a replay or a deletion naming a dead letter
	// the store does not hold.
	DeadLetterNotFound = corequeue.DeadLetterNotFound
	// InvalidDeadLetter refuses a dead letter [DeadLetterWriter] cannot write
	// as it was given; the "field" field names which value.
	InvalidDeadLetter = corequeue.InvalidDeadLetter
	// InvalidMessage refuses a live message [MessageWriter] cannot write as it
	// was given; the "field" field names which value.
	InvalidMessage = corequeue.InvalidMessage
	// RetryRequested is what a dead letter records when [RetryAfter] marked an
	// error with no public words of its own, or none at all, and the message
	// ran out of attempts.
	RetryRequested = corequeue.RetryRequested
	// MessageExpired is what a dead letter records of a message that outlived
	// [Policy].MaxAge before any consumer was handed it.
	MessageExpired = corequeue.MessageExpired
	// Acknowledged is what a [Handler] returns once it has acknowledged its
	// delivery itself, inside the transaction that holds its effects:
	// [Consume] then neither acknowledges nor hands the message back.
	Acknowledged = corequeue.Acknowledged
	// SQLQueueMisconfigured refuses an SQLConfig no SQL broker could run: no
	// transactor, one it cannot join a transaction of, a dialect it cannot
	// spell, or a table name it cannot use.
	SQLQueueMisconfigured = corequeue.SQLQueueMisconfigured
	// QueueBackendFailed reports a refusal from a durable broker's storage:
	// the filesystem's, or a statement the SQL broker's database did not
	// complete.
	QueueBackendFailed = corequeue.QueueBackendFailed
	// QueueDirectoryUnusable refuses a queue directory that is missing, is
	// not a directory, or is writable by accounts that must not be able to
	// inject or drain messages.
	QueueDirectoryUnusable = corequeue.QueueDirectoryUnusable
	// ConsumerMisconfigured refuses a consumer with no Handler, or one whose
	// author has not asserted HandlerIsIdempotent.
	ConsumerMisconfigured = corequeue.ConsumerMisconfigured
	// HandlerPanicked reports a recovered handler panic. The message was
	// nacked and the consumer kept running.
	//
	// There is deliberately no HandlerFailed beside it: a handler that
	// returns an error has its error recorded in the dead letter unmodified,
	// because a verdict wrapped around it would displace the reason whoever
	// reads that record actually needs.
	HandlerPanicked = corequeue.HandlerPanicked
)

Functions

func Consume 

func Consume(ctx context.Context, broker Broker, cfg ConsumerConfig) error

Consume runs cfg.Handler against broker until ctx is done, on cfg.Parallelism goroutines it owns.

It returns nil when ctx ends — a cancelled consumer is a stopped consumer, not a failed one — and returns the broker's error when the STORAGE fails. A handler's error is never returned: it is nacked, retried, and eventually dead-lettered, which is the entire point of the domain.

A stop hands back, uncounted, every message a handler did not finish, after cfg.DrainTimeout; a handler's context ends a margin before its lease does, unless cfg.AutoExtend renews it. See the package documentation.

cfg.HandlerIsIdempotent must be true; see the package documentation.

func DoNotRetry 

func DoNotRetry(cause error) error

DoNotRetry marks cause as a failure no retry can fix: returned from a Handler, it makes Consume dead-letter the message at once, with cause, instead of handing it back for another attempt. An SDK error keeps its own reason, code and public words in the dead letter; any other error, or none, is recorded as NotRetryable.

func RetryAfter added in v0.10.0 

func RetryAfter(cause error, delay time.Duration) error

RetryAfter marks cause as a failure to retry no sooner than delay from now: returned from a Handler, it makes Consume hand the message back through the broker's DelayedNacker with that delay instead of the policy's retry delay, its delivery counted. A negative delay is read as zero; DoNotRetry wins over it.

func SQLMigration 

func SQLMigration(dialect sql.Dialect, table string, version uint64) (sql.Migration, error)

SQLMigration returns the migration that creates the one table an SQL queue named table keeps on dialect, numbered version for the caller's own version table. Its Down drops the table, and every message in it. Its statement does nothing when the table exists. Its Checksum fingerprints Up's statements, so a version table that recorded it refuses a build whose statements differ.

func SQLPriorityMigration added in v0.12.0 

func SQLPriorityMigration(dialect sql.Dialect, table string, version uint64) (sql.Migration, error)

SQLPriorityMigration returns the migration that adds the priority column and its index a broker with SQLConfig.Priority reads to a table SQLMigration created, numbered version for the caller's own version table. It does nothing when the table has the column or does not exist; its Down drops the index and the column.

func SQLTrailMigration added in v0.11.0 

func SQLTrailMigration(dialect sql.Dialect, table string, version uint64) (sql.Migration, error)

SQLTrailMigration returns the migration that adds the trail column a broker with SQLConfig.Trail writes to a table an earlier SQLMigration created, numbered version for the caller's own version table. It does nothing when the table has the column or does not exist; its Down drops the column.

Types

type AckOutcome added in v0.12.0 

type AckOutcome = corequeue.AckOutcome

AckOutcome is the public alias for what Acker.AckMany did with one receipt. 

const (
	// AckNotAttempted is the zero [AckOutcome]: [Acker].AckMany returned an
	// error before it reached the receipt.
	AckNotAttempted AckOutcome = corequeue.AckNotAttempted
	// AckAcknowledged is a receipt whose message was removed for good.
	AckAcknowledged AckOutcome = corequeue.AckAcknowledged
	// AckLeaseExpired is a receipt whose lease had lapsed: nothing was
	// removed, as [Broker].Ack answers [LeaseExpired].
	AckLeaseExpired AckOutcome = corequeue.AckLeaseExpired
	// AckUnknownReceipt is a receipt the broker never issued: nothing was
	// removed, as [Broker].Ack answers [UnknownReceipt].
	AckUnknownReceipt AckOutcome = corequeue.AckUnknownReceipt
)

The four outcomes Acker.AckMany reports for one receipt.

type AckResult added in v0.12.0 

type AckResult = corequeue.AckResultValue

AckResult is the public alias for what Acker.AckMany reports.

type Acker added in v0.12.0 

type Acker = corequeue.Acker

Acker is the public alias for the capability of acknowledging several leased messages in one call, with one outcome per receipt. Every broker here implements it.

type BatchPublisher added in v0.12.0 

type BatchPublisher = corequeue.BatchPublisher

BatchPublisher is the public alias for the capability of publishing several messages in one call. Every broker here implements it.

type Broker 

type Broker = corequeue.Broker

Broker is the public alias for the queue contract. It is FROZEN at four methods; capabilities arrive as siblings (DeadLetterReader, LeaseExtender, Waker, Rejecter, Releaser, DelayedNacker, DelayedPublisher, OncePublisher, DeadLetterManager, DeadLetterWriter, MessageReader, MessageWriter, MessageLister, Counter, Purger, DeadLetterPager, DeadLetterBulkManager, BatchPublisher, Acker) reached by type assertion.

func NewFile 

func NewFile(cfg FileConfig) (broker Broker, err error)

NewFile returns the durable broker: its whole state is the directory in cfg.Dir, so it survives the process that published into it and two brokers over one directory are one queue.

It refuses at construction — never at first use — a policy it cannot honour, a directory it cannot use safely, and a platform with no atomic replace or no flushable directory handle. Windows is such a platform: after the directory checks — which read its access control lists there — NewFile returns the SDK's UNSUPPORTED_PLATFORM, matched with errors.Is(err, proc.UnsupportedPlatform).

The broker holds two directory descriptors and additionally implements io.Closer, which releases them — reached by type assertion, as the session file store's is, so Broker grows no method: 

if closer, ok := broker.(io.Closer); ok { defer closer.Close() }

The messages stay on disk; every call after Close fails.

func NewMemory 

func NewMemory(cfg MemoryConfig) (broker Broker, err error)

NewMemory returns the in-process test double: the port's full semantics, no directory, no flush, and nothing that survives the process.

It refuses the same policies NewFile refuses, through the same guard, which is what makes it an honest double.

func NewSQL 

func NewSQL(cfg SQLConfig) (broker Broker, err error)

NewSQL returns a durable broker whose queue is one table of the caller's database, created by SQLMigration. Every call runs on the transaction its context carries for cfg.Transactor, so a message published inside that transaction exists if and only if it commits. It sends no statement.

It refuses at construction — never at first use — a policy it cannot honour and a transactor, dialect or table it cannot use (SQLQueueMisconfigured). The transactor must be the SDK's, or another core/data/sql Joiner and Deferrer.

type ConsumerConfig 

type ConsumerConfig = svcqueue.ConsumerConfig

ConsumerConfig is the public alias for Consume's configuration.

type Counter added in v0.12.0 

type Counter = corequeue.Counter

Counter is the public alias for the capability of counting a queue's backlog — waiting, leased, dead, the oldest and the next due — without reading a payload. Every broker here implements it.

type DeadLetter 

type DeadLetter = corequeue.DeadLetterValue

DeadLetter is the public alias for one abandoned message and its cause.

type DeadLetterBulkManager added in v0.12.0 

type DeadLetterBulkManager = corequeue.DeadLetterBulkManager

DeadLetterBulkManager is the public alias for the capability of replaying the oldest dead letters, or deleting those that failed before an instant, in one call. Every broker here implements it.

type DeadLetterManager 

type DeadLetterManager = corequeue.DeadLetterManager

DeadLetterManager is the public alias for the capability of replaying a dead letter into its queue, its count reset, or deleting it. Every broker here implements it.

type DeadLetterPager added in v0.12.0 

type DeadLetterPager = corequeue.DeadLetterPager

DeadLetterPager is the public alias for the capability of paging through the dead letters after an identifier, oldest enqueued first. Every broker here implements it.

type DeadLetterReader 

type DeadLetterReader = corequeue.DeadLetterReader

DeadLetterReader is the public alias for the capability of reading the dead-letter store back. Every broker here implements it.

type DeadLetterWriter added in v0.5.0 

type DeadLetterWriter = corequeue.DeadLetterWriter

DeadLetterWriter is the public alias for the capability of writing a dead letter into the store as it was recorded elsewhere — what a copy of a queue from one broker to another needs. Every broker here implements it.

type DelayedNacker added in v0.10.0 

type DelayedNacker = corequeue.DelayedNacker

DelayedNacker is the public alias for the capability of handing a leased message back after a delay its caller names. Every broker here implements it, and Consume uses it for a RetryAfter failure.

type DelayedPublisher added in v0.10.0 

type DelayedPublisher = corequeue.DelayedPublisher

DelayedPublisher is the public alias for the capability of publishing a message no consumer receives before an instant. Every broker here implements it.

type Delivery 

type Delivery = corequeue.DeliveryValue

Delivery is the public alias for one message handed to one consumer, carrying the lease that proves the claim and the count that says whether this is a retry.

type FileConfig 

type FileConfig = svcqueue.FileConfig

FileConfig is the public alias for NewFile's configuration.

type Handler 

type Handler = corequeue.Handler

Handler is the public alias for the function that processes one delivery.

type Lease 

type Lease = corequeue.LeaseValue

Lease is the public alias for a consumer's exclusive claim on one message.

type LeaseExtender 

type LeaseExtender = corequeue.LeaseExtender

LeaseExtender is the public alias for the capability of renewing a lease a handler is still working under. Every broker here implements it.

type LiveMessage added in v0.7.0 

type LiveMessage = corequeue.LiveMessageValue

LiveMessage is the public alias for one live message as a broker records it.

type MemoryConfig 

type MemoryConfig = svcqueue.MemoryConfig

MemoryConfig is the public alias for NewMemory's configuration.

type Message 

type Message = corequeue.MessageValue

Message is the public alias for one unit of work as the broker minted it.

type MessageLister added in v0.12.0 

type MessageLister = corequeue.MessageLister

MessageLister is the public alias for the capability of ranging over the live messages through an iterator, as MessageReader lists them a page at a time. Every broker here implements it.

type MessageReader added in v0.7.0 

type MessageReader = corequeue.MessageReader

MessageReader is the public alias for the capability of listing the live messages — waiting and leased — without leasing one. Every broker here implements it.

type MessageWriter added in v0.7.0 

type MessageWriter = corequeue.MessageWriter

MessageWriter is the public alias for the capability of writing a live message as another broker recorded it — its count, its instants. Every broker here implements it.

type Nack 

type Nack = corequeue.NackValue

Nack is the public alias for what Broker.Nack decided: retried, or dead-lettered.

type OncePublisher added in v0.10.0 

type OncePublisher = corequeue.OncePublisher

OncePublisher is the public alias for the capability of publishing at most once per key over a window. Every broker here implements it.

type Policy 

type Policy = corequeue.PolicyValue

Policy is the public alias for the delivery discipline a broker enforces.

type Priority added in v0.12.0 

type Priority = corequeue.Priority

Priority is the public alias for the level a message is delivered at. 

const (
	// PriorityLowest is the lowest of the five levels.
	PriorityLowest Priority = corequeue.PriorityLowest
	// PriorityLow is below the normal level.
	PriorityLow Priority = corequeue.PriorityLow
	// PriorityNormal is the zero [Priority], the level [Broker].Publish publishes at.
	PriorityNormal Priority = corequeue.PriorityNormal
	// PriorityHigh is above the normal level.
	PriorityHigh Priority = corequeue.PriorityHigh
	// PriorityHighest is the highest of the five levels.
	PriorityHighest Priority = corequeue.PriorityHighest
)

The five levels a PriorityPublisher publishes at, highest received first.

type PriorityPublisher added in v0.12.0 

type PriorityPublisher = corequeue.PriorityPublisher

PriorityPublisher is the public alias for the capability of publishing a message at a level, the visible messages of the highest level received first, strictly. Every broker here implements it; the SQL broker keeps a level once SQLPriorityMigration has run and SQLConfig.Priority is set.

type Purged added in v0.12.0 

type Purged = corequeue.PurgeValue

Purged is the public alias for what Purger.Purge removed.

type Purger added in v0.12.0 

type Purger = corequeue.Purger

Purger is the public alias for the capability of removing every live message of a queue, never a dead letter. Every broker here implements it.

type Receipt 

type Receipt = corequeue.ReceiptValue

Receipt is the public alias for the opaque handle to one lease. Do not parse it and do not construct one.

type Rejecter 

type Rejecter = corequeue.Rejecter

Rejecter is the public alias for the capability of dead-lettering a leased message at once, with the cause that condemned it. Every broker here implements it, and Consume uses it for a DoNotRetry failure.

type Releaser added in v0.9.0 

type Releaser = corequeue.Releaser

Releaser is the public alias for the capability of handing a leased message back at once without counting its delivery. Every broker here implements it, and Consume uses it for the messages a stop or a spent lease kept the handler from starting, or finishing.

type SQLConfig 

type SQLConfig = svcqueue.SQLConfig

SQLConfig is the public alias for NewSQL's configuration: Transactor, Dialect and Table (required), Policy, Clock, Trail, which keeps a dead letter's trail once SQLTrailMigration has run, and Priority, which keeps a message's level once SQLPriorityMigration has run.

type Stats added in v0.12.0 

type Stats = corequeue.StatsValue

Stats is the public alias for a queue's backlog as Counter reads it.

type Wake 

type Wake = corequeue.WakeValue

Wake is the public alias for what an idle consumer waits on: a signal the next Publish or Nack in this process closes, and how long until something the broker holds becomes receivable on its own.

type Waker 

type Waker = corequeue.Waker

Waker is the public alias for the capability of telling an idle consumer when to look again. Every broker here implements it, and Consume uses it; a broker of your own that omits it is simply polled.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.