leader

package
v1.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Dec 14, 2025 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Index

Constants

View Source
const (
	StateInit      = "INIT"
	StateCandidate = "CANDIDATE"
	StateLeader    = "LEADER"
	StateFollower  = "FOLLOWER"
	StateDemoted   = "DEMOTED"
	StateStopped   = "STOPPED"
)

Election state constants

Variables

View Source
var (
	// ErrNotLeader is returned when an operation requires leadership
	// but the instance is not currently the leader.
	ErrNotLeader = errors.New("not leader")

	// ErrAlreadyStarted is returned when attempting to start an election
	// that is already started.
	ErrAlreadyStarted = errors.New("already started")

	// ErrAlreadyStopped is returned when attempting to stop an election
	// that is already stopped.
	ErrAlreadyStopped = errors.New("already stopped")

	// ErrElectionFailed is returned when leader election fails.
	ErrElectionFailed = errors.New("election failed")

	// ErrHeartbeatFailed is returned when a heartbeat operation fails.
	ErrHeartbeatFailed = errors.New("heartbeat failed")

	// ErrTokenValidationFailed is returned when token validation fails.
	ErrTokenValidationFailed = errors.New("token validation failed")

	// ErrInvalidConfig is returned when the election configuration is invalid.
	ErrInvalidConfig = errors.New("invalid config")

	// ErrBucketNotFound is returned when the specified KV bucket does not exist.
	ErrBucketNotFound = errors.New("bucket not found")

	// ErrPermissionDenied is returned when access to the KV bucket is denied.
	ErrPermissionDenied = errors.New("permission denied")

	// ErrConnectionLost is returned when the NATS connection is lost.
	ErrConnectionLost = errors.New("connection lost")

	// ErrTokenInvalid is returned when a fencing token is invalid.
	ErrTokenInvalid = errors.New("fencing token is invalid")

	// ErrTokenMismatch is returned when a fencing token does not match.
	ErrTokenMismatch = errors.New("fencing token mismatch")
)

Package-level errors that can be returned by the library.

Functions

func CalculateBackoff

func CalculateBackoff(cfg BackoffConfig, attempt int) time.Duration

func CleanupKVBucket

func CleanupKVBucket(conn *nats.Conn, bucketName string) error

CleanupKVBucket deletes a KV bucket. If the bucket doesn't exist, it returns nil (no error). This is useful for cleaning up after integration tests.

func CreateKVBucket

func CreateKVBucket(conn *nats.Conn, bucketName string, ttl time.Duration) error

CreateKVBucket creates a KV bucket for testing. If the bucket already exists, it returns nil (no error). This is useful for integration tests where you need a KV bucket.

func EnsureKVBucket

func EnsureKVBucket(conn *nats.Conn, bucketName string, ttl time.Duration) (nats.KeyValue, error)

EnsureKVBucket ensures a KV bucket exists, creating it if necessary. This is a convenience function that combines CreateKVBucket and GetKVBucket.

func GetKVBucket

func GetKVBucket(conn *nats.Conn, bucketName string) (nats.KeyValue, error)

GetKVBucket gets an existing KV bucket. Returns an error if the bucket doesn't exist.

func IsPermanentError

func IsPermanentError(err error) bool

IsPermanentError classifies errors as permanent (demote immediately) vs transient (retry). Permanent errors should not be retried.

Permanent errors include:

  • Revision mismatch (another leader exists)
  • Key not found
  • Permission denied
  • Invalid configuration

Use this function to determine if an error should cause immediate demotion or if it should be retried with backoff.

Example:

if IsPermanentError(err) {
    // Demote immediately, don't retry
    election.demote()
} else {
    // Retry with backoff
    retryWithBackoff(err)
}

func IsTransientError

func IsTransientError(err error) bool

IsTransientError classifies errors as transient (retry with backoff) vs permanent (fail fast). Transient errors are typically network issues, timeouts, or temporary unavailability.

Transient errors include:

  • Timeouts
  • Connection lost/refused
  • Temporary unavailability
  • Network errors

Use this function to determine if an error should be retried with backoff.

Example:

if IsTransientError(err) {
    // Retry with exponential backoff
    retryWithBackoff(err)
} else {
    // Fail fast
    return err
}

func RetryWithBackoff

func RetryWithBackoff(ctx context.Context, cfg RetryConfig, fn func() error) error

func StartEmbeddedNATSServer

func StartEmbeddedNATSServer(ctx context.Context) (*server.Server, error)

StartEmbeddedNATSServer starts an embedded NATS server with JetStream enabled. The server uses a temporary directory for JetStream storage, which is cleaned up when the server is shut down.

