queue

package
v0.7.0 Latest Latest
Warning

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

Go to latest
Published: Sep 26, 2026 License: MIT Imports: 3 Imported by: 0

Documentation

Overview

Package queue — the ADR 0039 capability siblings of the frozen Broker port, and the dead-letter record one of them returns.

Package queue — range 0.2.23.* (ADR 0054 core/queue block).

Package queue — declares the sentinel *errs.Error port outcomes. Each var's name equals its errs.Define Reason in SCREAMING_SNAKE form.

Package queue — hosts the three values that travel between a producer, a broker and a consumer: the message, the lease, and the delivery that pairs them.

Package queue — hosts the delivery policy every broker is configured with, and the guard both implementations run so that the two refuse identical inputs identically.

Package queue declares the SDK's ASYNCHRONOUS, DURABLE message queue port: the Broker a producer publishes into and a consumer leases from, the Handler that processes one delivery, and the values that carry a message, its lease and its eventual dead letter. A core sibling admitted by ADR 0054.

queue is not events, and the line was drawn before this package existed

ADR 0053 §D1 wrote the frontier down while queue was still hypothetical, precisely so it could not be renegotiated by whoever wrote it. This package is the right-hand column, on every axis:

                     | 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
Process dies in      | the event never happened | the message is still
flight               |                          | there

A caller who wants "several things happen right now, on my goroutine, inside my transaction" wants internal/core/events. This package will not give them a faster version of it: Broker.Publish costs a disk flush, because that flush IS the guarantee.

The delivery guarantee is AT-LEAST-ONCE, and the type says so

Exactly-once delivery does not exist over a transport. What exists is at-least-once delivery plus an idempotent consumer, and every system that advertises the first is selling the second with the consumer's half left as an exercise. This port therefore promises at-least-once and puts the consequence in the signature rather than in a paragraph: DeliveryValue.Deliveries is a field of every delivery, it counts from 1, and a value greater than 1 means this consumer — or another one, in another process — has already seen these bytes.

A message is removed at ACKNOWLEDGEMENT, never at read. Broker.Receive LEASES a message: it becomes invisible to every other consumer until PolicyValue.VisibilityTimeout elapses. If the consumer acknowledges, the message is gone. If the consumer dies — SIGKILL, a power cut, an OOM — it never acknowledges, the lease lapses, and the message is delivered again with Deliveries incremented. That is the entire mechanism, and it is why the guarantee is at-least-once and not at-most-once.

What this port does NOT guarantee

  • **Exactly-once.** See above. Make the handler idempotent; the SDK's consumer engine (internal/service/queue.Consume) makes you assert that in code.
  • **Order under retry.** A queue with one consumer and no failures delivers in publication order. A retried message becomes visible again LATER than messages published after it, so a single failure reorders the stream — and more than one consumer removes ordering entirely. There is no per-key ordering here; see the ADR's Deferred section.
  • **Durability beyond what fsync actually does.** An implementation that flushes has done everything a process can do; a lying disk cache, a virtualised host that acknowledges early, or a filesystem mounted with barriers off will still lose an acknowledged write, and no library can detect that.

Layers

The two concrete brokers — one in memory, one on the filesystem — the consumer engine, and the dead-letter record live in internal/service/queue. This package owns the contract, the domain values, the policy and its guard, and the typed sentinels a refusal carries.

Package queue — the ADR 0039 sibling that lets a consumer sleep until there may be work, and the value it answers with (ADR 0104).

Index

Constants

View Source
const CodeInvalidBatchSize errs.Code = 0x00_02_17_05 // 0.2.23.5

CodeInvalidBatchSize identifies a Receive or DeadLetters asked for a non-positive number of messages.

View Source
const CodeLeaseExpired errs.Code = 0x00_02_17_04 // 0.2.23.4

CodeLeaseExpired identifies an Ack, Nack or Extend on a lease that has lapsed. The call changed nothing and the message is elsewhere.

View Source
const CodeMessageTooLarge errs.Code = 0x00_02_17_02 // 0.2.23.2

CodeMessageTooLarge identifies a Publish whose payload exceeds PolicyValue.MaxMessageBytes.

View Source
const CodeQueueMisconfigured errs.Code = 0x00_02_17_01 // 0.2.23.1

CodeQueueMisconfigured identifies a PolicyValue field a broker cannot honour, refused at construction rather than at first use.

View Source
const CodeUnknownReceipt errs.Code = 0x00_02_17_03 // 0.2.23.3

CodeUnknownReceipt identifies an Ack, Nack or Extend naming a lease this broker never issued, or a receipt it cannot parse.

