queue

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 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 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 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
• Variables
• func Consume(ctx context.Context, broker Broker, cfg ConsumerConfig) error
• func DoNotRetry(cause error) error
• func NewFile(cfg FileConfig) (broker Broker, err error)
• func NewMemory(cfg MemoryConfig) (broker Broker, err error)
• func NewSQL(cfg SQLConfig) (broker Broker, err error)
• func SQLMigration(dialect sql.Dialect, table string, version uint64) (sql.Migration, error)
• type Broker
  • func (Broker) Ack(ctx context.Context, receipt ReceiptValue) error
  • func (Broker) Nack(ctx context.Context, receipt ReceiptValue, cause error) (NackValue, error)
  • func (Broker) Publish(ctx context.Context, payload []byte) (MessageValue, error)
  • func (Broker) Receive(ctx context.Context, max int) ([]DeliveryValue, error)
• type ConsumerConfig
• type DeadLetter
• type DeadLetterManager
  • func (DeadLetterManager) DeleteDeadLetter(ctx context.Context, id string) error
  • func (DeadLetterManager) ReplayDeadLetter(ctx context.Context, id string) error
• type DeadLetterReader
  • func (DeadLetterReader) DeadLetters(ctx context.Context, max int) ([]DeadLetterValue, error)
• type Delivery
• type FileConfig
• type Handler
• type Lease
• type LeaseExtender
  • func (LeaseExtender) Extend(ctx context.Context, receipt ReceiptValue, by time.Duration) (LeaseValue, error)
• type MemoryConfig
• type Message
• type Nack
• type Policy
  • func (Policy) Normalized() PolicyValue
  • func (Policy) Validate() error
• type Receipt
• type Rejecter
  • func (Rejecter) Reject(ctx context.Context, receipt ReceiptValue, cause error) error
• type SQLConfig
• type Wake
• type Waker
  • func (Waker) Wake() WakeValue

Constants

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 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 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 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.

Variables

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 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 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 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.

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 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.

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, DeadLetterManager) 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. 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.
	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.
	BatchSize           int
	// 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 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
	// 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
}

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 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 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
	// 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 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
	// 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 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 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
}

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 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 SQLConfig

SQLConfig is the public alias for NewSQL's configuration: Transactor, Dialect and Table (required), Policy, and Clock.

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
	// 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
}

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.