Documentation
¶
Overview ¶
Package changefeed reads a database's change history for the two features built on it: triggers (a dispatcher with a durable position) and realtime (consumers that open when a client subscribes). It is driver-neutral: a Source (internal/mongodb) opens the streams; this package owns what is shared — the vocabulary of a change, the watched set, resuming after errors, history that is gone, and the lazy open/close of a consumer.
The design is design/triggers-and-realtime.md §3 and §4.
Index ¶
Constants ¶
const ( ReasonSetChanged = "set_changed" ReasonError = "error" ReasonGap = "history_lost" )
Reopen reasons, as the label of the reopen metric.
const DefaultGrace = 30 * time.Second
DefaultGrace is how long a Lazy consumer stays open after its last subscriber left.
Variables ¶
var ErrHistoryLost = errors.New("changefeed: the change history no longer reaches the position")
ErrHistoryLost means the history no longer reaches the position asked for: events between it and now are gone.
Functions ¶
This section is empty.
Types ¶
type Change ¶
type Change struct {
Realm string
DB string
Coll string
System bool // a change of a system collection: only Key, Op and Fields are filled
Op Op
Key string // the document's id
// At is the cluster time of the change, to the second; Increment orders the changes of
// one second. Together they identify the change within the history.
At time.Time
Increment uint32
// Version is the document's `_meta.version` after the change (before it, for a removal);
// 0 when the images are not available.
Version int64
// Before and After are the document before and after, in API shape; nil when the image is
// not available (a created document has no Before, a removed one has no After; a missing
// image on an update or removal is the case that must fail closed).
Before, After map[string]any
// Fields are the names of the fields an update reported as changed (dot paths); empty for
// creations, removals and replacements.
Fields []string
// Replace is true when the change was a replacement of the whole document.
Replace bool
}
Change is one change of one document.
type Consumer ¶
type Consumer struct {
// Name labels the consumer's metrics and logs ("realtime", "dispatcher").
Name string
Source Source
Log *slog.Logger
Observer Observer
// Backoff is the wait before the nth consecutive reopen after an error; nil is 100ms
// doubling to 5s.
Backoff func(attempt int) time.Duration
// contains filtered or unexported fields
}
Consumer reads one stream over a watched set and hands what it reads to a Handler. It resumes by itself: after an error from the position it last reached, after a change of the watched set (a configuration reload) from the same position, and — when the history no longer reaches the position — from now, reporting a Gap.
func (*Consumer) Run ¶
Run reads until ctx is done, starting after the given position (nil: from now). It returns the last position it reached, which a caller that keeps one durably stores.
func (*Consumer) SetWatched ¶
SetWatched replaces the watched set. A running consumer reopens its stream after the last position it reached; one with nothing watched waits.
type Event ¶
Event is what a stream hands back: a Change, or — with Change nil — only the news that the position moved with nothing to deliver (a quiet period).
type Gap ¶
type Gap struct {
// From is the position that could not be resumed; the consumer continues from now.
From Position
At time.Time
}
Gap reports ErrHistoryLost to a handler.
type Handler ¶
type Handler interface {
Change(ctx context.Context, c Change, at Position)
// Idle reports a position moved with nothing delivered.
Idle(ctx context.Context, at Position)
Gap(ctx context.Context, g Gap)
}
Handler receives what a consumer reads. Calls are sequential; a slow handler slows the consumer, so handlers that can't keep up must buffer or drop on their own.
type Lazy ¶
type Lazy struct {
// contains filtered or unexported fields
}
Lazy shares one Consumer between subscribers and keeps it closed while there are none: the stream opens, from now, when the first subscriber arrives, and closes after Grace once the last one has left. An instance with no subscribers therefore reads nothing (design/triggers-and-realtime.md §3).
func (*Lazy) Acquire ¶
Acquire registers a subscriber, opening the stream if it is the first. The returned function removes it; calling it more than once is harmless.
func (*Lazy) Close ¶
func (l *Lazy) Close()
Close stops the stream whatever the subscribers; later Acquire calls do nothing.
func (*Lazy) SetWatched ¶
SetWatched replaces what the consumer reads, from the configuration in force. It takes effect at once on an open stream (which resumes after its position) and otherwise at the next open.
func (*Lazy) Subscribers ¶
Subscribers is how many handlers are registered.
type Observer ¶
type Observer interface {
FeedState(consumer string, open bool, watched int)
FeedReopened(consumer, reason string)
FeedEvent(consumer string, op Op, system bool)
FeedLag(consumer string, lag time.Duration)
}
Observer receives the reader-level metrics. internal/metrics implements it; every method may be called on a nil receiver there.
type Position ¶
type Position []byte
Position is an opaque place in the change history. A stream resumed at it continues exactly after it.
type Stream ¶
type Stream interface {
// Next blocks until there is a change, the position moves, or ctx is done.
Next(ctx context.Context) (Event, error)
// Position is where the stream is now.
Position() Position
Close()
}
Stream is one open read of the change history.
type WatchSet ¶
type WatchSet struct {
// Collections are application collections, read with their pre- and post-change images.
Collections []Namespace
// System are backd's own collections. They are projected where the data is read, down to
// document keys and the names of the fields that changed — never values — so secrets
// never leave the database.
System []Namespace
}
WatchSet is what a consumer reads.
Source Files