View Source
const DefaultMaxMessageBytes int = 1 << 20

DefaultMaxMessageBytes bounds a payload when PolicyValue.MaxMessageBytes is left at zero. One mebibyte is the value the industry converged on — NATS's default, Kafka's default `message.max.bytes`, four times SQS's maximum — and the reason it is a CLAMP rather than a refusal is that its zero has exactly one sensible reading and the wrong answer is survivable: too small is a refused publish at the call site that made the mistake, and the alternative to any bound at all is one producer filling a disk.

View Source
const MaxDeadlineOffset time.Duration = 100 * 365 * 24 * time.Hour

MaxDeadlineOffset is the furthest ahead of now a deadline built from a policy may lie: the ceiling on PolicyValue.VisibilityTimeout and PolicyValue.RetryDelay, REFUSED above it rather than clamped.

It is not a judgement about how long a lease should last — ADR 0054 argues against even an hour. It is the representable range, with a margin. Every deadline is now plus one of these durations, and a durable broker records it as Unix nanoseconds in an int64 — that is the form a filename carries, and time.Time.UnixNano's own range — which ends on 2262-04-11. Past that instant the number wraps negative, and a name carrying it cannot be read back: the message it names is never reclaimed and never delivered, and its receipt reads UnknownReceipt. math.MaxInt64, the value somebody reaches for to mean "never", is 292 years, so it wraps a deadline set today.

A century keeps every deadline representable for any clock reading before 2162, and no lease or retry delay has a reading anywhere near it. Validate has no clock, so a fixed ceiling is the only bound it can check; an implementation that computes a deadline from its own clock at runtime — a lease extension, whose duration Validate never sees — checks the instant itself.

Variables

View Source
var (
	// QueueMisconfigured refuses a PolicyValue a broker cannot honour — a
	// zero with opposite readings, or a duration past MaxDeadlineOffset — and
	// an Extend duration that is non-positive or reaches past what a durable
	// broker can record. The "field" field names which value.
	//
	// It is a refusal and not a clamp for the fields whose zero has two
	// opposite readings — see PolicyValue's documentation, which also names
	// the two fields this sentinel deliberately does NOT guard.
	QueueMisconfigured = errs.Define(CodeQueueMisconfigured, "QUEUE_MISCONFIGURED",
		"The queue policy is not one this broker can honour",
		"core/queue: a policy field has no usable value and no default the SDK could invent; the field names which",
		errs.WithExitCode(exitConfig))

	// MessageTooLarge refuses a Publish whose payload exceeds the policy's
	// bound. It is refused at the producer, because that is the only place
	// the payload can still be made smaller — a consumer discovering it
	// cannot do anything about it, and a broker accepting it lets one
	// producer fill a device every other producer shares.
	//
	// Its fields carry the two SIZES and never the payload.
	MessageTooLarge = errs.Define(CodeMessageTooLarge, "MESSAGE_TOO_LARGE",
		"The message payload exceeds the queue's configured maximum size",
		"core/queue: payload larger than PolicyValue.MaxMessageBytes; the fields carry both sizes, never the bytes",
		errs.WithExitCode(exitConfig))

	// UnknownReceipt refuses an acknowledgement naming a lease this broker
	// never issued, or one whose encoding it cannot read.
	//
	// It is distinct from LeaseExpired on purpose. "This was yours and you
	// were too slow" and "this was never yours" are different bugs: the
	// first is a visibility timeout too short for the handler, the second is
	// a receipt built by hand, kept across a restart of an in-memory broker,
	// or handed to the wrong queue. Collapsing them would send every reader
	// looking for the wrong one half the time.
	UnknownReceipt = errs.Define(CodeUnknownReceipt, "UNKNOWN_RECEIPT",
		"The receipt does not name a lease this queue issued",
		"core/queue: receipt unparseable or never issued by this broker; the field carries no payload",
		errs.WithExitCode(exitConfig))

	// LeaseExpired reports an Ack, Nack or Extend whose lease had already
	// lapsed. NOTHING happened: the message is back in the queue or in the
	// dead-letter store, and another consumer may already hold it.
	//
	// Reporting it beats returning nil, for the reason core/lock gives about
	// releasing a lock one no longer holds: 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 consumer that gets this has produced a duplicate — which the
	// at-least-once guarantee permits — and the one thing it must not do is
	// believe the work is finished.
	LeaseExpired = errs.Define(CodeLeaseExpired, "LEASE_EXPIRED",
		"The lease on that message has expired and the acknowledgement was ignored",
		"core/queue: visibility timeout elapsed before Ack/Nack/Extend; the message is queued again or dead-lettered",
		errs.WithExitCode(exitTempFail))

	// InvalidBatchSize refuses a Receive or DeadLetters asked for a
	// non-positive count.
	//
	// Returning an empty slice instead would be ADR 0031's inert policy: a
	// consumer loop asking for zero messages spins forever, processes
	// nothing, reports no error, and looks exactly like an idle queue.
	InvalidBatchSize = errs.Define(CodeInvalidBatchSize, "INVALID_BATCH_SIZE",
		"A batch size must be positive",
		"core/queue: Receive/DeadLetters called with max <= 0, which would poll forever and deliver nothing",
		errs.WithExitCode(exitConfig))
)

Functions

This section is empty.

Types

type Broker

type Broker interface {
	// 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.
	//
	// 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/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)
	// 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/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], or — when it has
	// now been delivered [PolicyValue.MaxDeliveries] times — it is moved to
	// the dead-letter store together with cause. See [NackValue].
	//
	// 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)
}

Broker is one named queue: producers publish into it, consumers lease from it, and it — not the consumer — owns the delivery count and the decision to dead-letter. Implementations MUST be safe for concurrent use by multiple goroutines AND, where the implementation is durable, by multiple processes.

IFACE-PLUGIN: the concrete brokers stay unexported behind their constructors in internal/service/queue.

The interface is FROZEN at these four methods. pkg/v1/queue aliases it, Go interfaces are structural, and a fifth method would break every downstream implementation at compile time with no deprecation window (ADR 0039). A new capability arrives as a sibling interface reached by type assertion — DeadLetterReader and LeaseExtender are the two that ship.

Why the BROKER owns the delivery count

Because the consumer is the thing that dies. A count held by the consumer engine is lost by exactly the event the count exists to survive, so a process that crashes on every third message would retry forever and never dead-letter. The count travels with the message, in whatever the implementation's durable representation is, and the broker compares it against PolicyValue.MaxDeliveries on every Broker.Nack and on every lapsed lease.

type DeadLetterReader

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.
	//
	// max must be positive ([InvalidBatchSize]).
	DeadLetters(ctx context.Context, max int) ([]DeadLetterValue, error)
}

DeadLetterReader reads the messages a broker gave up on.

It is a SIBLING of Broker reached by type assertion, never a fifth method on it, because reading the dead letters back is a capability and not every broker has one: a connector onto a third-party system usually dead-letters into a second queue that is read with the same Broker interface, and one that dead-letters into an operator's console cannot return anything at all. Both brokers in internal/service/queue implement it.

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

type DeadLetterValue

type DeadLetterValue 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, 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.
	Deliveries int
}

DeadLetterValue is one message the broker gave up on, together with why.

A dead letter without its cause is an investigation nobody can run, so the cause is part of the record rather than a line in whatever log has since rotated away. What is kept of the cause is CLAUDE.md rule 4's split, and the split is deliberate: the wire-safe Public half and the dotted-quad code go to the store, the Private half stays in the log of the process that failed, and MessageValue.ID joins the two. A dead-letter store is read by whoever is investigating — frequently not the process, the host or the trust domain that produced the failure — so it is the wrong place for the half of an error the SDK defines as log-only.

type DeliveryValue

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

DeliveryValue is one message handed to one consumer, with everything the consumer needs to decide what to do about it.

type Handler

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

Handler processes one delivery. It is the port a consumer implements, and it is a FUNCTION type rather than an interface — the shape internal/core/CLAUDE.md already admits for resilience.Operation, scheduler.Job, lifecycle.Start and events.Listener. ADR 0039's rule is that a published port must not grow a method; a func type satisfies it structurally, because it cannot grow one at all.

It receives the whole DeliveryValue and not just the payload, so that DeliveryValue.Deliveries is in front of whoever writes the handler. A handler WILL be called more than once for the same message — that is the guarantee, not a malfunction — and the count is the only thing that lets it say so in a log.

Returning nil acknowledges. Returning an error nacks: the message is retried, or dead-lettered once it has exhausted PolicyValue.MaxDeliveries. ctx is the consumer's and IS honoured: unlike events.Listener, a handler that outlives its context is holding a lease it is about to lose, so abandoning the work is the correct response and the message will simply be redelivered.

type LeaseExtender

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

LeaseExtender renews a lease a handler is still working under.

It is the honest answer to the one question PolicyValue.VisibilityTimeout cannot settle: the timeout must be longer than the slowest handler, and the slowest handler is not known in advance. Rather than inviting a caller to set an hour-long visibility timeout — which turns every consumer crash into an hour of silence — a handler that legitimately needs longer says so.

