Documentation
¶
Overview ¶
Package mutationlog is the single source of truth for the stream of graph mutations that flows through Lantern's leaderless full-replica replication design.
The log is an in-memory, append-only ring buffer. Each Entry carries a monotonically increasing [Seq] (assigned atomically by Log.Append), an hlc.Timestamp supplied by the caller, and an opaque MutationOp payload. Keeping the payload as an interface keeps this package proto-free so it can be reused both by internal replication (#181, #184) and by external CDC (#180) without a circular dependency on pb/.
A bounded number of entries are retained for replay. When a subscriber requests a Seq older than Log.FirstSeq that entry has already been evicted, so Log.Subscribe returns ErrGapped and the caller is then expected to snapshot and resubscribe (RFC §7). The replay window is delivered ahead of live traffic without being bounded by the subscriber buffer; only the live tail shares a bounded outbound channel, so a slow consumer of *live* entries exerts back-pressure rather than allowing the log to drift forward without it (see Log.Subscribe for the #812 rationale).
A WAL hook lets a future durability layer (RFC D1) intercept appends before they fan out. The default NopWAL is a no-op, matching today's in-memory-only behaviour.
This package depends only on the standard library and core/hlc.
Index ¶
- Constants
- Variables
- type Entry
- type Log
- func (l *Log) Append(op MutationOp, ts hlc.Timestamp, stampers ...SeqStamper) (Entry, error)
- func (l *Log) Cap() int
- func (l *Log) Close() error
- func (l *Log) Evicted() uint64
- func (l *Log) FirstSeq() (uint64, bool)
- func (l *Log) LastSeq() (uint64, bool)
- func (l *Log) Len() int
- func (l *Log) Subscribe(fromSeq uint64) (<-chan Entry, func() error, error)
- type MutationOp
- type NopWAL
- type Options
- type SeqStamper
- type WAL
Constants ¶
const ( // DropCauseBufferFull is reported when the dispatcher's non-blocking // send to a subscriber's outbound channel fails because the channel // is full. The subscriber is marked gapped, its channel is closed, // and it is unregistered from the log. DropCauseBufferFull = "buffer_full" )
Drop-cause labels passed to Options.OnDrop. New causes may be added in future revisions; consumers should treat unknown values as opaque strings rather than enumerate (#260).
Variables ¶
var ErrClosed = errors.New("mutationlog: log is closed")
ErrClosed is returned by Log.Append and Log.Subscribe after Log.Close has been called.
var ErrGapped = errors.New("mutationlog: requested sequence has been evicted")
ErrGapped is returned to a subscriber when its requested fromSeq has already been evicted from the ring buffer. Callers must snapshot and resubscribe from a fresh sequence number.
Functions ¶
This section is empty.
Types ¶
type Entry ¶
type Entry struct {
// Seq is the log-local position of the entry. It increases by exactly
// one with each successful [Log.Append] call within a single [Log].
Seq uint64
// HLC is the Hybrid Logical Clock stamp supplied by the caller at
// append time. Different origins may produce entries whose HLC ordering
// disagrees with the local Seq ordering; that is by design.
HLC hlc.Timestamp
// Op is the opaque mutation payload.
Op MutationOp
}
Entry is a single durable record in the log.
type Log ¶
type Log struct {
// contains filtered or unexported fields
}
Log is an append-only, bounded, in-memory mutation log.
The zero value is not ready for use; construct with New.
Locking model (#260, #745): two mutexes split the hot path so Log.Append no longer iterates subscribers under its own critical section.
- mu is an RWMutex protecting the ring buffer, sequence counters, and the closed flag. The mutation paths take the write lock: Append (WAL write + store + seq update + handoff to the dispatcher), Subscribe (replay + sub registration), and Close. The pure-read status methods (FirstSeq, LastSeq, Len, Evicted) take only the read lock, so a burst of status polling no longer serializes against itself and proceeds concurrently whenever no append/subscribe is mid-flight (#745).
- subsMu protects the subscribers map and per-subscriber state. Held by the dispatcher goroutine during fan-out and by cancel.
Lock order when both are taken: mu → subsMu. The dispatcher only takes subsMu, and Append only takes mu (the channel send happens under mu to preserve global Seq ordering), so the two paths cannot deadlock.
func New ¶
New constructs a Log with the given options. It also starts a background dispatcher goroutine; call Log.Close to stop it.
func (*Log) Append ¶
func (l *Log) Append(op MutationOp, ts hlc.Timestamp, stampers ...SeqStamper) (Entry, error)
Append assigns the next sequence number to op, persists the entry through the WAL hook, stores it in the ring buffer, and hands it off to the dispatcher goroutine which performs per-subscriber fan-out. The returned Entry carries the assigned Seq.
Append is safe for concurrent use; calls are serialised so Seq strictly increases by one for each successful return. The hand-off to the dispatcher happens under the log mutex so the dispatcher observes entries in strict Seq order (#260); under sustained overload Append will block briefly waiting for the dispatcher to drain its inbox.
The optional last argument is a SeqStamper: when non-nil it is invoked with the assigned seq while Log still holds its lock and before the entry is placed into the ring or handed to the dispatcher. This is the seam that lets producers stamp the originating-writer's seq onto the in-flight payload (typically `pb.Mutation.Seq`) without racing the dispatcher, which would otherwise read the payload concurrently as it fans out to subscribers. The leaderless Subscribe contract (#415) keys per-hop dedup on (origin, origin_seq), so this stamping is mandatory for any payload that re-enters the system via Subscribe relay.
func (*Log) Close ¶
Close stops accepting appends, signals the dispatcher to drain, and waits for it to exit. The dispatcher closes every still-live subscriber channel on its way out. Calling Close more than once returns nil.
func (*Log) Evicted ¶
Evicted returns the cumulative count of entries dropped from the ring buffer because Append at full capacity displaced the oldest entry. The counter is monotonic for the lifetime of the Log.
func (*Log) FirstSeq ¶
FirstSeq returns the lowest Seq still resident in the ring buffer. It returns 0 (and false) when the log is empty.
func (*Log) LastSeq ¶
LastSeq returns the Seq of the most recently appended entry. It returns 0 (and false) when the log is empty.
func (*Log) Len ¶
Len returns the number of entries currently resident in the ring buffer. 0 <= Len() <= Cap().
func (*Log) Subscribe ¶
Subscribe returns a channel that receives entries starting at fromSeq. Any entries with Seq >= fromSeq still resident in the ring buffer are replayed first, in order, followed by live entries.
If fromSeq is older than Log.FirstSeq the entry has already been evicted and Subscribe returns ErrGapped; the caller must snapshot and resubscribe from a fresh sequence number.
The returned cancel func unregisters the subscriber and closes the channel. It is safe to call cancel more than once; subsequent calls return nil.
Replay vs. back-pressure (#812): the replay window is delivered by a dedicated forwarder goroutine with a blocking send, so a catch-up larger than Options.SubscriberBuffer is NOT mistaken for a slow subscriber — it merely back-pressures the producer until the consumer drains it. SubscriberBuffer continues to bound only the *live* tail: once caught up, a consumer that then falls behind by more than SubscriberBuffer entries is marked gapped, its channel is closed, and it must snapshot and resubscribe. Before #812 the replay was loaded straight into the SubscriberBuffer-bounded channel and Subscribe returned ErrGapped the instant the retained ring exceeded SubscriberBuffer, which deterministically broke replication to any peer whose log had grown past that bound.
type MutationOp ¶
type MutationOp any
MutationOp is an opaque payload carried by an Entry. The mutationlog package never inspects it; callers (proto encoders, applicators, CDC emitters) own the concrete types.
type Options ¶
type Options struct {
// Capacity is the maximum number of entries retained in the ring buffer
// for replay. Must be > 0; defaults to 1024 when zero.
Capacity int
// SubscriberBuffer is the per-subscriber outbound channel size. Smaller
// values increase back-pressure sensitivity; defaults to 512 when zero.
//
// At sustained write rates a too-small buffer turns transient scheduling
// jitter into permanent gap-closes: the fan-out path uses a
// non-blocking send and closes the subscriber on a full channel
// (see [Log.Append]). 512 gives ~256 ms of headroom at 2k writes/s,
// which is well above typical scheduler stalls on a loaded host.
SubscriberBuffer int
// WAL receives every appended entry before it fans out to subscribers.
// Nil defaults to [NopWAL].
WAL WAL
// OnDrop, when non-nil, is invoked synchronously from the dispatcher
// goroutine each time a live fan-out drops an entry to a subscriber.
// The cause argument is one of the DropCause* constants in this
// package. Implementations must not block (they run on the hot
// fan-out path) and must not call back into the [Log] — the
// dispatcher holds the subscriber mutex at the call site (#260).
OnDrop func(cause string)
}
Options configures a Log.
A zero Options is valid: Capacity defaults to 1024 entries and WAL defaults to NopWAL. SubscriberBuffer defaults to 512.
type SeqStamper ¶ added in v0.3.0
type SeqStamper func(seq uint64)
SeqStamper is invoked synchronously by Log.Append with the freshly-assigned seq while the log mutex is still held. Implementers typically write the seq onto a field of the in-flight payload so that subsequent readers (notably the dispatcher fanning the entry out to subscribers) observe a payload whose own seq matches the log entry's seq. The stamper MUST NOT block and MUST NOT call back into the Log.
type WAL ¶
WAL is the hook surface for a future write-ahead-log implementation. Implementations must be safe for concurrent use. Log.Append calls WAL.Write while holding the log mutex, so implementations should keep the call cheap (a buffered write is fine; a synchronous fsync is not).
Source Files