Documentation
¶
Overview ¶
Package position defines the replication position contract. This file implements the MySQL GTID-set position: a set of uuid:interval pairs with containment as the natural order. It is deliberately free of the go-mysql dependency — the mysql source converts to and from go-mysql's type at its own boundary.
Package position defines the replication position contract. This file implements the PostgreSQL log sequence number (LSN) position.
Package position defines the replication position contract (GTID | LSN), including the containment that upholds the DBLog caught-up proof.
Variable-keyspace positions — DESIGN, NOT IMPLEMENTED.
This file records the design for the Kinesis position type before the contract changes. It is deliberately not wired anywhere yet; it exists so the decision is conscious and the shapes are agreed before a source that needs them lands.
The problem. Offsets{Parts map[int32]int64} assumes a stable key set: Kafka partitions appear once and only grow. Kinesis shards SPLIT and MERGE: shards close and children appear with parent lineage, so the key space changes over time and reading order is mandatory (drain a parent fully before its child, or event order — and therefore upsert — breaks).
Two consequences the current contract does not express:
- A position recorded yesterday references shards that no longer exist. Contains must treat a closed, fully-drained shard as satisfied, not pending.
- Sequence numbers are decimal strings up to 56 digits — they do not fit int64, so the coordinate is a string.
The shape:
type ShardPos struct {
Sequence string // string: a Kinesis sequence number does not fit int64
Closed bool // shard closed and fully drained
Parent string // lineage, for ordering the read (parent before child)
}
type StreamPosition struct {
Stream string
Shards map[string]ShardPos
}
// Contains: every shard of other is covered; a Closed, drained shard
// counts as covered even if it no longer exists in the current space.
func (p *StreamPosition) Contains(other Position) bool
Caps would declare the capabilities a monotonic-offset source (Kafka) lacks and a reshardable one (Kinesis) has:
type Caps struct {
ReshardableKeyspace bool // shards may appear/disappear (Kinesis)
MonotonicSequence bool // the coordinate never reappears
}
StreamPosition must implement position.Position (String/Compare/Contains) so the runner and coordinator consume it through the existing contract, with no type assertion.
Index ¶
- Constants
- Variables
- func Ahead(global Position, ps map[string]Position) []string
- func StringOrNone(p Position) string
- type GTID
- type LSN
- type Meeter
- type Offsets
- func (o *Offsets) Clone() *Offsets
- func (o *Offsets) Compare(other Position) int
- func (o *Offsets) Contains(other Position) bool
- func (o *Offsets) MarshalJSON() ([]byte, error)
- func (o *Offsets) Meet(other Position) (Position, bool)
- func (o *Offsets) Set(topic string, partition int32, offset int64)
- func (o *Offsets) String() string
- func (o *Offsets) UnmarshalJSON(data []byte) error
- type Position
Constants ¶
const Incomparable = math.MaxInt
Incomparable is the Compare result for two positions with no defined order. Some positions are opaque cookies (plugin offsets, contract §8.2) — their only defined relation is identity. Callers MUST treat this as "cannot decide": never skip data, never advance the confirmed point past it.
Variables ¶
var ErrNoPosition = errors.New("position: no known position")
Functions ¶
func Ahead ¶ added in v0.3.0
Ahead returns the keys of ps whose position is strictly after global, sorted for stable output. On resume from global, a stream that is ahead has already-committed data past the resume point, so it replays from global (idempotent under upsert) — the crash-recovery set. A nil global yields no keys.
func StringOrNone ¶ added in v0.4.0
StringOrNone renders p for a log line, "none" when there is no position (a fresh start).
Types ¶
type GTID ¶
type GTID struct {
// contains filtered or unexported fields
}
GTID is a MySQL GTID set (a set of uuid:interval pairs) as a Position.
func ParseGTID ¶
ParseGTID parses a GTID set string such as "uuid1:1-5,uuid2:7". Empty input yields an empty set.
func (*GTID) Compare ¶
Compare orders sets. Containment is the natural order: a set containing another is greater. Incomparable sets (disjoint uuid universes) fall back to their maximum transaction number — a heuristic that may return 0 for sets that are structurally different (e.g. uuid1:1 vs uuid2:1). Callers that need strict equality should also check Contains in both directions.
type LSN ¶
type LSN uint64
LSN is a PostgreSQL log sequence number: a byte offset into the write-ahead log, canonically rendered as high/low hex ("0/1A2B3C"). As a resume position it means "everything committed at or before this offset".
type Meeter ¶ added in v0.1.1
Meeter is implemented by positions with a PARTIAL order, where two values can diverge without either containing the other. Meet returns their greatest lower bound — the position every input contains — or false when there is none (different topics, different sources).
MinSafe needs this because a partial order has no minimum among the inputs to select: for Kafka offsets {p0:100,p1:5} and {p0:5,p1:100}, neither is a safe resume point, and the only safe one ({p0:5,p1:5}) is not in the list. A totally ordered position (GTID, LSN) does not implement it — its minimum is always one of the inputs.
type Offsets ¶
type Offsets struct {
// Topics maps a topic name to its per-partition offsets.
Topics map[string]map[int32]int64
}
Offsets is a Kafka consumer position: per-partition offsets for one or more topics. It implements Position with a PARTIAL order — two positions are ordered only when one contains the other; positions whose partitions diverge are Incomparable, and Meet gives their greatest lower bound.
The topic is part of the key, not a single field: one reader subscribes to every topic its spec tables name, and partition numbers repeat across topics. Keying by partition alone collapsed orders/p0 and payments/p0 onto one entry, so whichever topic wrote last destroyed the other's offset and the resume either replayed or skipped.
The wire form is JSON; the canonical text form is "topic:p0=10,p1=20", with multiple topics joined by ";" in topic order.
func NewOffsets ¶ added in v0.3.0
NewOffsets builds a position for a single topic.
func ParseOffsets ¶
ParseOffsets parses the canonical text form "topic:p0=10,p1=20", with multiple topics separated by ";".
func (*Offsets) Clone ¶ added in v0.3.0
Clone returns a deep copy. Callers hand positions to other goroutines (the reader keeps mutating its own while a batch carries a snapshot), so sharing the inner maps would race.
func (*Offsets) Compare ¶
Compare implements the PARTIAL order of Offsets: containment. A position is greater when it contains the other (every partition of every topic at or beyond the other's offset), and the two are equal when each contains the other. Anything else has no defined order and returns Incomparable — notably positions that diverge ({p0:100,p1:5} vs {p0:5,p1:100}), whether the divergence is within one topic or across topics.
It does NOT fall back to the maximum partition offset. That heuristic reported 0 for divergent positions, which made covered() (coordinator/ flow.go) treat an uncovered batch as covered and skip data that was never replayed. Incomparable is the value every caller already handles as "cannot decide — never skip".
Because the order is partial, Min/MinSafe cannot pick a safe minimum by selection alone; Meet computes the greatest lower bound instead.
func (*Offsets) Contains ¶
Contains returns true when o's offsets are all at least as large as other's, for every partition of every topic present in other. Topics and partitions present in o but absent in other are ignored (future work).
func (*Offsets) MarshalJSON ¶
MarshalJSON serializes the offsets as a JSON object.
func (*Offsets) Meet ¶ added in v0.1.1
Meet returns the greatest lower bound of o and other: the per-partition minimum offset, for every topic both sides carry. It is the only safe resume point when two positions diverge — selecting either one would resume PAST the partitions where the other is behind, and that data would never be read.
A partition (or topic) missing from one side is absent from the meet: an unknown offset cannot be bounded, so the resume must not assume progress on it.
func (*Offsets) Set ¶ added in v0.3.0
Set records the offset of one partition, creating the topic entry when it is the first partition seen for that topic.
func (*Offsets) UnmarshalJSON ¶
UnmarshalJSON deserializes the offsets from a JSON object. It detects the legacy {"topic":"...","parts":[...]} format and migrates it into the canonical {"topics":[{topic,parts}]} shape.
type Position ¶
type Position interface {
String() string
Compare(other Position) int
Contains(other Position) bool
}
func Min ¶
Min returns the smallest position of a homogeneous list: the one every other position extends. Under a single source every committed position extends the previous one, so the order is well defined — containment for GTID sets, linearity for LSNs. Compare carries that order for both.
An incomparable pair is folded with Meet when the type provides one, so a partially ordered position (Kafka offsets) yields the greatest lower bound rather than an arbitrary side. Without a meet the current best is kept: conservative when the order is undefined.
P3: for a RESUME or RETENTION decision, prefer MinSafe — where Min keeps going, MinSafe reports the pair it cannot bound instead of returning a minimum that may be ahead of an uncommitted position.
func MinSafe ¶
MinSafe is Min for a RESUME decision: it returns a position that every input contains, so resuming from it can never skip uncommitted data.
For a totally ordered position (GTID, LSN) that is the smallest input. For a PARTIAL order (Kafka offsets, folded one-per-partition-owner by WK-001 C7) no input need be safe: {p0:100,p1:5} and {p0:5,p1:100} each sit ahead of the other on one partition. Selecting either resumes past the other's uncommitted range, so an incomparable pair is folded with Meet into their greatest lower bound ({p0:5,p1:5}) instead.
An incomparable pair with no meet is an error, not a silent skip: there is no safe choice, so fail fast rather than resume past data (P1).
func Parse ¶
Parse decodes a committed position string using the parser for a source kind ("mysql", "postgres", "kafka"); an unknown or empty kind parses as a MySQL GTID set, the default. The position format is a property of the source, so a caller that stores opaque position strings but must compare them (a sink keeping one position per partition, WK-001 §2.6) carries the kind as a hint instead of reimplementing the format.
Source Files