queue

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