package coordination

import "github.com/krabka-io/krabka-streams-go/coordination"

Package coordination elects one leader per role, and it proves that leadership on every guarded write.

A control plane that dispatches commands to external devices needs two properties from its log. A command is durable in a quorum before the dispatch, and a deposed leader cannot dispatch at all. The second property is the hard one. A leader that loses its lease does not know that it lost the lease, so it cannot fence itself. Something outside the leader must refuse its writes.

The epoch is the safety mechanism and the lease is not

The leadership epoch is the producer epoch that Kafka's transaction coordinator mints for the transactional id of a role. That one decision supplies every property this package promises:

The lease adds no safety of its own. It is a liveness and anti-flap device. It tells a standby when to stop waiting for a quiet holder. A wrong lease makes a failover early or late. A wrong lease never makes two writers authoritative. Read that inversion against a hand-rolled design, where the lease deadline decides who may write. This package never asks the lease that question.

The fencing token is a pair

The producer epoch is an int16 and it wraps. Kafka handles the exhaustion with a fresh producer id and an epoch of zero. An implementation that compares the epoch alone accepts a stale writer after about 32000 leadership changes. FencingToken holds the pair, and FencingToken.Compare reads the producer id first. See NoEpoch for the token of a role that no member has taken.

Rank comes from the log

A candidate appends a registration record to StateTopic. The offset of that record in the partition is the join sequence of the candidate. Log compaction keeps the offset of every record it retains, so a reader that walks the partition in offset order sees the registrations in registration order. BuildRoleState folds that walk into a RoleState, and the roster it builds is in offset order.

A recovered node registers again. The new record takes a higher offset, so the node lands at the tail of the roster and takes the last rank. That is the no-failback rule, and it needs no counter and no coordinator. A rank from a configuration file would put the recovered node at the front, and the node would then preempt the member that replaced it. The cluster would pay for a second failover that it does not need.

The stagger is an optimisation

A challenger of rank n waits n challenge staggers past the deadline of the lease. This saves epoch churn, and it does nothing else. InitProducerId is atomic at the coordinator, so a simultaneous challenge by every standby still leaves exactly one member with the newest epoch. The losers keep an older epoch, and each one learns that it lost on its first guarded write. Set the stagger to make that outcome rare. Do not set it to make the outcome safe, because the outcome is already safe.

This package pins no Kafka client

The library pins no Kafka client, so this package reaches a broker through the small interfaces in Transport. An adapter over franz-go or confluent-kafka-go is a few dozen lines of glue, and it is the same boundary that the columnar group runner draws. Everything else is here: the record codec, the roster and the rank logic, the challenge timing, the lease clock, the renewal loop, AcquireLeadership, and DescribeLeadership.

Clock assumptions

The lease deadline is a wall-clock instant. The holder writes it, and a challenger compares it against the clock of the challenger. Skew between the two clocks moves the failover time by the size of the skew. Skew does not affect safety, for the reason the first section gives. An operator that needs a bound on the failover time needs a bound on the skew.

Index

Constants

const (
    // DefaultLeaseDuration is the default extent of a lease.
    DefaultLeaseDuration = 30 * time.Second

    // DefaultRenewInterval is the default gap between two renewals by the
    // holder.
    DefaultRenewInterval = 10 * time.Second

    // DefaultChallengeStagger is the default extra delay that one rank of
    // succession adds.
    DefaultChallengeStagger = 5 * time.Second
)

The default timings of a lease. Change one with WithLeaseDuration, WithRenewInterval, or WithChallengeStagger. 

const (
    // MaxRoleLen is the maximum length of a [Role], in bytes.
    MaxRoleLen = 249

    // MaxMemberIDLen is the maximum length of a [MemberID], in bytes.
    MaxMemberIDLen = 249
)

The maximum length of the two names of the topic, in bytes. The bound is the bound Kafka puts on a topic name, so a role stays short enough to log and to compare. 

const DefaultPartitions = 16

DefaultPartitions is the partition count that this package assumes for StateTopic. Pass the real count of the cluster with WithPartitions when an operator created the topic with another count. 

const DefaultPollInterval = time.Second

DefaultPollInterval is the default gap between two reads of StateTopic while a member waits for a role. It also bounds the retry of a renewal that failed for a reason other than a fence. 

const StateTopic = "__coordination_state"

StateTopic is the compacted internal topic that holds the coordination state of a cluster. Every record of one role goes to one partition, so the records of a role carry a total order. See RolePartition.

The topic needs cleanup.policy=compact. A client writes it with acks=all, and an operator should set min.insync.replicas to at least 2. The claim "durable in a quorum before the dispatch" rests on that topic configuration, and not on this package.

Variables

var ErrFenced = errors.New("fenced: another member holds the role")

ErrFenced reports that the broker refused a write, because another member holds the role now.

This is the expected end of a leadership. A LeaseWriter wraps this error for broker error code 47 INVALID_PRODUCER_EPOCH and for broker error code 90 PRODUCER_FENCED. AcquireLeadership and Leadership.Renew find it with errors.Is.

A deposed leader learns that it lost the role from this error, and from nothing else. No clock takes part in the check. The epoch it held is superseded, so the cluster already rejects every write it makes. The caller stops the work of the role at once. 

var ErrNotHeld = errors.New("the member no longer holds the role")

ErrNotHeld reports that a caller asked a *Leadership to act after the leadership ended. 

var NoEpoch = FencingToken{/* contains filtered or unexported fields */}

NoEpoch is the token of a role that no member has ever taken. Kafka writes -1 for "no producer", and that pair proves no leadership. NoEpoch ranks below every minted token, because NewFencingToken rejects a negative value. Treat NoEpoch as a constant.

Functions

func EncodeKey source

func EncodeKey(key Key) []byte

EncodeKey encodes a key of StateTopic.

The function writes the empty member string for a lease key, whatever key.Member holds, because the frozen layout gives a lease no member.

func EncodeLease source

func EncodeLease(lease Lease) []byte

EncodeLease encodes the value of a lease record.

func EncodeRegistration source

func EncodeRegistration(registration Registration) []byte

EncodeRegistration encodes the value of a registration record.

func IsTombstone source

func IsTombstone(value []byte) bool

IsTombstone reports whether a record value is a tombstone. A tombstone of a registration key deregisters one member, and a tombstone of a lease key clears the lease of one role. No value of this topic is legally empty, so an empty value and a nil value are both tombstones.

func RolePartition source

func RolePartition(role Role, partitions int) (int, error)

RolePartition returns the partition of StateTopic that holds every record of role. It rejects a partition count below one.

The rule is Kafka's own key partitioning: murmur2 of the role name in UTF-8, masked with Utils.toPositive, and then the remainder of partitions. The mask clears the sign bit. It is not an absolute value, and the two differ for every negative hash. krabka-client-rs and krabka-streams-java write the same topic, so they compute the same rule. A change here needs the same change in the two ports.

Both writers of a role pin the partition with this rule. The pin is a correctness requirement and not a preference. Kafka's default partitioner hashes the record key. The registration key of a role and the lease key of the same role differ, because the registration key names a member and the lease key does not. A partitioner that reads the key puts the two kinds in two partitions, and the total order that the succession rules rank on is gone.

Example

Both writers of a role pin the partition from the role name, so the records of one role keep a total order. The rule is Kafka's own key partitioning. 

package main

import (
    "fmt"
    "log"

    "github.com/krabka-io/krabka-streams-go/coordination"
)

func main() {
    role, err := coordination.NewRole("controller")
    if err != nil {
        log.Fatal(err)
    }

    partition, err := coordination.RolePartition(role, 16)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Printf("role %s writes to %s-%d\n", role, coordination.StateTopic, partition)
}

Output:

role controller writes to __coordination_state-12

Types

type Action source

type Action int

Action is the step that one member takes about a role now. 

const (
    // ActionNotRegistered says that the member is not in the roster of the
    // role, so it has no rank. It registers first, reads the partition again,
    // and evaluates again.
    ActionNotRegistered Action = iota

    // ActionHold says that this member holds the role and that its lease is
    // live. It renews the lease, and it calls InitProducerId for nothing.
    ActionHold

    // ActionChallenge says that this member calls InitProducerId now.
    ActionChallenge

    // ActionWait says that this member waits and then evaluates the role
    // state again. [Decision.WaitUntil] carries the instant.
    ActionWait
)

func (Action) String source

func (a Action) String() string

String names the action.

type Clock source

type Clock interface {
    // NowMillis returns the current time, in milliseconds since the Unix
    // epoch.
    NowMillis() int64

    // After returns a channel that receives one value after the extent
    // passes. An extent at or below zero receives at once.
    After(extent time.Duration) <-chan time.Time
}

Clock is the source of the current time, and the source of every wait.

The succession rules and the lease clock take an instant as a parameter, so they need no clock at all. This interface is the seam for the code around them that reads a real clock and that waits. A test supplies *ManualClock, drives time by hand, and sleeps for nothing.

type Coordinator source

type Coordinator interface {
    // AcquireEpoch mints a new epoch for the role and fences the previous
    // holder. The implementation calls InitProducerId with
    // transactional.id = <role>, and it keeps the producer that the epoch
    // binds, because [LeaseWriter.WriteLease] writes through that producer.
    //
    // The transaction coordinator picks the epoch, so the quorum mints the
    // value and the value only grows. The implementation should pass the
    // lease duration as transaction_timeout_ms, so the coordinator aborts the
    // open transaction of a dead holder without a client.
    AcquireEpoch(ctx context.Context, role Role) (FencingToken, error)

    // DescribeEpoch reports the token that the transaction coordinator holds
    // for the role. It reports false for a role that no member has ever
    // taken. The implementation calls DescribeTransactions, which joins no
    // group and takes no epoch.
    DescribeEpoch(ctx context.Context, role Role) (FencingToken, bool, error)
}

Coordinator mints and reports the leadership epoch of a role. An adapter implements it over the transaction coordinator of a cluster.

The two calls carry different weight. AcquireEpoch takes the role, and DescribeEpoch only looks.

type Decision source

type Decision struct {
    // Action is the step that the member takes now.
    Action Action

    // WaitUntil is the instant of the next evaluation, in milliseconds since
    // the Unix epoch. It carries a value for [ActionWait] only.
    WaitUntil int64
}

Decision is what one member does about a role right now. Evaluate returns one.

func Evaluate source

func Evaluate(state RoleState, member MemberID, nowMillis int64, config LeaseConfig) Decision

Evaluate decides what one member does about a role at this instant.

The rules are:

  1. A member that is not in the roster gets ActionNotRegistered. It has no rank, because rank comes from the registration record.
  2. A member that the lease names, while that lease is live, gets ActionHold.
  3. A challenger of rank n gets ActionChallenge from the deadline plus n challenge staggers on. With no lease, the anchor is the registration instant of the member instead of a deadline, so rank 0 challenges at once and rank n challenges n staggers later.
  4. Every other member gets ActionWait. The instant it carries is the earliest instant at which this answer changes for an unchanged role state.

The anchor of rule 3 always comes from the role state, and never from the current instant. A member that has no lease to wait for anchors on its own registration record. An anchor of "now plus n staggers" would move forward on every evaluation, and a standby of rank 1 or more would then wait for ever while rank 0 is dead.

The member of an expired lease keeps rank 0, so it reclaims its own role at its own deadline. See RoleState.RankOf.

A caller evaluates again when it reads a new record, and at the latest at the instant that Decision.WaitUntil names.

Example

The succession rules decide when a member calls InitProducerId. They never decide who is authoritative, because the transaction coordinator decides that. 

package main

import (
    "fmt"
    "log"
    "time"

    "github.com/krabka-io/krabka-streams-go/coordination"
)

func main() {
    role, err := coordination.NewRole("controller")
    if err != nil {
        log.Fatal(err)
    }
    config, err := coordination.NewLeaseConfig(30*time.Second, 10*time.Second, 5*time.Second)
    if err != nil {
        log.Fatal(err)
    }

    records := []coordination.StateRecord{
        registration(role, "node-1", 0, 0),
        registration(role, "node-2", 1, 0),
        registration(role, "node-3", 2, 0),
        lease(role, "node-1", 3, 0, 30000),
    }
    state, err := coordination.BuildRoleState(role, records)
    if err != nil {
        log.Fatal(err)
    }

    for _, name := range []string{"node-1", "node-2", "node-3"} {
        member, err := coordination.NewMemberID(name)
        if err != nil {
            log.Fatal(err)
        }
        rank, _ := state.RankOf(member)
        decision := coordination.Evaluate(state, member, 30000, config)
        fmt.Printf("%s rank %d: %s %d\n", name, rank, decision.Action, decision.WaitUntil)
    }
}

// registration builds the log record of one registration.
func registration(role coordination.Role, name string, offset, registeredAt int64) coordination.StateRecord {
    member, err := coordination.NewMemberID(name)
    if err != nil {
        log.Fatal(err)
    }
    return coordination.StateRecord{
        Offset: offset,
        Key:    coordination.EncodeKey(coordination.RegistrationKey(role, member)),
        Value: coordination.EncodeRegistration(coordination.Registration{
            Member:       member,
            RegisteredAt: registeredAt,
        }),
    }
}

// lease builds the log record of one lease.
func lease(role coordination.Role, name string, offset, grantedAt, deadline int64) coordination.StateRecord {
    member, err := coordination.NewMemberID(name)
    if err != nil {
        log.Fatal(err)
    }
    token, err := coordination.NewFencingToken(11, 1)
    if err != nil {
        log.Fatal(err)
    }
    return coordination.StateRecord{
        Offset: offset,
        Key:    coordination.EncodeKey(coordination.LeaseKey(role)),
        Value: coordination.EncodeLease(coordination.Lease{
            Member:    member,
            Token:     token,
            GrantedAt: grantedAt,
            Deadline:  deadline,
        }),
    }
}

Output:

node-1 rank 0: challenge 0
node-2 rank 0: challenge 0
node-3 rank 1: wait 35000

type FencingToken source

type FencingToken struct {
    // contains filtered or unexported fields
}

FencingToken is the quorum-minted proof that one member holds a role.

The transaction coordinator mints the pair when a member calls InitProducerId for the transactional id of the role. The quorum picks the values, the broker enforces them, and a write from an older pair fails with a fenced error. A holder passes the token to every guarded write.

The pair is load-bearing. FencingToken.Compare reads the producer id first and the producer epoch second, so the order is lexicographic. A producer epoch is an int16 and it wraps. Kafka then allocates a new producer id and resets the epoch to zero. A comparison on the epoch alone would rank that fresh token below the stale one, and the old leader would keep the role. Never compare the epoch on its own.

func NewFencingToken source

func NewFencingToken(producerID int64, producerEpoch int16) (FencingToken, error)

NewFencingToken builds a token from a minted producer id and producer epoch. It rejects a negative value in either position.

func ParseFencingToken source

func ParseFencingToken(text string) (FencingToken, error)

ParseFencingToken parses the producer_id:producer_epoch form that FencingToken.String writes.

func (FencingToken) Compare source

func (t FencingToken) Compare(other FencingToken) int

Compare orders two tokens lexicographically. It reads the producer id first, and it reads the producer epoch only for two tokens of one producer id. It returns a negative number when t ranks below other, zero for two equal tokens, and a positive number when t supersedes other.

The producer id leads, because Kafka allocates producer ids from a monotonic block allocator and resets the epoch to zero at every new id. The pair stays monotonic where the epoch alone does not.

Example

The producer epoch is an int16 and it wraps. Kafka answers the exhaustion with a fresh producer id and an epoch of zero, so a comparison of the epoch alone would keep a deposed leader in place. 

package main

import (
    "fmt"
    "log"
    "math"

    "github.com/krabka-io/krabka-streams-go/coordination"
)

func main() {
    exhausted, err := coordination.NewFencingToken(4, math.MaxInt16)
    if err != nil {
        log.Fatal(err)
    }
    fresh, err := coordination.NewFencingToken(5, 0)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Printf("%s supersedes %s: %t\n", fresh, exhausted, fresh.Supersedes(exhausted))
    fmt.Printf("the epoch alone would say: %t\n", fresh.ProducerEpoch() > exhausted.ProducerEpoch())
}

Output:

5:0 supersedes 4:32767: true
the epoch alone would say: false

func (FencingToken) ProducerEpoch source

func (t FencingToken) ProducerEpoch() int16

ProducerEpoch returns the producer epoch that the transaction coordinator minted.

func (FencingToken) ProducerID source

func (t FencingToken) ProducerID() int64

ProducerID returns the producer id that the transaction coordinator minted.

func (FencingToken) String source

func (t FencingToken) String() string

String writes the producer_id:producer_epoch form.

func (FencingToken) Supersedes source

func (t FencingToken) Supersedes(other FencingToken) bool

Supersedes reports whether t ranks above other under FencingToken.Compare.

type FormatError source

type FormatError struct {
    // Part names the record part that failed to decode. It is "key",
    // "registration value", or "lease value".
    Part string
    // contains filtered or unexported fields
}

FormatError reports a malformed record of StateTopic.

func (*FormatError) Error source

func (e *FormatError) Error() string

Error describes the fault and names the record part that carries it.

type KafkaTransport source

type KafkaTransport struct {
    // contains filtered or unexported fields
}

KafkaTransport implements Transport with franz-go.

func NewKafkaTransport source

func NewKafkaTransport(lease LeaseConfig, brokers ...string) (*KafkaTransport, error)

NewKafkaTransport connects a coordination transport to brokers.

func (*KafkaTransport) AcquireEpoch source

func (t *KafkaTransport) AcquireEpoch(ctx context.Context, role Role) (FencingToken, error)

AcquireEpoch initializes a transactional producer whose id is the role.

func (*KafkaTransport) Append source

func (t *KafkaTransport) Append(ctx context.Context, partition TopicPartition, key, value []byte) error

Append writes a registration outside a transaction.

func (*KafkaTransport) Close source

func (t *KafkaTransport) Close()

Close releases every producer owned by the transport.

func (*KafkaTransport) DescribeEpoch source

func (t *KafkaTransport) DescribeEpoch(ctx context.Context, role Role) (FencingToken, bool, error)

DescribeEpoch reads the producer id and epoch held by role.

func (*KafkaTransport) ReadPartition source

func (t *KafkaTransport) ReadPartition(ctx context.Context, partition TopicPartition) ([]StateRecord, error)

ReadPartition returns committed records from the beginning through the current idle edge.

func (*KafkaTransport) WriteLease source

func (t *KafkaTransport) WriteLease(ctx context.Context, partition TopicPartition, token FencingToken, key, value []byte) error

WriteLease writes and commits one lease record under token.

type Key source

type Key struct {
    // Kind names the record that the key holds.
    Kind RecordKind

    // Role is the role that the record belongs to.
    Role Role

    // Member is the member of a registration key. A lease key carries the
    // zero [MemberID], because a lease belongs to the role and not to one
    // member.
    Member MemberID
}

Key is the decoded key of one record of StateTopic. The key is the compaction key, so the topic keeps the last record of every role and every member.

func DecodeKey source

func DecodeKey(data []byte) (Key, error)

DecodeKey decodes a key of StateTopic. It returns a *FormatError for a truncated buffer, for trailing bytes, for an unknown version, for an unknown kind, for a negative string length, for a role or a member id that its constructor rejects, and for a lease key that names a member.

func LeaseKey source

func LeaseKey(role Role) Key

LeaseKey builds the key of the lease of a role. It carries no member.

func RegistrationKey source

func RegistrationKey(role Role, member MemberID) Key

RegistrationKey builds the key of the registration of one member of a role.

type Leadership source

type Leadership struct {
    // contains filtered or unexported fields
}

Leadership is the live leadership of one member over one role.

The handle carries the epoch that the transaction coordinator minted, and the caller passes that epoch to every guarded write. The broker fences a write of an older epoch, so the caller proves its authority with the write itself and asks no question first.

The handle renews the lease in the background. The renewal keeps the standbys from a challenge, and it adds no safety. Leadership.Done closes when the leadership ends, and Leadership.Err then says why. A caller watches Done and stops the work of the role at once.

A Leadership is safe for concurrent use.

func AcquireLeadership source

func AcquireLeadership(ctx context.Context, transport Transport, role Role, member MemberID, options ...Option) (*Leadership, error)

AcquireLeadership competes for a role, and it returns when this member holds the role.

The call registers the member, waits for its turn under the succession rules, mints the epoch of the role, and writes the first lease record. It returns a *Leadership that renews the lease in the background. Every write that the caller guards with Leadership.Token is fenced by the broker.

The call reads the coordination topic in a loop, so the caller bounds the wait with the context. It returns the context error when the deadline passes before this member wins the role.

A role state that names this member takes the same path as a challenge. The record came from an earlier incarnation of the member, because this call holds no epoch yet. The new epoch fences that earlier incarnation, and it costs one epoch.

func (*Leadership) Done source

func (l *Leadership) Done() <-chan struct{}

Done returns a channel that closes when the leadership ends. A fence closes it, and Leadership.Resign closes it. Leadership.Err says which.

func (*Leadership) Err source

func (l *Leadership) Err() error

Err returns the reason the leadership ended, and nil while it holds. The error of a fence wraps ErrFenced, and the error of a resignation wraps ErrNotHeld.

func (*Leadership) Lease source

func (l *Leadership) Lease() Lease

Lease returns the lease record that this member wrote last.

func (*Leadership) Member source

func (l *Leadership) Member() MemberID

Member returns the member that holds the role.

func (*Leadership) Partition source

func (l *Leadership) Partition() TopicPartition

Partition returns the partition of the coordination topic that holds every record of the role.

func (*Leadership) Renew source

func (l *Leadership) Renew(ctx context.Context) error

Renew writes a fresh lease record under the same epoch. The background loop calls it at every renew interval, and a caller that drives its own schedule calls it too.

A renewal that the broker fences ends the leadership. Renew then returns an error that wraps ErrFenced, and Leadership.Done closes.

func (*Leadership) Resign source

func (l *Leadership) Resign(ctx context.Context) error

Resign gives the role up. It stops the renewal loop and writes a tombstone on the lease key, so the first standby challenges at once instead of at the deadline. It ends the leadership, whatever the write does.

Resign takes no epoch back. The member keeps the epoch it minted until another member mints a newer one, and the broker still accepts a write of the resigned member until then. A caller that resigns stops its own guarded writes.

func (*Leadership) Role source

func (l *Leadership) Role() Role

Role returns the role that this member holds.

func (*Leadership) Token source

func (l *Leadership) Token() FencingToken

Token returns the quorum-minted proof of this leadership. The caller passes it to every guarded write.

type LeadershipStatus source

type LeadershipStatus struct {
    // Role is the role that the status describes.
    Role Role

    // Token is the token that the transaction coordinator holds now. It is
    // [NoEpoch] when no member has ever taken the role.
    Token FencingToken

    // Held reports whether a member has taken the role.
    Held bool
}

LeadershipStatus reports the token that the transaction coordinator holds for one role. DescribeLeadership builds it.

func DescribeLeadership source

func DescribeLeadership(ctx context.Context, coordinator Coordinator, role Role) (LeadershipStatus, error)

DescribeLeadership reports which token holds a role now.

The call reaches the transaction coordinator only. It reads no topic, it joins no group, and it takes no epoch, so any process makes it. This is the third-party check of the authority of a writer.

Example

A third party checks the authority of a writer with one request. It reads no topic, it joins no group, and it takes no epoch. 

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/krabka-io/krabka-streams-go/coordination"
)

func main() {
    role, err := coordination.NewRole("controller")
    if err != nil {
        log.Fatal(err)
    }
    token, err := coordination.NewFencingToken(4242, 7)
    if err != nil {
        log.Fatal(err)
    }

    status, err := coordination.DescribeLeadership(context.Background(),
        stubCoordinator{token: token}, role)
    if err != nil {
        log.Fatal(err)
    }

    stale, err := coordination.NewFencingToken(4242, 6)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(status)
    fmt.Printf("the holder writes: %t\n", status.Authoritative(token))
    fmt.Printf("a deposed member writes: %t\n", status.Authoritative(stale))
}

// stubCoordinator answers with one token. A real adapter calls
// DescribeTransactions.
type stubCoordinator struct {
    token coordination.FencingToken
}

func (s stubCoordinator) AcquireEpoch(context.Context, coordination.Role) (coordination.FencingToken, error) {
    return s.token, nil
}

func (s stubCoordinator) DescribeEpoch(context.Context, coordination.Role) (coordination.FencingToken, bool, error) {
    return s.token, true, nil
}

Output:

role controller: epoch 4242:7
the holder writes: true
a deposed member writes: false

func (LeadershipStatus) Authoritative source

func (s LeadershipStatus) Authoritative(token FencingToken) bool

Authoritative reports whether the token is the token that the coordinator holds now. A third party checks the authority of a writer with this one question, and it joins no group.

func (LeadershipStatus) String source

func (s LeadershipStatus) String() string

String describes the status.

type Lease source

type Lease struct {
    // Member is the member that holds the role.
    Member MemberID

    // Token is the quorum-minted proof of the claim of the holder.
    Token FencingToken

    // GrantedAt is the time the holder took the lease, in milliseconds since
    // the Unix epoch. A renewal moves it forward.
    GrantedAt int64

    // Deadline is the time the lease expires, in milliseconds since the Unix
    // epoch.
    Deadline int64
}

Lease is the value of a record that names the member that holds a role now, and the time the claim of that member ends.

Token is the authority. The broker fences a write that carries an older token, and no clock takes part in that check. Deadline is an anti-flap device. A standby waits for the deadline to pass before it challenges the holder, so a short pause does not move the role.

func DecodeLease source

func DecodeLease(data []byte) (Lease, error)

DecodeLease decodes the value of a lease record. It returns a *FormatError for a truncated buffer, for trailing bytes, for an unknown version, for a negative string length, for a member id that its constructor rejects, and for a negative producer id or producer epoch.

Example

A member writes a lease record under the epoch of its role. A reader of the coordination topic decodes the record with the codec of this package. 

package main

import (
    "fmt"
    "log"

    "github.com/krabka-io/krabka-streams-go/coordination"
)

func main() {
    member, err := coordination.NewMemberID("node-1")
    if err != nil {
        log.Fatal(err)
    }
    token, err := coordination.NewFencingToken(4242, 7)
    if err != nil {
        log.Fatal(err)
    }
    written := coordination.Lease{
        Member:    member,
        Token:     token,
        GrantedAt: 1700000000000,
        Deadline:  1700000030000,
    }

    value := coordination.EncodeLease(written)
    read, err := coordination.DecodeLease(value)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Printf("%d bytes\n", len(value))
    fmt.Printf("%s holds the role with token %s until %d\n", read.Member, read.Token, read.Deadline)
}

Output:

36 bytes
node-1 holds the role with token 4242:7 until 1700000030000

type LeaseConfig source

type LeaseConfig struct {
    // contains filtered or unexported fields
}

LeaseConfig holds the three timings of one role. The timings are independent of the record layout, so one process holds several roles with different timings.

The lease adds no safety. It decides when a standby challenges a quiet holder, and nothing else. A wrong timing makes a failover early or late. A wrong timing never makes two members authoritative.

Build a value with NewLeaseConfig or take DefaultLeaseConfig.

func DefaultLeaseConfig source

func DefaultLeaseConfig() LeaseConfig

DefaultLeaseConfig returns the config of DefaultLeaseDuration, DefaultRenewInterval, and DefaultChallengeStagger.

func NewLeaseConfig source

func NewLeaseConfig(duration, renewInterval, challengeStagger time.Duration) (LeaseConfig, error)

NewLeaseConfig builds a lease policy from three extents. It rejects an extent at or below zero, and it rejects a renew interval at or above the lease duration, because the holder then never renews in time.

A caller should set renewInterval to at most a third of duration. The holder then keeps two more attempts before the deadline. This constructor accepts a larger value. Ask LeaseConfig.RenewsWithMargin which side of that bound a value is on.

func (LeaseConfig) ChallengeDelay source

func (c LeaseConfig) ChallengeDelay(rank int) time.Duration

ChallengeDelay returns the extra delay that a challenger of this rank takes. Rank 0 takes none. The value saturates, so a very large rank does not wrap the instant that a caller adds it to.

func (LeaseConfig) ChallengeStagger source

func (c LeaseConfig) ChallengeStagger() time.Duration

ChallengeStagger returns the extra delay that one rank of succession adds.

func (LeaseConfig) Duration source

func (c LeaseConfig) Duration() time.Duration

Duration returns the extent of a lease that a member takes now.

func (LeaseConfig) Grant source

func (c LeaseConfig) Grant(member MemberID, token FencingToken, nowMillis int64) Lease

Grant builds the lease record that a member writes after it mints token. A renewal writes the same record with the same token and a later nowMillis, so this one method covers a first grant and a renewal.

func (LeaseConfig) RenewInterval source

func (c LeaseConfig) RenewInterval() time.Duration

RenewInterval returns the gap between two renewals by the holder.

func (LeaseConfig) RenewsWithMargin source

func (c LeaseConfig) RenewsWithMargin() bool

RenewsWithMargin reports whether the renew interval leaves the holder two spare attempts. The interval this package recommends is at most a third of the lease duration. A config outside that bound still works, and the holder then loses the role after fewer missed renewals.

func (LeaseConfig) Timing source

func (c LeaseConfig) Timing(lease Lease) LeaseTiming

Timing binds a lease record to this policy and answers the clock questions about it.

type LeaseTiming source

type LeaseTiming struct {
    // contains filtered or unexported fields
}

LeaseTiming is the lease clock of one lease record under one policy.

Every method takes and returns milliseconds since the Unix epoch, because a deadline is a coordinate and not an extent. LeaseTiming.RemainingAt is the one exception, and it returns the extent that is left.

func NewLeaseTiming source

func NewLeaseTiming(lease Lease, config LeaseConfig) LeaseTiming

NewLeaseTiming binds a lease record to a policy.

func (LeaseTiming) ChallengeAt source

func (t LeaseTiming) ChallengeAt(rank int) int64

ChallengeAt returns the instant at which a challenger of this rank challenges. Rank 0 challenges at the deadline, and each later rank adds one challenge stagger. The stagger saves epoch churn and gives no safety.

func (LeaseTiming) ExpiresAt source

func (t LeaseTiming) ExpiresAt() int64

ExpiresAt returns the instant the lease expires, in milliseconds since the Unix epoch.

func (LeaseTiming) Lease source

func (t LeaseTiming) Lease() Lease

Lease returns the lease record behind this clock.

func (LeaseTiming) LiveAt source

func (t LeaseTiming) LiveAt(nowMillis int64) bool

LiveAt reports whether the lease is live at this instant. The lease is live before its deadline and expired from the deadline on. The rank 0 challenger challenges at exactly the deadline, so the two tests leave no gap and no overlap.

func (LeaseTiming) RemainingAt source

func (t LeaseTiming) RemainingAt(nowMillis int64) time.Duration

RemainingAt returns the extent that is left before the deadline, and zero after the deadline.

func (LeaseTiming) RenewAt source

func (t LeaseTiming) RenewAt() int64

RenewAt returns the instant at which the holder writes its next renewal. The value is the grant instant plus the renew interval, and it never passes the deadline. A holder that follows it always writes before it loses the lease.

func (LeaseTiming) RenewDueAt source

func (t LeaseTiming) RenewDueAt(nowMillis int64) bool

RenewDueAt reports whether the holder writes a renewal at this instant.

type LeaseWriter source

type LeaseWriter interface {
    // WriteLease writes one record under the token, inside a transaction of
    // the producer that minted the token. A nil value clears the lease. The
    // producer pins the partition, and it applies no partitioner of its own.
    //
    // The implementation returns an error that wraps [ErrFenced] when the
    // broker rejects the token. That rejection is how a deposed holder learns
    // that it lost the role.
    WriteLease(ctx context.Context, partition TopicPartition, token FencingToken, key, value []byte) error
}

LeaseWriter writes the lease record of a role in a transaction under the epoch of the role.

A lease record authenticates itself. The broker fences any writer that does not hold the current epoch, so a lease record that reached the log is proof that its author held the epoch.

type ManualClock source

type ManualClock struct {
    // contains filtered or unexported fields
}

ManualClock is a clock that a test moves by hand. A wait that ManualClock.After returns receives when ManualClock.Set or ManualClock.Advance moves the clock to the instant of the wait.

The clock is safe for concurrent use, so a test shares one clock between a holder and its challengers and moves them all together.

func NewManualClock source

func NewManualClock(nowMillis int64) *ManualClock

NewManualClock builds a clock that reads this instant, in milliseconds since the Unix epoch.

func (*ManualClock) Advance source

func (c *ManualClock) Advance(step time.Duration)

Advance moves the clock forward by this extent and releases every wait that the move makes due.

func (*ManualClock) After source

func (c *ManualClock) After(extent time.Duration) <-chan time.Time

After returns a channel that receives when the clock reaches the instant this extent names. A wait that is already due receives at once.

func (*ManualClock) NowMillis source

func (c *ManualClock) NowMillis() int64

NowMillis returns the instant the clock reads now.

func (*ManualClock) Set source

func (c *ManualClock) Set(nowMillis int64)

Set moves the clock to this instant and releases every wait that the move makes due. It moves the clock backwards too, and a backwards move releases nothing.

type MemberID source

type MemberID struct {
    // contains filtered or unexported fields
}

MemberID is the validated identity of one member that competes for a role.

A member picks its own id and keeps that id across a reconnect, so the succession order survives a short network failure.

NewMemberID is the only way to build a member id with a name, so a non-zero MemberID value is proof that the id is well formed. The zero value names no member, and a lease key carries it.

func NewMemberID source

func NewMemberID(name string) (MemberID, error)

NewMemberID parses a member id. It rejects an empty id and an id longer than MaxMemberIDLen bytes.

func (MemberID) String source

func (m MemberID) String() string

String returns the member id.

type Option source

type Option func(*settings) error

Option configures AcquireLeadership, DescribeRoleState, and RoleTopicPartition.

func WithChallengeStagger source

func WithChallengeStagger(stagger time.Duration) Option

WithChallengeStagger sets the extra delay that one rank of succession adds; the default is DefaultChallengeStagger.

func WithClock source

func WithClock(clock Clock) Option

WithClock reads the time from another clock than SystemClock. A test passes NewManualClock and drives every wait by hand.

func WithLeaseConfig source

func WithLeaseConfig(config LeaseConfig) Option

WithLeaseConfig takes a whole lease policy; the default is DefaultLeaseConfig.

func WithLeaseDuration source

func WithLeaseDuration(duration time.Duration) Option

WithLeaseDuration sets the extent of a lease; the default is DefaultLeaseDuration.

func WithPartitions source

func WithPartitions(partitions int) Option

WithPartitions sets the partition count of the coordination topic; the default is DefaultPartitions. Every member of a cluster passes the count that the cluster created the topic with, because the count picks the partition of a role.

func WithPollInterval source

func WithPollInterval(interval time.Duration) Option

WithPollInterval sets the longest gap between two reads of the coordination topic while a member waits; the default is DefaultPollInterval. A shorter gap finds a resigned holder sooner and costs more fetches.

func WithRenewInterval source

func WithRenewInterval(interval time.Duration) Option

WithRenewInterval sets the gap between two renewals by the holder; the default is DefaultRenewInterval.

func WithTopic source

func WithTopic(topic string) Option

WithTopic reads and writes another topic than StateTopic.

type RecordKind source

type RecordKind int16

RecordKind names the record of StateTopic that a key holds. The key carries the kind, so one topic holds both kinds and one compaction key covers both. 

const (
    // KindRegistration marks the record that one member writes to announce
    // that it is available. A registration carries no authority.
    KindRegistration RecordKind = 0

    // KindLease marks the record that names the member that holds a role now,
    // and the time the claim of that member ends.
    KindLease RecordKind = 1
)

func (RecordKind) String source

func (k RecordKind) String() string

String names the kind.

type Registrar source

type Registrar interface {
    // Append appends one record to the partition. The producer behind it
    // pins the partition, and it applies no partitioner of its own.
    Append(ctx context.Context, partition TopicPartition, key, value []byte) error
}

Registrar appends the registration record of one candidate.

This call carries no authority. A candidate holds no epoch when it announces itself, and it must take no epoch to do so, so the append sits outside a transaction. The offset that the broker assigns is the join sequence that the succession rules rank on.

type Registration source

type Registration struct {
    // Member is the member that registered.
    Member MemberID

    // RegisteredAt is the time the member registered, in milliseconds since
    // the Unix epoch.
    RegisteredAt int64
}

Registration is the value of a record that announces one member of a role. A member writes a registration when it joins, and it writes a tombstone on the same key when it leaves.

func DecodeRegistration source

func DecodeRegistration(data []byte) (Registration, error)

DecodeRegistration decodes the value of a registration record. It returns a *FormatError for a truncated buffer, for trailing bytes, for an unknown version, for a negative string length, and for a member id that its constructor rejects.

type Role source

type Role struct {
    // contains filtered or unexported fields
}

Role is the validated name of one coordination role, such as the name of an active controller.

A role becomes a Kafka transactional.id. The transaction coordinator mints one producer epoch for that id, so the role names the fenced group of members that compete for the same leadership.

NewRole is the only way to build a role with a name, so a non-zero Role value is proof that the name is well formed.

func NewRole source

func NewRole(name string) (Role, error)

NewRole parses a role name. It rejects an empty name and a name longer than MaxRoleLen bytes.

func (Role) String source

func (r Role) String() string

String returns the role name.

type RoleState source

type RoleState struct {
    // Roster holds the candidates of the role in registration order, which is
    // offset order.
    Roster []RosterEntry

    // Lease is the lease of the role, and nil when no member holds it.
    Lease *Lease
}

RoleState is the state that a reader folds out of the records of one role.

func BuildRoleState source

func BuildRoleState(role Role, records []StateRecord) (RoleState, error)

BuildRoleState folds the records of one partition into the state of one role. It reports the first record that does not decode.

The caller reads the partition of the role in offset order. The fold gives the same answer for another order, because it keeps the record of the highest offset for every key.

func DescribeRoleState source

func DescribeRoleState(ctx context.Context, reader StateReader, role Role, options ...Option) (RoleState, error)

DescribeRoleState reads the partition of one role and folds it into the roster and the lease of that role. A tool that reports the state of a cluster calls it, and it takes no epoch.

func (RoleState) Entry source

func (s RoleState) Entry(member MemberID) (RosterEntry, bool)

Entry returns the roster entry of a member, and it reports whether the member registered.

func (RoleState) Holder source

func (s RoleState) Holder() (MemberID, bool)

Holder returns the member that the lease names, and it reports whether the role has a lease.

The holder of an expired lease is still the holder. Ask LeaseTiming.LiveAt whether the lease is live.

func (RoleState) RankOf source

func (s RoleState) RankOf(member MemberID) (int, bool)

RankOf returns the challenge rank of a member, and it reports whether the member registered.

The rank is the index of the member in the roster after the removal of the current holder, so the first standby takes rank 0. The holder itself keeps rank 0. A holder reaches the rank test only after its own lease expired, and it still owns the newest epoch at that point, so it reclaims the role for less churn than a failover costs.

type RoleStateBuilder source

type RoleStateBuilder struct {
    // contains filtered or unexported fields
}

RoleStateBuilder folds the records of one role into a RoleState.

The builder keeps the record of the highest offset for every key, which is what log compaction keeps. A later registration for a member replaces the earlier one and moves the member to the tail of the roster. A registration tombstone removes the member. A lease tombstone clears the lease.

The builder drops a record of another role, so a caller folds a partition that holds several roles and needs no filter of its own.

func NewRoleStateBuilder source

func NewRoleStateBuilder(role Role) *RoleStateBuilder

NewRoleStateBuilder builds an empty fold for one role.

func (*RoleStateBuilder) Apply source

func (b *RoleStateBuilder) Apply(record StateRecord) error

Apply applies one record of the coordination partition. It drops a record of another role, and it reports a record that does not decode.

The key carries the identity, because the key is the compaction key. The builder takes the member and the record kind from the key, and it takes neither from the value.

func (*RoleStateBuilder) Build source

func (b *RoleStateBuilder) Build() RoleState

Build returns the role state, with the roster in offset order.

func (*RoleStateBuilder) Role source

func (b *RoleStateBuilder) Role() Role

Role returns the role that this fold collects.

type RosterEntry source

type RosterEntry struct {
    // Member is the member that registered.
    Member MemberID

    // Offset is the offset of the registration record of the member in the
    // partition of the role. The offset is the join sequence of the member,
    // because log compaction keeps the offset of every record it retains.
    Offset int64

    // RegisteredAt is the time the member registered, in milliseconds since
    // the Unix epoch.
    RegisteredAt int64
}

RosterEntry is one member of the roster of a role.

type StateReader source

type StateReader interface {
    // ReadPartition returns every committed record of the partition, oldest
    // first. The read takes committed records only, so an aborted lease write
    // stays invisible. The reader walks from the first offset of the
    // partition to the last stable offset.
    ReadPartition(ctx context.Context, partition TopicPartition) ([]StateRecord, error)
}

StateReader reads one partition of StateTopic. An adapter implements it with a plain assign, seek, and poll loop over the partition.

type StateRecord source

type StateRecord struct {
    // Offset is the offset of the record in its partition. The succession
    // rules rank candidates on the offset of a registration.
    Offset int64

    // Key is the encoded record key. [DecodeKey] reads it.
    Key []byte

    // Value is the encoded record value. An empty value is a tombstone. See
    // [IsTombstone].
    Value []byte
}

StateRecord is one record of StateTopic as it sits in the log. A StateReader returns these, and BuildRoleState folds them.

type SystemClock source

type SystemClock struct{}

SystemClock reads the clock of the host. It is the clock that AcquireLeadership takes by default.

func (SystemClock) After source

func (SystemClock) After(extent time.Duration) <-chan time.Time

After returns a channel that receives one value after the extent passes.

func (SystemClock) NowMillis source

func (SystemClock) NowMillis() int64

NowMillis returns the host time, in milliseconds since the Unix epoch.

type TopicPartition source

type TopicPartition struct {
    // Topic is the topic name.
    Topic string

    // Partition is the partition number.
    Partition int
}

TopicPartition identifies one partition of one topic. This package declares its own copy, because it depends on no other package of this library.

func RoleTopicPartition source

func RoleTopicPartition(role Role, options ...Option) (TopicPartition, error)

RoleTopicPartition returns the partition of the coordination topic that holds every record of the role. Only WithTopic and WithPartitions change the answer.

type Transport source

type Transport interface {
    Coordinator
    StateReader
    Registrar
    LeaseWriter
}

Transport is every cluster operation that this package performs, and it is nothing else. One adapter type satisfies all four interfaces.

The library pins no Kafka client, so a caller writes the adapter over the client it already runs. The adapter needs two producers for one role. The transactional producer carries transactional.id = <role>, and it writes the lease records of that role. A second, plain producer appends the registration records, because Kafka puts every send of a transactional producer inside a transaction, and a candidate holds no epoch to open one with.