Documentation
¶
Overview ¶
Package replicate is the transport-neutral core for replicating a db table/archive's change stream into another store: an in-memory Change (a db.WatchEvent enriched with replication metadata) and an idempotent Sink that applies Changes into a *db.DB or *db.ArchiveTable. It depends only on classad + db (no HTTP, no CEDAR), so every transport builds on it: the HTTP/SSE feed (classad/changefeed) and a dbrpc/CEDAR replicator both convert their wire form to a Change and feed the same Sink.
Index ¶
Constants ¶
const SrcAttr = "Src"
SrcAttr is the attribute the sinks stamp with a change's source identity, so a fanned-in target can be queried/deduped by source.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Change ¶
type Change struct {
Kind Kind
Src string // source identity (for fan-in stamping / dedup)
Ver uint64 // strictly monotonic per-source version (idempotency key)
Key string // source storage key
Ad *classad.ClassAd // upsert only
Cursor []byte // opaque resume token (the db.Watch cursor); client-side resume only
// TS is the record's retention timestamp (unix millis), source-stamped from the archive's age
// attribute. It is the COMPARABLE ack watermark: a sink acks the max TS it has durably
// persisted, and the source's GC floor is min(ack TS) over live subscribers -- which maps onto
// the archive's time-based retention. 0 when the source has no age attribute (GC gating off).
TS int64
// For KindGap only: the estimated lost time window (unix millis).
FromMillis, ToMillis int64
}
Change is one change as the sink sees it, in memory. Ad is set only on an upsert.
func ChangeFromWatch ¶
ChangeFromWatch maps a db.WatchEvent to a Change, stamping src and a caller-assigned monotonic version. ok is false for a WatchResync (the transport loop handles reconnect, not the sink).
type CursorStore ¶
type CursorStore interface {
Load() ([]byte, error) // nil, nil = none stored yet
Save(cursor []byte) error
}
CursorStore persists a sink's resume cursor across restarts.
type FileCursorStore ¶
type FileCursorStore struct{ Path string }
FileCursorStore persists the cursor to a file (atomic replace).
func (FileCursorStore) Load ¶
func (f FileCursorStore) Load() ([]byte, error)
func (FileCursorStore) Save ¶
func (f FileCursorStore) Save(c []byte) error
type Kind ¶
type Kind string
Kind classifies a change. It mirrors db.WatchKind 1:1, plus an advisory Gap.
const ( KindUpsert Kind = "upsert" // Key was added/updated; Ad holds the new value. KindDelete Kind = "delete" // Key was removed; Ad is nil. KindReset Kind = "reset" // discard derived state for Src; a fresh snapshot follows. KindSynced Kind = "synced" // caught up to head; Cursor is a safe resume/ack point. KindGap Kind = "gap" // advisory: records were lost to source retention (see From/To). )
type MemCursorStore ¶
type MemCursorStore struct {
// contains filtered or unexported fields
}
MemCursorStore is an in-memory CursorStore (tests, or an ephemeral sink).
func (*MemCursorStore) Load ¶
func (m *MemCursorStore) Load() ([]byte, error)
func (*MemCursorStore) Save ¶
func (m *MemCursorStore) Save(c []byte) error
type Sink ¶
type Sink interface {
// Apply applies one change. For KindSynced it should flush/commit; KindReset/KindGap are
// advisory (the sink may clear derived state / record the gap).
Apply(c Change) error
// Commit durably records cursor as the resume point (everything up to it is applied).
Commit(cursor []byte) error
// Cursor returns the last committed resume cursor (nil = from the beginning).
Cursor() []byte
}
Sink applies Changes into a target store and tracks a durable resume cursor. It is the shared, transport-neutral core: an HTTP client (classad/changefeed) and a dbrpc/CEDAR replicator both feed Changes to it. Apply must be idempotent enough for at-least-once delivery (a reconnect re-delivers the tail): the mutable sink is last-write-wins by key; the archive sink is append-only (see NewArchiveSink).
func NewArchiveSink ¶
func NewArchiveSink(a *db.ArchiveTable, src string, store CursorStore) (Sink, error)
NewArchiveSink builds a Sink that imports a source's changes into archive a, stamping SrcAttr=src.
func NewTableSink ¶
NewTableSink builds a Sink that imports a source's changes into mutable table d, stamping SrcAttr.
Source Files