The server will be automatically shut down when the context is cancelled. It's recommended to call StopEmbeddedNATSServer or server.Shutdown() for clean shutdown.

func StopEmbeddedNATSServer

func StopEmbeddedNATSServer(s *server.Server) error

StopEmbeddedNATSServer properly shuts down the embedded NATS server and cleans up resources. This function should be called when you're done with the server to ensure clean shutdown.

After calling this function, the server cannot be used anymore.

func WaitForCondition

func WaitForCondition(t *testing.T, condition func() bool, timeout time.Duration, msg string)

WaitForCondition waits for a condition to become true within the given timeout. This is a test helper function that polls the condition at 10ms intervals.

Example:

WaitForCondition(t, func() bool {
    return election.IsLeader()
}, 5*time.Second, "election to become leader")

func WaitForHeartbeat

func WaitForHeartbeat(t *testing.T, election Election, initialHeartbeat time.Time, timeout time.Duration)

WaitForHeartbeat waits for a heartbeat to occur after the initial heartbeat time. This verifies that the heartbeat loop is running and updating the leadership key.

Example:

initialHeartbeat := election.Status().LastHeartbeat
WaitForHeartbeat(t, election, initialHeartbeat, 2*time.Second)

func WaitForLeader

func WaitForLeader(t *testing.T, election Election, expectLeader bool, timeout time.Duration)

WaitForLeader waits for an election to reach the expected leader state. This is a convenience wrapper around WaitForCondition.

Example:

WaitForLeader(t, election, true, 5*time.Second)  // Wait to become leader
WaitForLeader(t, election, false, 5*time.Second) // Wait to become follower

Types

type BackoffConfig

type BackoffConfig struct {
	InitialBackoff    time.Duration
	MaxBackoff        time.Duration
	BackoffMultiplier float64
	Jitter            float64
}

func DefaultBackoffConfig

func DefaultBackoffConfig() BackoffConfig

type CircuitBreaker

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

func NewCircuitBreaker

func NewCircuitBreaker(failureThreshold int, cooldownPeriod time.Duration) *CircuitBreaker

func (*CircuitBreaker) Call

func (cb *CircuitBreaker) Call(fn func() error) error

type CircuitState

type CircuitState int
const (
	CircuitStateClosed CircuitState = iota
	CircuitStateOpen
	CircuitStateHalfOpen
)

type ConnectionMonitor

type ConnectionMonitor interface {
	Start(ctx context.Context) error
	Stop() error
	Status() ConnectionStatus
	OnDisconnect(func())
	OnReconnect(func())
	SetStatus(ConnectionStatus) // Set status (used after verification)
}

func NewNATSConnectionMonitor

func NewNATSConnectionMonitor(conn *nats.Conn) ConnectionMonitor

type ConnectionStatus

type ConnectionStatus int
const (
	ConnectionStatusConnected ConnectionStatus = iota
	ConnectionStatusDisconnected
	ConnectionStatusReconnected
	ConnectionStatusClosed
)

type Election

type Election interface {
	// Start begins participating in the election.
	// This method starts background goroutines for:
	//   - Attempting to acquire leadership
	//   - Watching for leader changes (if follower)
	//   - Sending heartbeats (if leader)
	//   - Validating tokens (if leader)
	//   - Monitoring connection health
	//
	// Start returns an error if:
	//   - The election is already started (ErrAlreadyStarted)
	//   - The configuration is invalid (ErrInvalidConfig)
	//   - The NATS connection cannot access JetStream
	//
	// The context is used for cancellation. If the context is cancelled,
	// the election will stop gracefully.
	//
	// Example:
	//   ctx := context.Background()
	//   if err := election.Start(ctx); err != nil {
	//       log.Fatal(err)
	//   }
	Start(ctx context.Context) error

	// Stop stops participating in the election and cleans up resources.
	// This method:
	//   - Stops all background goroutines
	//   - Does NOT delete the leadership key (if leader)
	//   - Does NOT wait for OnDemote callback
	//
	// For graceful shutdown with options, use StopWithContext instead.
	//
	// Stop returns an error if the election is already stopped (ErrAlreadyStopped).
	//
	// Example:
	//   if err := election.Stop(); err != nil {
	//       log.Printf("Error stopping election: %v", err)
	//   }
	Stop() error

	// StopWithContext stops the election with configurable options.
	// This method provides graceful shutdown capabilities:
	//   - Optional key deletion for fast failover
	//   - Optional waiting for OnDemote callback
	//   - Timeout handling via context
	//
	// Options:
	//   - DeleteKey: If true, deletes the leadership key immediately,
	//     allowing a new leader to be elected without waiting for TTL expiration.
	//   - WaitForDemote: If true, waits for the OnDemote callback to complete
	//     before returning. This ensures leader tasks are stopped before shutdown.
	//   - Timeout: Maximum time to wait. If 0, uses context timeout or 5s default.
	//
	// Example:
	//   ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	//   defer cancel()
	//   err := election.StopWithContext(ctx, StopOptions{
	//       DeleteKey:      true,
	//       WaitForDemote:  true,
	//   })
	StopWithContext(ctx context.Context, opts StopOptions) error

	// Status returns the current status of the election.
	// This includes:
	//   - Current state (INIT, CANDIDATE, LEADER, FOLLOWER, DEMOTED, STOPPED)
	//   - Whether this instance is the leader
	//   - Current leader ID
	//   - Fencing token (if leader)
	//   - Last heartbeat time (if leader)
	//   - Last state transition time
	//   - Current KV revision (if leader)
	//   - NATS connection status
	//
	// Example:
	//   status := election.Status()
	//   if status.IsLeader {
	//       log.Printf("I am the leader, token: %s", status.Token)
	//   } else {
	//       log.Printf("Current leader: %s", status.LeaderID)
	//   }
	Status() ElectionStatus

	// IsLeader returns true if this instance currently holds leadership.
	// This is a convenience method equivalent to Status().IsLeader.
	//
	// Example:
	//   if election.IsLeader() {
	//       performLeaderTasks()
	//   }
	IsLeader() bool

	// LeaderID returns the instance ID of the current leader.
	// Returns empty string if no leader is known.
	// This may be this instance's ID (if leader) or another instance's ID (if follower).
	//
	// Example:
	//   leaderID := election.LeaderID()
	//   if leaderID == "" {
	//       log.Println("No leader known")
	//   } else if leaderID == myInstanceID {
	//       log.Println("I am the leader")
	//   } else {
	//       log.Printf("Leader is: %s", leaderID)
	//   }
	LeaderID() string

	// Token returns the current fencing token for this instance if it is the leader.
	// Returns empty string if not leader.
	//
	// The fencing token is a UUID that uniquely identifies the current leadership.
	// It should be validated before performing critical operations to ensure
	// this instance is still the leader.
	//
	// Example:
	//   if election.IsLeader() {
	//       token := election.Token()
	//       // Use token for fencing in external systems
	//   }
	Token() string

	// OnPromote registers a callback that is executed when this instance becomes leader.
	// The callback receives:
	//   - A context that is cancelled when leadership is lost
	//   - The fencing token for this leadership term
	//
	// The callback is called in a separate goroutine, so it should not block.
	// If the callback panics, the panic is caught and logged, but does not
	// affect the election process.
	//
	// Multiple calls to OnPromote replace the previous callback.
	//
	// Example:
	//   election.OnPromote(func(ctx context.Context, token string) {
	//       log.Printf("Became leader with token: %s", token)
	//       go startLeaderTasks(ctx, token)
	//   })
	OnPromote(func(ctx context.Context, token string))

	// OnDemote registers a callback that is executed when this instance loses leadership.
	// The callback is called in a separate goroutine, so it should not block.
	// If the callback panics, the panic is caught and logged, but does not
	// affect the election process.
	//
	// Multiple calls to OnDemote replace the previous callback.
	//
	// Example:
	//   election.OnDemote(func() {
	//       log.Println("Lost leadership")
	//       stopLeaderTasks()
	//   })
	OnDemote(func())

	// ValidateToken validates the current token against the KV store.
	// Returns true if the token is valid (this instance is still leader),
	// false if invalid (this instance is no longer leader or not leader).
	//
	// This method performs a synchronous check by reading from the KV store.
	// For periodic validation, use ValidationInterval in ElectionConfig instead.
	//
	// Example:
	//   valid, err := election.ValidateToken(ctx)
	//   if err != nil {
	//       log.Printf("Error validating token: %v", err)
	//   } else if !valid {
	//       log.Println("Token invalid - no longer leader")
	//   }
	ValidateToken(ctx context.Context) (bool, error)

	// ValidateTokenOrDemote validates the token and demotes if invalid.
	// Returns true if the token is valid, false if invalid (and demoted).
	//
	// This is a convenience method that combines ValidateToken with automatic
	// demotion. Use this before critical operations to ensure this instance
	// is still the leader.
	//
	// Example:
	//   if !election.ValidateTokenOrDemote(ctx) {
	//       log.Println("Token invalid, operation aborted")
	//       return
	//   }
	//   // Perform critical operation...
	ValidateTokenOrDemote(ctx context.Context) bool
}

Election is the main interface for participating in leader election.