It is a sibling for the same reason DeadLetterReader is: a broker over a transport with no renewal verb cannot implement it, and the assertion is how a caller finds that out. Both brokers in internal/service/queue implement it.

type LeaseValue

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

LeaseValue is a consumer's exclusive claim on one message: the receipt that proves it and the instant it lapses.

The two travel together because neither is usable alone. A receipt without a deadline cannot tell a long-running handler whether it still owns the work; a deadline without a receipt cannot acknowledge anything.

type MessageValue

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

MessageValue is one unit of work as the broker minted it. Every field is assigned by Broker.Publish; a caller never fills one in, which is why Publish takes a payload rather than one of these.

type NackValue

type NackValue 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.
	DeadLettered bool
}

NackValue is what Broker.Nack decided.

It exists because "I reported the failure" and "that message is now dead" are different facts, and a consumer that cannot tell them apart cannot log the one that matters. Every field is a scalar: a nack is on the failure path, but it is on the failure path of a queue that is expected to retry, so it is not rare enough to allocate on.

type PolicyValue

type PolicyValue 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]).
	RetryDelay 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
}

PolicyValue is the delivery discipline a broker enforces: how long a lease lasts, how many attempts a message gets, how long a failure waits, and how large a payload may be.

Its zero value is REFUSED, and that is the decision

ADR 0031 gives two branches — clamp where a working default needs no explanation, refuse where any value the SDK invented would be arbitrary — and this one struct uses both, which is why they are worth reading side by side.

  • PolicyValue.VisibilityTimeout and PolicyValue.MaxDeliveries are REFUSED at zero. Each has two natural readings that are OPPOSITES, and in each case one of them silently destroys the guarantee: a zero visibility timeout is either "redeliver immediately", which is a delivery storm, or "never redeliver", which is a lost message; zero deliveries is either "unlimited", which lets one poison message loop forever and the dead-letter store stay empty, or "none", which is a queue that delivers nothing. This is core/lock's argument for refusing a zero TTL, and it is the same argument because it is the same class of value: a lifetime belongs to the work being protected, and only the caller knows what that is.
  • PolicyValue.RetryDelay and PolicyValue.MaxMessageBytes are CLAMPED. A zero retry delay means "as soon as the lease lapses", which is one reading, is harmless, and is what a caller who has not thought about backoff wants. A zero size bound means DefaultMaxMessageBytes.

Both branches in one struct is the point. A domain in which every zero is refused teaches a reader nothing except that the author was nervous.

The two durations are bounded from ABOVE as well, by MaxDeadlineOffset, and that refusal is about arithmetic rather than about readings: a deadline further out than that cannot be recorded by the durable broker at all.

func (PolicyValue) Normalized

func (p PolicyValue) 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 (PolicyValue) Validate

func (p PolicyValue) 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 broker and the file broker refuse identical inputs identically — the property that makes the first usable as a test double for the second. It is the same instrument core/vfs's ValidatePath and ValidatePerm are, for the same reason.

type ReceiptValue

type ReceiptValue string

ReceiptValue is the handle to ONE lease: the only thing Broker.Ack, Broker.Nack and LeaseExtender.Extend accept, and the only proof that this consumer is the one currently holding the message.

It is opaque and it is NOT the message identifier. Two deliveries of the same message carry the same MessageValue.ID and different receipts, which is what lets a broker refuse an acknowledgement from a consumer whose lease has already lapsed and been taken by somebody else. Do not parse it, do not construct one, and do not store one past the lease it names.

It is a defined string rather than a struct with an unexported field because a receipt has to survive being handed to a goroutine, put in a map and printed in a log, and because it is not a secret: it names a lease on a queue the holder can already read. What it must not be is FORGEABLE into somebody else's lease, and that is a property of how the implementation mints it, not of the type.

type WakeValue added in v0.6.0

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

WakeValue is what an idle consumer waits on: an event, or an instant the broker already knows about.

type Waker added in v0.6.0

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
}

Waker tells an idle consumer when to look at a broker again, so it can sleep until there may be work instead of polling on a fixed cadence.

It is a SIBLING of Broker for the reason DeadLetterReader is: waking is a capability, and a broker over a transport that cannot see its own publications — or one whose state lives in another process — has no honest answer to give. The consumer engine asks by type assertion and polls when the answer is no, so an implementation that omits it loses latency, never correctness.

A wake is a HINT, in both directions. A signal may arrive with nothing to receive — another consumer took it — and work may appear with no signal at all: a durable broker cannot see a publication made by another process. So a consumer that uses it still polls, at whatever cadence bounds the second case, and treats the first as a cheap empty Receive.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL