Documentation
¶
Overview ¶
Package pubsub is an in-process publish/subscribe primitive for the Lantern core. It owns message lifecycles (Publish → enqueue → consumer → Ack/salvage) and the FullPolicy semantics that govern channel saturation.
Observability ¶
Subscriptions emit telemetry through the Observer interface. core/ is a leaf module and never imports server/metrics (see AGENTS.md "Dependency boundaries"); the server-side Prometheus collectors are plugged in per subscription via WithObserver(...). The default is a no-op so library consumers pay zero overhead.
Authors adding new drop or dispatch paths must invoke the matching Observer method exactly once per event and, when a new drop policy is introduced, extend the DropPolicies slice so server-side label pre-warming stays exhaustive.
Index ¶
- Constants
- Variables
- type FullPolicy
- type Message
- type Observer
- type Subscription
- type SubscriptionOption
- func WithBufferSize(n int) SubscriptionOption
- func WithConcurrency(n int) SubscriptionOption
- func WithFullPolicy(p FullPolicy) SubscriptionOption
- func WithObserver(o Observer) SubscriptionOption
- func WithSalvageInterval(d time.Duration) SubscriptionOption
- func WithTTL(d time.Duration) SubscriptionOption
- type Topic
- func (t *Topic[T]) Name() string
- func (t *Topic[T]) NewSubscription(name string, concurrency int, interval time.Duration, ttl time.Duration) *Subscription[T]
- func (t *Topic[T]) NewSubscriptionWithOptions(name string, opts ...SubscriptionOption) *Subscription[T]
- func (t *Topic[T]) Publish(body T)
- func (t *Topic[T]) Subscriptions() map[string]*Subscription[T]
Constants ¶
const ( DropPolicyNewest = "drop_newest" DropPolicyOldest = "drop_oldest" DropPolicyNewestAfterOldest = "drop_newest_after_oldest" )
Drop policy labels passed to Observer.RecordDrop. Exposed as a bounded set so server-side collectors can pre-warm the label rows and dashboards render the full series from process start.
Variables ¶
var DropPolicies = []string{ DropPolicyNewest, DropPolicyOldest, DropPolicyNewestAfterOldest, }
DropPolicies is the canonical ordered list of policy labels emitted by RecordDrop. Useful for pre-warming Prometheus CounterVec rows.
Functions ¶
This section is empty.
Types ¶
type FullPolicy ¶ added in v0.1.1
type FullPolicy int
FullPolicy selects the behavior of Publish when a Subscription's delivery channel is full. The default, FullPolicyBlock, preserves the historical contract: Publish blocks until a worker drains the channel. DropNewest and DropOldest trade strict delivery for liveness and are observable via a slog.Warn line on every drop.
const ( // FullPolicyBlock blocks Publish until the channel has room (default). FullPolicyBlock FullPolicy = iota // FullPolicyDropNewest discards the incoming id when the channel is full. // The Message envelope is acked and returned to the pool immediately so // no slot is wasted in the in-flight map. FullPolicyDropNewest // FullPolicyDropOldest discards the oldest queued id and enqueues the // new one. The dropped Message is acked and returned to the pool. FullPolicyDropOldest )
type Message ¶
type Message[T any] struct { // contains filtered or unexported fields }
func (*Message[T]) Ack ¶
func (m *Message[T]) Ack()
Ack marks the message as successfully processed and removes it from the subscription's in-flight set, so the salvage path will no longer try to redeliver it.
After Ack (or Nack), the receiver must not be retained: the *Message[T] is returned to a sync.Pool and may be reused by a later Publish (#231). Read any fields you need before calling Ack/Nack.
func (*Message[T]) ID ¶
ID returns the message's per-subscription monotonic identifier. IDs are unique within a single *Subscription but not across subscriptions or across process restarts (#231 — replaced the prior UUID string for hot-path cost).
func (*Message[T]) LastViewedAt ¶
func (*Message[T]) Nack ¶
func (m *Message[T]) Nack()
Nack signals that processing failed and the message should be retried. The message is re-queued immediately on the subscription channel; the salvage timer is not involved.
After Nack returns the receiver may be re-dispatched concurrently to another worker. Do not retain it. See Ack for the pooling contract.
type Observer ¶ added in v0.1.1
type Observer interface {
RecordEnqueueDepth(depth int)
RecordDrop(policy string)
ObserveDispatch(d time.Duration)
}
Observer is the optional sink for pubsub subscription telemetry. core/ is a leaf module and must not import server/metrics (AGENTS.md dependency boundary), so the server-side Prometheus collectors are injected via this interface. Implementations must be safe for concurrent use; pubsub may call any method from any worker goroutine. A nil observer is treated as the no-op.
Method semantics:
- RecordEnqueueDepth is invoked once per successful enqueue with the channel length after the send. Sampled rather than gauged so the collector can aggregate (histogram) without per-subscription cardinality.
- RecordDrop is invoked once per drop-path execution with one of the policy strings exported below. Each drop must increment exactly once, even when the drop-oldest fallback also drops the newest message (that case fires RecordDrop("drop_oldest") followed by RecordDrop("drop_newest_after_oldest")).
- ObserveDispatch is invoked once per consumer return with the wall- clock duration from the originating Publish (message.createdAt) to the moment the consumer callback returned.
type Subscription ¶
type Subscription[T any] struct { // contains filtered or unexported fields }
func (*Subscription[T]) Name ¶
func (s *Subscription[T]) Name() string
func (*Subscription[T]) Subscribe ¶
Subscribe drains the subscription queue and dispatches messages to consumer until ctx is cancelled. It is single-flight per *Subscription: concurrent or reentrant calls return immediately without dispatching. After Subscribe returns (ctx done) the subscription may be Subscribe'd to again.
Concurrency is enforced by a fixed worker pool of size s.concurrency that drains s.ch directly (#231). The prior model of one goroutine per message, gated by a sync semaphore, allocated a goroutine and paid a sync.Cond wake-up on every Publish; the worker pool amortises both costs to zero.
func (*Subscription[T]) Topic ¶
func (s *Subscription[T]) Topic() *Topic[T]
type SubscriptionOption ¶ added in v0.1.1
type SubscriptionOption func(*subscriptionConfig)
SubscriptionOption configures a Subscription created via Topic.NewSubscriptionWithOptions. Options compose: later options overwrite earlier ones.
func WithBufferSize ¶ added in v0.1.1
func WithBufferSize(n int) SubscriptionOption
WithBufferSize sets the capacity of the delivery channel. Values < 1 are clamped to 1 to keep the channel buffered (an unbuffered channel would serialise every Publish against every worker).
func WithConcurrency ¶ added in v0.1.1
func WithConcurrency(n int) SubscriptionOption
WithConcurrency sets the number of worker goroutines that Subscribe spawns to drain the delivery channel. Values < 1 are clamped to 1 at Subscribe time. The historical positional constructor maps its concurrency argument to this option.
func WithFullPolicy ¶ added in v0.1.1
func WithFullPolicy(p FullPolicy) SubscriptionOption
WithFullPolicy selects the behavior of Publish when the delivery channel is full. See FullPolicy for the available modes.
func WithObserver ¶ added in v0.1.1
func WithObserver(o Observer) SubscriptionOption
WithObserver installs an Observer that receives enqueue-depth samples, drop counters, and dispatch-duration observations for this subscription. A nil observer reverts to the no-op (no telemetry). core/ never imports server/metrics directly; the server wires a Prometheus-backed Observer here (#240).
func WithSalvageInterval ¶ added in v0.1.1
func WithSalvageInterval(d time.Duration) SubscriptionOption
WithSalvageInterval sets the period of the watch ticker that drives expiry eviction and re-delivery reminders.
func WithTTL ¶ added in v0.1.1
func WithTTL(d time.Duration) SubscriptionOption
WithTTL sets the maximum age of an in-flight Message before salvage evicts it (acks on the consumer's behalf and returns the envelope to the pool).
type Topic ¶
type Topic[T any] struct { // contains filtered or unexported fields }
func (*Topic[T]) NewSubscription ¶
func (t *Topic[T]) NewSubscription(name string, concurrency int, interval time.Duration, ttl time.Duration) *Subscription[T]
NewSubscription creates (or returns the existing) Subscription with the historical positional arguments. New code should prefer NewSubscriptionWithOptions which exposes buffer size and full-policy knobs; this wrapper keeps existing call sites compiling.
func (*Topic[T]) NewSubscriptionWithOptions ¶ added in v0.1.1
func (t *Topic[T]) NewSubscriptionWithOptions(name string, opts ...SubscriptionOption) *Subscription[T]
NewSubscriptionWithOptions creates (or returns the existing) Subscription configured via functional options. Defaults match the positional NewSubscription wrapper: concurrency 1, interval/ttl 1 minute, buffer 65536, FullPolicyBlock. If a Subscription with name already exists it is returned unchanged; options on a second call are ignored.
func (*Topic[T]) Publish ¶
func (t *Topic[T]) Publish(body T)
Publish performs best-effort concurrent fan-out: each subscription receives the message on its own goroutine, so a single slow or saturated subscriber cannot block delivery to its peers (#230). Delivery order across subscriptions is unspecified, matching the prior map-iteration behavior.
Publish returns immediately after scheduling the fan-out goroutines; it does not wait for each subscription's channel send to complete. This trades synchronous backpressure (which no caller currently observes) for independence between subscribers. The pattern is intended for topics with tens of subscribers; if you need to fan out to thousands, prefer a long-lived per-subscription delivery goroutine instead.
func (*Topic[T]) Subscriptions ¶
func (t *Topic[T]) Subscriptions() map[string]*Subscription[T]
Subscriptions returns a snapshot of the topic's subscriptions. The returned map is a clone, safe for callers to iterate without holding the topic lock.
Source Files
Directories