An Election instance represents a single participant in a leader election for a specific group/role. Multiple instances can participate in the same election, but only one will be leader at a time.

Example usage:

cfg := ElectionConfig{
    Bucket:            "leaders",
    Group:             "control-manager",
    InstanceID:        "instance-1",
    TTL:               10 * time.Second,
    HeartbeatInterval: 1 * time.Second,
}
election, err := NewElectionWithConn(nc, cfg)
if err != nil {
    log.Fatal(err)
}

election.OnPromote(func(ctx context.Context, token string) {
    log.Println("Became leader, token:", token)
})

election.OnDemote(func() {
    log.Println("Lost leadership")
})

if err := election.Start(context.Background()); err != nil {
    log.Fatal(err)
}
defer election.Stop()

func NewElection

func NewElection(nc JetStreamProvider, cfg ElectionConfig) (Election, error)

NewElection creates a new election instance with a JetStream provider. This is the low-level constructor that accepts any JetStreamProvider. For most use cases, use NewElectionWithConn instead.

The configuration is validated before creating the election instance. Returns an error if:

  • Configuration is invalid (ErrInvalidConfig)
  • JetStream is not available

Example:

adapter := &myJetStreamProvider{}
election, err := NewElection(adapter, cfg)
if err != nil {
    log.Fatal(err)
}

func NewElectionWithConn

func NewElectionWithConn(nc *nats.Conn, cfg ElectionConfig) (Election, error)

NewElectionWithConn creates a new election instance with a NATS connection. This is a convenience function that wraps the connection.

The configuration is validated before creating the election instance. Returns an error if:

  • Configuration is invalid (ErrInvalidConfig)
  • JetStream is not available

Example:

nc, err := nats.Connect(nats.DefaultURL)
if err != nil {
    log.Fatal(err)
}
election, err := NewElectionWithConn(nc, cfg)
if err != nil {
    log.Fatal(err)
}

type ElectionConfig

type ElectionConfig struct {
	// Bucket is the name of the NATS KV bucket to use for leadership storage.
	// The bucket must exist before creating an election instance.
	// Example: "leaders", "elections", "my-app-leaders"
	Bucket string

	// Group is the election group/role identifier.
	// Multiple instances with the same Group participate in the same election.
	// Different Groups represent independent elections (multi-role support).
	// Example: "control-manager", "scheduler", "coordinator"
	Group string

	// InstanceID is a unique identifier for this instance.
	// Must be unique across all instances participating in the same election.
	// Typically: hostname, pod name, container ID, or UUID.
	// Example: "instance-1", "pod-abc123", "host-192.168.1.1"
	InstanceID string

	// TTL is the time-to-live for the leadership key in the KV store.
	// The key will expire if not refreshed within this duration.
	// Must be >= HeartbeatInterval * 3 (enforced by validation).
	// Recommended: 3-5x HeartbeatInterval for safety margin.
	// Example: 10 * time.Second (with HeartbeatInterval: 1 * time.Second)
	TTL time.Duration

	// HeartbeatInterval is how often the leader refreshes the leadership key.
	// Must be < TTL / 3 (enforced by validation).
	// Shorter intervals provide faster failover but more KV operations.
	// Recommended: 1-2 seconds for most use cases.
	// Example: 1 * time.Second
	HeartbeatInterval time.Duration

	// Logger is an optional structured logger for observability.
	// If nil, logging is disabled (no-op logger).
	// Implementations should use zap.Field for structured data.
	// Example: zap.NewProduction(), zap.NewDevelopment()
	Logger Logger

	// ValidationInterval is how often to validate the fencing token in the background.
	// If 0, background validation is disabled (not recommended for production).
	// Recommended: 5-10 seconds for most use cases.
	// Example: 5 * time.Second
	ValidationInterval time.Duration

	// DisconnectGracePeriod is how long to wait before demoting on NATS disconnect.
	// If 0, uses default: max(3 * HeartbeatInterval, 5 seconds).
	// During this period, the leader continues to act as leader, allowing
	// for brief network interruptions without triggering failover.
	// Example: 5 * time.Second
	DisconnectGracePeriod time.Duration

	// Metrics is an optional metrics recorder for observability.
	// If nil, metrics recording is disabled.
	// Use NewPrometheusMetrics() to create a Prometheus implementation.
	// Example: leader.NewPrometheusMetrics(prometheus.DefaultRegisterer)
	Metrics Metrics

	// HealthChecker is an optional health checker for self-monitoring.
	// If nil, health checking is disabled.
	// If provided, health is checked before each heartbeat.
	// If health check fails MaxConsecutiveFailures times consecutively,
	// the leader will automatically demote itself.
	// Example: &MyHealthChecker{}
	HealthChecker HealthChecker

	// MaxConsecutiveFailures is the maximum number of consecutive health check
	// failures before automatic demotion.
	// Only used if HealthChecker is set.
	// Default: 3
	// Example: 3
	MaxConsecutiveFailures int

	// Priority is used for leader selection when AllowPriorityTakeover is true.
	// Higher number = higher priority.
	// Default: 0 (no priority, first-come-first-served).
	// Must be > 0 if AllowPriorityTakeover is true (enforced by validation).
	// Example: 10 (lower priority), 20 (higher priority)
	Priority int

	// AllowPriorityTakeover enables priority-based preemption.
	// If true, a higher-priority instance can take over from a lower-priority leader.
	// If false, priority is ignored and first-come-first-served applies.
	// Default: false (disabled for safety).
	//
	// WARNING: Priority takeover can cause service disruption. Use with caution.
	// Prefer voluntary demotion (health checks, graceful shutdown) over forced preemption.
	//
	// Use cases:
	//   - Primary/backup datacenter scenarios
	//   - Maintenance mode (lower priority to allow takeover)
	//   - Clear priority hierarchy
	//
	// Example: true (enable priority takeover)
	AllowPriorityTakeover bool
}

ElectionConfig configures an election instance.

Required fields must be set. Optional fields have sensible defaults. The configuration is validated when creating an election instance.

Example:

cfg := ElectionConfig{
    // Required
    Bucket:            "leaders",
    Group:             "control-manager",
    InstanceID:        "instance-1",
    TTL:               10 * time.Second,
    HeartbeatInterval: 1 * time.Second,

    // Optional
    ValidationInterval: 5 * time.Second,
    Logger:            myLogger,
    Metrics:           myMetrics,
}

type ElectionError

type ElectionError struct {
	// Code is a short error code identifying the error type.
	Code string

	// InstanceID is the instance ID where the error occurred.
	InstanceID string

	// Reason is a human-readable description of the error.
	Reason string

	// Err is the underlying error, if any.
	Err error
}

ElectionError represents an error that occurred during an election operation. It provides structured error information including error code, instance ID, reason, and the underlying error.

Use errors.Is() and errors.As() to check for specific error types:

var electionErr *ElectionError
if errors.As(err, &electionErr) {
    log.Printf("Election error [%s]: %s", electionErr.Code, electionErr.Reason)
}

func NewElectionError

func NewElectionError(code string, instanceID string, reason string, err error) *ElectionError

NewElectionError creates a new ElectionError with the given parameters.

Example:

err := NewElectionError(
    "ACQUIRE_FAILED",
    "instance-1",
    "key already exists",
    originalErr,
)

func (*ElectionError) Error

func (e *ElectionError) Error() string

func (*ElectionError) Unwrap

func (e *ElectionError) Unwrap() error

type ElectionStatus

type ElectionStatus struct {
	// State is the current state of the election.
	// Possible values: INIT, CANDIDATE, LEADER, FOLLOWER, DEMOTED, STOPPED
	State string

	// IsLeader is true if this instance is currently the leader.
	IsLeader bool

	// LeaderID is the instance ID of the current leader.
	// Empty if no leader is known.
	LeaderID string

	// Token is the fencing token for this instance if it is the leader.
	// Empty if not leader.
	Token string

	// LastHeartbeat is the time of the last successful heartbeat (if leader).
	// Zero if not leader or no heartbeat yet.
	LastHeartbeat time.Time

	// LastTransition is the time of the last state transition.
	LastTransition time.Time

	// Revision is the current KV revision (if leader).
	// Zero if not leader.
	Revision uint64
}

ElectionStatus contains the current status of an election instance. Use Election.Status() to get the current status.

Example:

status := election.Status()
log.Printf("State: %s, IsLeader: %v, LeaderID: %s",
    status.State, status.IsLeader, status.LeaderID)

type Entry

type Entry interface {
	Key() string
	Value() []byte
	Revision() uint64
}

Entry represents a key-value entry in the KV store. This is an abstraction over NATS KV entries to enable testing.

type HealthChecker

type HealthChecker interface {
	// Check returns true if the instance is healthy and should retain leadership.
	// Return false if the instance is unhealthy and should demote.
	//
	// If health check fails MaxConsecutiveFailures times in a row,
	// the leader will automatically demote itself.
	Check(ctx context.Context) bool
}

HealthChecker defines the interface for health checking. This is OPTIONAL - if not provided (nil), health checking is disabled.

If you want health checking, implement this interface. The Check method is called before each heartbeat to verify the leader's health.

Examples:

  • If you have no resources to check, you can return true (always healthy)
  • If you have a database, check if it's accessible
  • If you have external dependencies, verify they're reachable

The context can be used for timeout handling and will be cancelled if the election is stopped. Implementations should be fast (< 100ms ideally).

type JetStreamContext

type JetStreamContext interface {
	KeyValue(bucket string) (KeyValue, error)
}

JetStreamContext is an abstraction over NATS JetStream context. This interface enables testing with mocks.

type JetStreamProvider

type JetStreamProvider interface {
	JetStream() (JetStreamContext, error)
}

JetStreamProvider is an interface for accessing NATS JetStream. This allows the library to work with different NATS connection types and enables testing with mocks.

The nats.Conn type implements this interface via its JetStream() method.

type KeyValue

type KeyValue interface {
	Create(key string, value []byte, opts ...interface{}) (uint64, error)
	Update(key string, value []byte, rev uint64, opts ...interface{}) (uint64, error)
	Get(key string) (Entry, error)
	Delete(key string) error
	Watch(key string, opts ...interface{}) (Watcher, error)
}

KeyValue is an abstraction over NATS KV operations. This interface enables testing with mocks.

type Logger

type Logger interface {
	Debug(msg string, fields ...zap.Field)
	Info(msg string, fields ...zap.Field)
	Warn(msg string, fields ...zap.Field)
	Error(msg string, fields ...zap.Field)
	Fatal(msg string, fields ...zap.Field)
}

Logger defines the interface for structured logging. Implementations should use zap.Field for structured data.

type Metrics

type Metrics interface {
	// Gauges
	SetIsLeader(value float64, labels prometheus.Labels)
	SetConnectionStatus(value float64, labels prometheus.Labels)

	// Counters
	IncTransitions(labels prometheus.Labels)
	IncFailures(labels prometheus.Labels)
	IncAcquireAttempts(labels prometheus.Labels)
	IncTokenValidationFailures(labels prometheus.Labels)

	// Histograms
	ObserveHeartbeatDuration(duration time.Duration, labels prometheus.Labels)
	ObserveLeaderDuration(duration time.Duration, labels prometheus.Labels)
}

Metrics defines the interface for recording metrics. All metrics are optional - if nil, metrics recording is disabled.

func NewPrometheusMetrics

func NewPrometheusMetrics(registry prometheus.Registerer) Metrics

NewPrometheusMetrics creates a new Prometheus metrics implementation. If registry is nil, returns a no-op implementation that discards all metrics.

The returned Metrics instance records the following metrics:

  • election_is_leader (gauge): 1 if leader, 0 otherwise
  • election_connection_status (gauge): 1 if connected, 0 otherwise
  • election_transitions_total (counter): State transitions
  • election_failures_total (counter): Election failures
  • election_acquire_attempts_total (counter): Acquisition attempts
  • election_token_validation_failures_total (counter): Token validation failures
  • election_heartbeat_duration_seconds (histogram): Heartbeat duration
  • election_leader_duration_seconds (histogram): Leadership duration

Example:

registry := prometheus.NewRegistry()
metrics := NewPrometheusMetrics(registry)
cfg := ElectionConfig{
    // ... other fields ...
    Metrics: metrics,
}

type MockConnAdapter

type MockConnAdapter struct {
	Conn *natsmock.MockConn
}

MockConnAdapter adapts *natsmock.MockConn to leader.JetStreamProvider interface.

func NewMockConnAdapter

func NewMockConnAdapter(conn *natsmock.MockConn) *MockConnAdapter

NewMockConnAdapter creates a new MockConnAdapter. This is the main entry point for creating mock adapters in tests.

Example:

nc := natsmock.NewMockConn()
adapter := NewMockConnAdapter(nc)
election, err := leader.NewElection(adapter, cfg)

func (*MockConnAdapter) JetStream

func (a *MockConnAdapter) JetStream() (JetStreamContext, error)

JetStream returns a JetStream context.

type MockEntryAdapter

type MockEntryAdapter struct {
	Entry natsmock.Entry
}

MockEntryAdapter adapts natsmock.Entry to leader.Entry interface.

func (*MockEntryAdapter) Key

func (a *MockEntryAdapter) Key() string

Key returns the entry key.

func (*MockEntryAdapter) Revision

func (a *MockEntryAdapter) Revision() uint64

Revision returns the entry revision.

func (*MockEntryAdapter) Value

func (a *MockEntryAdapter) Value() []byte

Value returns the entry value.

type MockJetStreamAdapter

type MockJetStreamAdapter struct {
	JS *natsmock.MockJetStream
}

MockJetStreamAdapter adapts *natsmock.MockJetStream to leader.JetStreamContext interface.

func (*MockJetStreamAdapter) KeyValue

func (a *MockJetStreamAdapter) KeyValue(bucket string) (KeyValue, error)

KeyValue returns a KeyValue store for the given bucket.

type MockKeyValueAdapter

type MockKeyValueAdapter struct {
	KV *natsmock.MockKeyValue
}

MockKeyValueAdapter adapts *natsmock.MockKeyValue to leader.KeyValue interface. This adapter enables using natsmock in tests that require the leader.KeyValue interface.

func (*MockKeyValueAdapter) Create

func (a *MockKeyValueAdapter) Create(key string, value []byte, opts ...interface{}) (uint64, error)

Create creates a key-value entry.

func (*MockKeyValueAdapter) Delete

func (a *MockKeyValueAdapter) Delete(key string) error

Delete deletes a key-value entry.

func (*MockKeyValueAdapter) Get

func (a *MockKeyValueAdapter) Get(key string) (Entry, error)

Get retrieves a key-value entry.

func (*MockKeyValueAdapter) Update

func (a *MockKeyValueAdapter) Update(key string, value []byte, rev uint64, opts ...interface{}) (uint64, error)

Update updates an existing key-value entry.

func (*MockKeyValueAdapter) Watch

func (a *MockKeyValueAdapter) Watch(key string, opts ...interface{}) (Watcher, error)

Watch watches for changes to a key.

type MockWatcherAdapter

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

MockWatcherAdapter adapts natsmock.Watcher to leader.Watcher interface.

func NewMockWatcherAdapter

func NewMockWatcherAdapter(watcher natsmock.Watcher) *MockWatcherAdapter

NewMockWatcherAdapter creates a new MockWatcherAdapter.

func (*MockWatcherAdapter) Stop

func (a *MockWatcherAdapter) Stop()

Stop stops the watcher.

func (*MockWatcherAdapter) Updates

func (a *MockWatcherAdapter) Updates() <-chan Entry

Updates returns a channel that receives entry updates. This method initializes the channel and goroutine only once, even if called multiple times.

type NATSConnectionProvider

type NATSConnectionProvider interface {
	NATSConnection() *nats.Conn
}

NATSConnectionProvider provides access to the underlying NATS connection for connection monitoring. This is optional - if not implemented, connection monitoring will be disabled.

The nats.Conn type can be wrapped to implement this interface. Connection monitoring enables automatic demotion on disconnect.

type RetryConfig

type RetryConfig struct {
	MaxAttempts    int
	BackoffConfig  BackoffConfig
	CircuitBreaker *CircuitBreaker
}

type StopOptions

type StopOptions struct {
	// DeleteKey deletes the leadership key on stop for fast failover.
	// If true, the key is deleted immediately, allowing a new leader
	// to be elected without waiting for TTL expiration.
	// If false, the key remains until TTL expires (slower failover).
	// Recommended: true for graceful shutdowns, false for crashes.
	DeleteKey bool

	// WaitForDemote waits for the OnDemote callback to complete before
	// returning. If true, ensures leader tasks are stopped before shutdown.
	// If false, the callback is called but not waited for (faster shutdown).
	// Recommended: true for graceful shutdowns.
	WaitForDemote bool

	// Timeout is the maximum time to wait for shutdown operations.
	// If 0, the context timeout is used. If context has no timeout,
	// a default of 5 seconds is used.
	// Recommended: 10-30 seconds depending on cleanup time.
	Timeout time.Duration
}

StopOptions configures the behavior of StopWithContext.

Example:

opts := StopOptions{
    DeleteKey:      true,  // Fast failover
    WaitForDemote:  true,  // Wait for cleanup
    Timeout:        10 * time.Second,
}
err := election.StopWithContext(ctx, opts)

type TimeoutError

type TimeoutError struct {
	// Operation is the name of the operation that timed out.
	Operation string

	// Timeout is the timeout duration that was exceeded.
	Timeout time.Duration

	// Err is the underlying error, if any.
	Err error
}

TimeoutError represents an error that occurred due to a timeout. It provides information about which operation timed out and the timeout duration.

Use errors.Is() to check for timeout errors:

if errors.Is(err, &TimeoutError{}) {
    log.Println("Operation timed out")
}

Or check for a specific timeout error:

var timeoutErr *TimeoutError
if errors.As(err, &timeoutErr) {
    log.Printf("Operation %s timed out after %v",
        timeoutErr.Operation, timeoutErr.Timeout)
}

func NewTimeoutError

func NewTimeoutError(operation string, timeout time.Duration, err error) *TimeoutError

NewTimeoutError creates a new TimeoutError with the given parameters.

Example:

err := NewTimeoutError("heartbeat update", 5*time.Second, nil)

func (*TimeoutError) Error

func (e *TimeoutError) Error() string

func (*TimeoutError) Is

func (e *TimeoutError) Is(target error) bool

func (*TimeoutError) Unwrap

func (e *TimeoutError) Unwrap() error

type TimingConfig

type TimingConfig struct {
	// Configuration values
	TTL                   time.Duration
	HeartbeatInterval     time.Duration
	DisconnectGracePeriod time.Duration
	ValidationInterval    time.Duration

	// Estimated operation delays
	InitialJitter         time.Duration // 10-100ms
	KVOperationDelay      time.Duration // ~10-50ms
	MaxRetryBackoff       time.Duration // up to 350ms (50ms + 100ms + 200ms)
	PeriodicCheckInterval time.Duration // 500ms (worst case)
	NATSPropagationBuffer time.Duration // ~500ms-1s for NATS propagation
}

TimingConfig holds timing parameters for chaos test scenarios

func DefaultTimingConfig

func DefaultTimingConfig() TimingConfig

DefaultTimingConfig returns a default timing configuration

func (TimingConfig) CalculateImmediateDeletionTimeout

func (tc TimingConfig) CalculateImmediateDeletionTimeout() (sleepDuration, waitTimeout time.Duration)

CalculateImmediateDeletionTimeout calculates the timeout for immediate key deletion scenarios This is used when DeleteKey=true (graceful shutdown)

func (TimingConfig) CalculateInitialAcquisitionTimeout

func (tc TimingConfig) CalculateInitialAcquisitionTimeout() time.Duration

CalculateInitialAcquisitionTimeout calculates the timeout for initial leader acquisition This accounts for jitter, KV operations, and potential retries

func (TimingConfig) CalculateNetworkPartitionTimeout

func (tc TimingConfig) CalculateNetworkPartitionTimeout() (sleepDuration, waitTimeout time.Duration)

CalculateNetworkPartitionTimeout calculates the timeout for network partition scenarios This accounts for grace period, TTL expiration, detection delays, and NATS propagation

func (TimingConfig) CalculateTTLExpirationTimeout

func (tc TimingConfig) CalculateTTLExpirationTimeout() (sleepDuration, waitTimeout time.Duration)

CalculateTTLExpirationTimeout calculates the timeout for TTL expiration scenarios This is used when a leader dies without graceful shutdown (no DeleteKey)

type TokenValidationError

type TokenValidationError struct {
	// LocalToken is the token stored locally by this instance.
	LocalToken string

	// KvToken is the token stored in the KV store.
	KvToken string

	// LeaderID is the current leader ID from the KV store.
	LeaderID string

	// Reason is a human-readable description of why validation failed.
	Reason string

	// Err is the underlying error, if any.
	Err error
}

TokenValidationError represents an error that occurred during token validation. It provides detailed information about the token mismatch.

Use errors.Is() and errors.As() to check for this error type:

var tokenErr *TokenValidationError
if errors.As(err, &tokenErr) {
    log.Printf("Token validation failed: local=%s, kv=%s",
        tokenErr.LocalToken, tokenErr.KvToken)
}

func (*TokenValidationError) Error

func (e *TokenValidationError) Error() string

func (*TokenValidationError) Unwrap

func (e *TokenValidationError) Unwrap() error

type ValidationError

type ValidationError struct {
	// Field is the name of the configuration field that failed validation.
	Field string

	// Value is the invalid value that was provided.
	Value interface{}

	// Reason is a human-readable explanation of why validation failed.
	Reason string

	// Err is the underlying error, if any.
	Err error
}

ValidationError represents a configuration validation error. It provides detailed information about which field failed validation and why.

Use errors.Is() and errors.As() to check for this error type:

var validationErr *ValidationError
if errors.As(err, &validationErr) {
    log.Printf("Invalid config: field %s = %v: %s",
        validationErr.Field, validationErr.Value, validationErr.Reason)
}

func NewValidationError

func NewValidationError(field string, value interface{}, reason string) *ValidationError

NewValidationError creates a new ValidationError with the given parameters.

Example:

err := NewValidationError("TTL", 1*time.Second,
    "must be >= HeartbeatInterval * 3")

func (*ValidationError) Error

func (e *ValidationError) Error() string

func (*ValidationError) Unwrap

func (e *ValidationError) Unwrap() error

type Watcher

type Watcher interface {
	Updates() <-chan Entry
	Stop()
}

Watcher watches for changes to a key in the KV store. This is an abstraction over NATS KV watchers to enable testing.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL