policy

package
v1.6.0 Latest Latest
Warning

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

 Go to latest Published: Jul 30, 2026 License: Apache-2.0 Imports: 13 Imported by: 0

Documentation

Overview

Package policy provides read strategies, write strategies, and failover policies for the helix dual-database client.

Read Strategies

Read strategies determine how read operations are routed between clusters. All strategies implement the ReadStrategy interface: 

type ReadStrategy interface {
    Select(ctx context.Context) types.ClusterID
}

Available strategies:

  • StickyRead: Routes all reads to a randomly selected cluster for cache affinity
  • PrimaryOnlyRead: Always routes reads to ClusterA
  • RoundRobinRead: Alternates between clusters for load distribution

Example: 

client, _ := helix.NewCQLClient(sessionA, sessionB,
    helix.WithReadStrategy(policy.NewStickyRead()),
)

Write Strategies

Write strategies determine how write operations are executed across clusters. All strategies implement the WriteStrategy interface: 

type WriteStrategy interface {
    Execute(
        ctx context.Context,
        writeA func(context.Context) error,
        writeB func(context.Context) error,
    ) (errA, errB error)
}

Available strategies:

Example: 

client, _ := helix.NewCQLClient(sessionA, sessionB,
    helix.WithWriteStrategy(policy.NewConcurrentDualWrite()),
)

For latency-aware dual writes that detect slow clusters: 

client, _ := helix.NewCQLClient(sessionA, sessionB,
    helix.WithWriteStrategy(policy.NewAdaptiveDualWrite(
        policy.WithAdaptiveDeltaThreshold(300 * time.Millisecond),
        policy.WithAdaptiveAbsoluteMax(2 * time.Second),
    )),
)

Failover Policies

Failover policies control how the client responds to cluster failures. All policies implement the FailoverPolicy interface: 

type FailoverPolicy interface {
    ShouldFailover(cluster types.ClusterID, err error) bool
    RecordSuccess(cluster types.ClusterID)
    RecordFailure(cluster types.ClusterID)
}

Available policies:

Example: 

client, _ := helix.NewCQLClient(sessionA, sessionB,
    helix.WithFailoverPolicy(policy.NewActiveFailover()),
)

For latency-aware failover that treats slow responses (>2s) as failures: 

lcb := policy.NewLatencyCircuitBreaker(
    policy.WithLatencyAbsoluteMax(2 * time.Second),
    policy.WithLatencyThreshold(3), // 3 failures before circuit opens
)
client, _ := helix.NewCQLClient(sessionA, sessionB,
    helix.WithFailoverPolicy(lcb),
)
// After each read, record latency:
// lcb.RecordLatency(cluster, latency)

Package policy provides read and write strategies for Helix dual-cluster operations.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type ActiveFailover 

type ActiveFailover struct{}

ActiveFailover implements an aggressive failover policy.

On any failure, immediately attempts failover to the secondary cluster. ShouldFailover always returns true — every error triggers a failover attempt, with no delay, threshold, or backoff.

WARNING: Oscillation risk. If both clusters are intermittently failing, ActiveFailover will flip-flop between them on every request, producing rapid and noisy failover transitions. In flaky dual-cluster scenarios, prefer CircuitBreaker or LatencyCircuitBreaker which require a threshold of consecutive failures before opening, dampening oscillation.

func NewActiveFailover 

func NewActiveFailover() *ActiveFailover

NewActiveFailover creates a new ActiveFailover policy.

Returns:

  • *ActiveFailover: A new active failover policy

func (*ActiveFailover) RecordFailure 

func (a *ActiveFailover) RecordFailure(_ types.ClusterID)

RecordFailure is a no-op for active failover.

Parameters:

  • cluster: The cluster that failed (unused)

func (*ActiveFailover) RecordSuccess 

func (a *ActiveFailover) RecordSuccess(_ types.ClusterID)

RecordSuccess is a no-op for active failover.

Parameters:

  • cluster: The cluster that succeeded (unused)

func (*ActiveFailover) ShouldFailover 

func (a *ActiveFailover) ShouldFailover(_ types.ClusterID, _ error) bool

ShouldFailover always returns true for active failover.

Every error triggers an immediate failover attempt regardless of cause or frequency. See the ActiveFailover type documentation for the oscillation risk this implies when both clusters are degraded.

Parameters:

  • cluster: The cluster that failed (unused)
  • err: The error (unused)

Returns:

  • bool: Always true

type AdaptiveDualWrite 

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

AdaptiveDualWrite implements a latency-aware concurrent dual-write strategy.

This strategy monitors the relative performance of both clusters and adapts its behavior accordingly:

  • Healthy cluster: Wait for completion, record latency
  • Degraded cluster: Fire-and-forget (don't block), rely on replay

A cluster is marked as degraded when:

  1. Its latency exceeds absoluteMax (e.g., 2s), OR
  2. It is consistently slower than its sibling by more than deltaThreshold (e.g., 150ms) for strikeThreshold consecutive writes

The "min floor" filter ignores relative differences when both clusters are fast (< minFloor), preventing false positives from minor variations.

Example: 

strategy := policy.NewAdaptiveDualWrite(
    policy.WithAdaptiveDeltaThreshold(150 * time.Millisecond),
    policy.WithAdaptiveAbsoluteMax(2 * time.Second),
    policy.WithAdaptiveStrikeThreshold(3),
)

Zero value: a bare AdaptiveDualWrite{} never panics and never silently drops writes — but it is not functionally adaptive: strikeThreshold is 0 (so recordStrike never degrades a cluster) and fireForgetSem is nil (so a cluster degraded some other way, e.g. ForceDegrade, falls back to a synchronous write in fireAndForget instead of true fire-and-forget). Use NewAdaptiveDualWrite or NewAdaptiveDualWriteChecked for the fully configured, adaptive behavior described above.

func NewAdaptiveDualWrite 

func NewAdaptiveDualWrite(opts ...AdaptiveDualWriteOption) *AdaptiveDualWrite

NewAdaptiveDualWrite creates a new AdaptiveDualWrite strategy.

Defaults:

  • deltaThreshold: 300ms (tuned for Cassandra: normal p99 jitter is ~3-20ms, GC pauses can cause 50-100ms spikes, so 300ms indicates real degradation)
  • absoluteMax: 2s
  • minFloor: 100ms (ignores noise when both clusters are fast)
  • strikeThreshold: 3
  • recoveryThreshold: 5
  • fireForgetTimeout: 30s
  • fireForgetLimit: 100

Parameters:

  • opts: Optional configuration options

Returns:

  • *AdaptiveDualWrite: A new adaptive dual-write strategy

For production configuration that should fail fast on invalid option values, use NewAdaptiveDualWriteChecked.

func NewAdaptiveDualWriteChecked added in v1.4.0 

func NewAdaptiveDualWriteChecked(opts ...AdaptiveDualWriteOption) (*AdaptiveDualWrite, error)

NewAdaptiveDualWriteChecked creates a new AdaptiveDualWrite strategy and returns a validation error when any option value is invalid.

Parameters:

  • opts: Optional configuration options

Returns:

  • *AdaptiveDualWrite: A new adaptive dual-write strategy
  • error: Joined types.OptionError values when one or more options are invalid

func (*AdaptiveDualWrite) Execute 

func (a *AdaptiveDualWrite) Execute(
	ctx context.Context,
	writeA func(context.Context) error,
	writeB func(context.Context) error,
) (resultA, resultB error)

Execute performs adaptive concurrent writes to both clusters.

For healthy clusters, writes are executed concurrently and waited upon. For degraded clusters, writes are fire-and-forget (background goroutine). Fire-and-forget writes use a dedicated context.Background() with fireForgetTimeout, independent of the caller's ctx — the caller's context cancellation does not cancel in-flight background writes.

After execution, latencies are compared to update cluster health state.

Parameters:

  • ctx: Context for the operation
  • writeA: Function to write to cluster A
  • writeB: Function to write to cluster B

Returns:

  • resultA: Error from cluster A (nil if successful, ErrWriteAsync if fire-and-forget)
  • resultB: Error from cluster B (nil if successful, ErrWriteAsync if fire-and-forget)

func (*AdaptiveDualWrite) ExecuteStrict added in v1.4.0 

func (a *AdaptiveDualWrite) ExecuteStrict(
	ctx context.Context,
	writeA func(context.Context) error,
	writeB func(context.Context) error,
) (resultA, resultB error)

ExecuteStrict performs adaptive concurrent writes without fire-and-forget dispatch.

Unlike Execute, ExecuteStrict never spawns background goroutines. If a cluster is currently degraded, its write is skipped and types.ErrClusterDegraded is returned for that cluster — the caller receives the skip signal rather than the fire-and-forget types.ErrWriteAsync. Healthy clusters are written synchronously as in AdaptiveDualWrite.Execute.

Health state (strikes, recovery credit) is updated for clusters that actually run; degraded-and-skipped clusters are not penalised further. Recovery of degraded clusters in strict-only workloads is driven by the recovery probe configured via WithRecoveryProbe.

func (*AdaptiveDualWrite) ForceDegrade 

func (a *AdaptiveDualWrite) ForceDegrade(cluster types.ClusterID)

ForceDegrade manually marks a cluster as degraded.

This is useful for testing or manual intervention. The call acquires the cluster's mutex so that fastStrikes is reset atomically with the degraded transition — preventing a stale fast-strike accumulation from immediately recovering the cluster on the very next recordFast call.

Emits types.EventWriteDegraded with Reason "manual" and Count set to the cluster's current slow-strike count when this call performs the transition. A manual degrade does not clear slowStrikes, so Count reports whatever had accumulated before the call rather than always being zero. Calling it on an already-degraded cluster still clears fastStrikes but emits nothing.

Parameters:

  • cluster: The cluster to degrade

func (*AdaptiveDualWrite) ForceRecover 

func (a *AdaptiveDualWrite) ForceRecover(cluster types.ClusterID)

ForceRecover manually marks a cluster as healthy.

This is useful for testing or manual intervention when you know a cluster has recovered (e.g., from external health checks).

Emits types.EventWriteRecovered with Reason "manual" when this call performs the transition. Calling it on an already-healthy cluster still clears the strike counters and last latency but emits nothing.

Parameters:

  • cluster: The cluster to recover

func (*AdaptiveDualWrite) IsDegraded 

func (a *AdaptiveDualWrite) IsDegraded(cluster types.ClusterID) bool

IsDegraded returns whether a cluster is currently in degraded (fire-and-forget) mode.

Parameters:

  • cluster: The cluster to check

Returns:

  • bool: true if the cluster is degraded

func (*AdaptiveDualWrite) LoggerConfigured added in v1.3.0 

func (a *AdaptiveDualWrite) LoggerConfigured() bool

LoggerConfigured reports whether the logger was explicitly set via WithAdaptiveLogger. Mirrors MetricsConfigured so the helix client can use the same auto-injection guard for both knobs.

func (*AdaptiveDualWrite) MetricsConfigured added in v1.3.0 

func (a *AdaptiveDualWrite) MetricsConfigured() bool

MetricsConfigured reports whether the metrics collector was explicitly set via WithAdaptiveMetrics. The helix client uses this to detect a caller-passed strategy without metrics and inject its own collector so the strategy's fire-and-forget path participates in unified instrumentation.

func (*AdaptiveDualWrite) RecordFastWrite 

func (a *AdaptiveDualWrite) RecordFastWrite(cluster types.ClusterID)

RecordFastWrite manually records a fast write for a cluster.

This is useful for external health probes or testing recovery. Call this when you know a cluster responded quickly (e.g., from a separate health check mechanism).

Parameters:

  • cluster: The cluster to record the fast write for

func (*AdaptiveDualWrite) RecordProbeSuccess added in v1.4.0 

func (a *AdaptiveDualWrite) RecordProbeSuccess(cluster types.ClusterID)

RecordProbeSuccess credits one successful recovery probe against the cluster. After the existing consecutive-fast threshold is reached, the cluster transitions back to healthy. Safe to call when the cluster is not degraded (no-op in that case).

This feeds the same recovery counter as natural fast writes (via AdaptiveDualWrite.RecordFastWrite) so strict-only workloads still benefit from Helix's auto-healing principle even when no write-side recovery signal is generated.

func (*AdaptiveDualWrite) Reset 

func (a *AdaptiveDualWrite) Reset()

Reset clears all health state, returning both clusters to healthy.

This is useful for testing or manual intervention. Each cluster that was actually degraded transitions back to healthy, so types.EventWriteRecovered with Reason "manual reset" is emitted for it; a cluster that was already healthy produces no event. The two clusters are reset one after the other, each under its own mutex — the first is fully released before the second is taken, so Reset never holds both locks at once.

Reset delivers its events once, after both clusters have been reset. When no other transition is running, a handler that inspects the strategy from inside the callback therefore sees both clusters healthy rather than cluster A recovered while cluster B is still degraded.

That is not a guarantee against concurrent callers. Delivery is shared: whichever goroutine gets there first delivers everything queued, so a transition on either cluster that runs while Reset sits between the two clusters can deliver cluster A's recovery before cluster B has been reset. A handler racing a Reset may therefore observe a partially applied reset. Per-cluster order still holds in every case — for one cluster, events are delivered in the order its transitions happened.

func (*AdaptiveDualWrite) SetClusterNames added in v1.3.0 

func (a *AdaptiveDualWrite) SetClusterNames(names types.ClusterNames)

SetClusterNames implements types.ClusterNamer so the helix client can propagate cluster names configured via WithClusterNames into log messages.

Safe to call concurrently with Execute.

func (*AdaptiveDualWrite) SetEventEmitter added in v1.6.0 

func (a *AdaptiveDualWrite) SetEventEmitter(em types.ClusterEventEmitter)

SetEventEmitter sets the cluster event emitter used for degrade and recover notifications. helix.NewCQLClient injects this automatically when a WithOnClusterEvent handler is registered; standalone users may call it directly at any time, since the emitter reference is swapped atomically and recovery probes may already be running. Delivery is best-effort: a transition racing exactly with an emitter install or removal may not reach either the old or the new emitter.

Emission ordering: a degrade or recover transition is recorded, and its event appended to an internal queue, under that cluster's state mutex, so events for one cluster are delivered in the order their transitions occurred. The emitter itself is always invoked with no policy locks held — see types.ClusterEventEmitter for the full contract an emitter must satisfy.

Parameters:

  • em: The event emitter; nil disables emission

func (*AdaptiveDualWrite) SetLogger added in v1.3.0 

func (a *AdaptiveDualWrite) SetLogger(l types.Logger)

SetLogger replaces the logger. No-op once LoggerConfigured returns true (caller's explicit choice wins) or if l is nil. Used by helix.NewCQLClient to propagate the client logger into the strategy when the caller did not provide one.

func (*AdaptiveDualWrite) SetMetrics added in v1.3.0 

func (a *AdaptiveDualWrite) SetMetrics(m types.MetricsCollector)

SetMetrics replaces the metrics collector. No-op once MetricsConfigured returns true (caller's explicit choice wins) or if m is nil.

type AdaptiveDualWriteOption 

type AdaptiveDualWriteOption func(*AdaptiveDualWrite)

AdaptiveDualWriteOption configures an AdaptiveDualWrite strategy.

func WithAdaptiveAbsoluteMax 

func WithAdaptiveAbsoluteMax(d time.Duration) AdaptiveDualWriteOption

WithAdaptiveAbsoluteMax sets the absolute latency cap.

If a cluster's latency exceeds this threshold, it is immediately considered for degradation (regardless of the other cluster's latency).

Default: 2s

Parameters:

  • d: Maximum acceptable latency

Returns:

  • AdaptiveDualWriteOption: Configuration option

func WithAdaptiveClusterNames added in v1.3.0 

func WithAdaptiveClusterNames(names types.ClusterNames) AdaptiveDualWriteOption

WithAdaptiveClusterNames sets the display names for clusters in log messages emitted by the fire-and-forget background path.

Parameters:

  • names: The cluster names

Returns:

  • AdaptiveDualWriteOption: Configuration option

func WithAdaptiveDeltaThreshold 

func WithAdaptiveDeltaThreshold(d time.Duration) AdaptiveDualWriteOption

WithAdaptiveDeltaThreshold sets the relative latency difference threshold.

If one cluster is slower than the other by more than this amount (and both are above minFloor), the slower one accumulates strikes.

Default: 300ms (tuned for Cassandra latency characteristics)

Parameters:

  • d: Latency difference threshold

Returns:

  • AdaptiveDualWriteOption: Configuration option

func WithAdaptiveFireForgetLimit 

func WithAdaptiveFireForgetLimit(n int) AdaptiveDualWriteOption

WithAdaptiveFireForgetLimit sets the maximum concurrent fire-and-forget writes.

When a cluster is degraded and writes are sent via fire-and-forget, this limit prevents resource exhaustion from too many pending goroutines. If the limit is reached, new fire-and-forget writes are dropped (returning ErrWriteDropped) and the replay system handles reconciliation.

Default: 100

Parameters:

  • n: Maximum concurrent fire-and-forget writes (must be positive, max 2^31-1)

Returns:

  • AdaptiveDualWriteOption: Configuration option

func WithAdaptiveFireForgetTimeout 

func WithAdaptiveFireForgetTimeout(d time.Duration) AdaptiveDualWriteOption

WithAdaptiveFireForgetTimeout sets the timeout for fire-and-forget writes.

When a cluster is degraded, writes are sent in a background goroutine with this timeout. This prevents resource leaks from hanging connections.

Default: 30s

Parameters:

  • d: Timeout for background writes

Returns:

  • AdaptiveDualWriteOption: Configuration option

func WithAdaptiveLogger added in v1.3.0 

func WithAdaptiveLogger(l types.Logger) AdaptiveDualWriteOption

WithAdaptiveLogger sets the logger for fire-and-forget background writes.

Marks the configuration as "logger explicitly set" so a parent caller (e.g. helix.NewCQLClient) does not auto-inject a different logger later via AdaptiveDualWrite.SetLogger. This mirrors the WithAdaptiveMetrics / explicit-wins contract.

Parameters:

  • l: The logger

Returns:

  • AdaptiveDualWriteOption: Configuration option

func WithAdaptiveMetrics added in v1.3.0 

func WithAdaptiveMetrics(m types.MetricsCollector) AdaptiveDualWriteOption

WithAdaptiveMetrics sets the metrics collector for fire-and-forget background writes. Without this option (or auto-injection by [helix.NewCQLClient]), real errors against a degraded cluster's background write are not surfaced to metrics — only the foreground ErrWriteAsync result is.

Parameters:

  • m: The metrics collector

Returns:

  • AdaptiveDualWriteOption: Configuration option

func WithAdaptiveMinFloor 

func WithAdaptiveMinFloor(d time.Duration) AdaptiveDualWriteOption

WithAdaptiveMinFloor sets the minimum latency floor.

Relative delta comparisons are ignored if both clusters respond faster than this threshold. This filters out noise from minor variations when both clusters are performing well.

Default: 100ms (accommodates typical Cassandra GC pauses and minor jitter)

Parameters:

  • d: Minimum latency floor

Returns:

  • AdaptiveDualWriteOption: Configuration option

func WithAdaptiveRecoveryThreshold 

func WithAdaptiveRecoveryThreshold(n int) AdaptiveDualWriteOption

WithAdaptiveRecoveryThreshold sets the consecutive fast writes to recover.

A degraded cluster must be fast for this many consecutive writes before it is restored to healthy status.

Default: 5

Parameters:

  • n: Number of consecutive fast writes required (must be positive, max 2^31-1)

Returns:

  • AdaptiveDualWriteOption: Configuration option

func WithAdaptiveStrikeThreshold 

func WithAdaptiveStrikeThreshold(n int) AdaptiveDualWriteOption

WithAdaptiveStrikeThreshold sets the consecutive slow writes before degradation.

A cluster must be slow for this many consecutive writes before it is marked as degraded and switched to fire-and-forget mode.

Default: 3

Parameters:

  • n: Number of consecutive slow writes required (must be positive, max 2^31-1)

Returns:

  • AdaptiveDualWriteOption: Configuration option

type CircuitBreaker 

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

CircuitBreaker implements a conservative failover policy.

Tracks consecutive failures and only triggers failover after a threshold is reached. This prevents flapping on transient errors.

Concurrency model:

  • muA / muB serialize all compound operations on cluster A / B state (load lastFailure → decide reset-or-add → store failures → store timestamp). Without this serialization the three atomic ops form a TOCTOU sequence that can lose counts or emit duplicate metrics under concurrent callers.
  • failuresA / failuresB remain atomic.Int32 so ShouldFailover can read them lock-free on the hot path; a one-call lag on a transition is fine.
  • trippedA / trippedB track whether the circuit has already tripped to open; they are plain bools guarded by muA / muB respectively, ensuring the "circuit opened" metric fires exactly once per trip.
  • seqA / seqB count latched transitions per cluster and are incremented under muA / muB. A call captures the value its own transitions produced and, once the state mutex is released, writes the state gauge and the transition log line only while that value is still current. Without it a goroutine descheduled between latching a transition and reporting it could overwrite a fresher state with a stale one.
  • reportMuA / reportMuB make that check and those writes a single step. They are separate from muA / muB on purpose: a caller-supplied metrics collector or logger never runs while a state mutex is held, so a slow one cannot stall state mutation or routing decisions. They are always released before events are delivered. They are plain sync.Mutex values, not reentrant: a metrics collector or logger that synchronously calls back into RecordFailure or RecordSuccess for the same cluster, from inside the very call the report mutex is serializing, deadlocks trying to reacquire it.
  • clusterNames is stored in an atomic.Pointer so SetClusterNames is safe to call concurrently with RecordFailure / RecordSuccess.

Zero value: a bare CircuitBreaker{} never panics — every method is safe to call — but it is not functionally a circuit breaker: threshold is 0 and metrics/logger are nil interfaces until finalizeCircuitBreaker runs, which only happens inside NewCircuitBreaker / NewCircuitBreakerChecked. Use one of those constructors to get a fully configured, functional CircuitBreaker.

func NewCircuitBreaker 

func NewCircuitBreaker(opts ...CircuitBreakerOption) *CircuitBreaker

NewCircuitBreaker creates a new CircuitBreaker policy.

Defaults: threshold=3, resetTimeout=30s

Parameters:

  • opts: Optional configuration options

Returns:

  • *CircuitBreaker: A new circuit breaker policy

For production configuration that should fail fast on invalid option values, use NewCircuitBreakerChecked.

func NewCircuitBreakerChecked added in v1.4.0 

func NewCircuitBreakerChecked(opts ...CircuitBreakerOption) (*CircuitBreaker, error)

NewCircuitBreakerChecked creates a new CircuitBreaker policy and returns a validation error when any option value is invalid.

Parameters:

  • opts: Optional configuration options

Returns:

  • *CircuitBreaker: A new circuit breaker policy
  • error: Joined types.OptionError values when one or more options are invalid

func (*CircuitBreaker) Failures 

func (c *CircuitBreaker) Failures(cluster types.ClusterID) int

Failures returns the current failure count for a cluster.

Parameters:

  • cluster: The cluster to check

Returns:

  • int: Number of consecutive failures

func (*CircuitBreaker) LoggerConfigured added in v1.4.0 

func (c *CircuitBreaker) LoggerConfigured() bool

LoggerConfigured reports whether the logger was explicitly set via WithCircuitBreakerLogger / WithLatencyLogger. Mirrors MetricsConfigured so the helix client can use the same auto-injection guard for both knobs.

func (*CircuitBreaker) MetricsConfigured added in v1.4.0 

func (c *CircuitBreaker) MetricsConfigured() bool

MetricsConfigured reports whether the metrics collector was explicitly set via WithCircuitBreakerMetrics (or WithLatencyMetrics for the embedded LatencyCircuitBreaker case). Helix's CQLClient uses this to detect a caller-passed policy without metrics and inject its own collector so circuit-breaker trips participate in unified instrumentation.

func (*CircuitBreaker) RecordFailure 

func (c *CircuitBreaker) RecordFailure(cluster types.ClusterID)

RecordFailure increments the failure counter for a cluster.

If the reset timeout has passed since the last failure, the counter is reset to 1 instead of incrementing. The compound load-check-store sequence is serialized by a per-cluster mutex to prevent TOCTOU races under concurrent callers. The "circuit opened" metric is emitted at most once per trip (guarded by the tripped flag inside the mutex).

A breaker that was open when the reset timeout elapsed closes on that same call: it emits types.EventCircuitBreakerClosed with Reason "reset timeout elapsed" and returns the state gauge to closed. With threshold 1 the reset also brings the counter straight back to the threshold, so the call closes and re-opens, emitting Closed then Open and leaving the gauge open.

Under concurrent callers the state gauge and the log line always describe the newest transition on that cluster: a call whose transition is superseded before it reports skips both, rather than putting back a state the breaker has already left. Its event is still delivered, in order.

Parameters:

  • cluster: The cluster that failed

func (*CircuitBreaker) RecordSuccess 

func (c *CircuitBreaker) RecordSuccess(cluster types.ClusterID)

RecordSuccess resets the failure counter for a cluster.

Parameters:

  • cluster: The cluster that succeeded

func (*CircuitBreaker) SetClusterNames 

func (c *CircuitBreaker) SetClusterNames(names types.ClusterNames)

SetClusterNames sets custom display names for clusters in log messages.

This method is called by the client during initialization to propagate cluster names configured via WithClusterNames. It is safe to call concurrently with RecordFailure and RecordSuccess.

Parameters:

  • names: The cluster names to use in log messages

func (*CircuitBreaker) SetEventEmitter added in v1.6.0 

func (c *CircuitBreaker) SetEventEmitter(em types.ClusterEventEmitter)

SetEventEmitter sets the cluster event emitter used for circuit breaker open/close notifications. helix.NewCQLClient injects this automatically when a WithOnClusterEvent handler is registered; standalone users may call it directly at any time, since the emitter reference is swapped atomically. Delivery is best-effort: a transition racing exactly with an emitter install or removal may not reach either the old or the new emitter.

Emission ordering: an open or close transition is recorded, and its event appended to an internal queue, under that cluster's state mutex, so events for one cluster are delivered in the order their transitions occurred. The emitter itself is always invoked with no breaker locks held — see types.ClusterEventEmitter for the full (relaxed) contract an emitter must satisfy.

Parameters:

  • em: The event emitter; nil disables emission

func (*CircuitBreaker) SetLogger added in v1.4.0 

func (c *CircuitBreaker) SetLogger(l types.Logger)

SetLogger replaces the logger. No-op once CircuitBreaker.LoggerConfigured returns true (caller's explicit choice wins) or if l is nil.

l's methods must not synchronously call back into RecordFailure or RecordSuccess on this breaker for the cluster they were invoked for — doing so deadlocks on the breaker's internal per-cluster report lock.

func (*CircuitBreaker) SetMetrics added in v1.4.0 

func (c *CircuitBreaker) SetMetrics(m types.MetricsCollector)

SetMetrics replaces the metrics collector. No-op once CircuitBreaker.MetricsConfigured returns true (caller's explicit choice wins) or if m is nil.

m's methods must not synchronously call back into RecordFailure or RecordSuccess on this breaker for the cluster they were invoked for — doing so deadlocks on the breaker's internal per-cluster report lock.

func (*CircuitBreaker) ShouldFailover 

func (c *CircuitBreaker) ShouldFailover(cluster types.ClusterID, _ error) bool

ShouldFailover returns true if the failure threshold has been reached AND the reset timeout has not yet elapsed since the last failure.

Once the reset timeout passes, ShouldFailover returns false to allow a half-open probe attempt: the next operation will be routed to the failed cluster, and its outcome (RecordSuccess closes the breaker; RecordFailure resets the counter to 1 and accumulates again) determines what happens next. Without this transition, a tripped breaker stays open indefinitely for any caller that stops sending traffic to the failed cluster (e.g. StickyRead routing all reads to the survivor) — there is no path to closure because no probe ever fires.

Note: this is "leaky" half-open — concurrent callers may all see false during the probe window and all be routed to the failed cluster. This is intentional: the per-cluster mutex in RecordFailure / RecordSuccess serializes the outcome, and at most (threshold) operations can fail against a still-broken cluster before the breaker re-trips.

Parameters:

  • cluster: The cluster that failed
  • err: The error (unused)

Returns:

  • bool: true if failover should occur

type CircuitBreakerOption 

type CircuitBreakerOption func(*CircuitBreaker)

CircuitBreakerOption configures a CircuitBreaker policy.

func WithCircuitBreakerClusterNames 

func WithCircuitBreakerClusterNames(names types.ClusterNames) CircuitBreakerOption

WithCircuitBreakerClusterNames sets the cluster display names for log messages.

Parameters:

  • names: The cluster names

Returns:

  • CircuitBreakerOption: Configuration option

func WithCircuitBreakerLogger 

func WithCircuitBreakerLogger(l types.Logger) CircuitBreakerOption

WithCircuitBreakerLogger sets the logger for the circuit breaker.

Parameters:

  • l: The logger

Returns:

  • CircuitBreakerOption: Configuration option

func WithCircuitBreakerMetrics 

func WithCircuitBreakerMetrics(m types.MetricsCollector) CircuitBreakerOption

WithCircuitBreakerMetrics sets the metrics collector for the circuit breaker.

Parameters:

  • m: The metrics collector

Returns:

  • CircuitBreakerOption: Configuration option

func WithResetTimeout 

func WithResetTimeout(d time.Duration) CircuitBreakerOption

WithResetTimeout sets the duration after which failure count resets.

Parameters:

  • d: Reset timeout duration

Returns:

  • CircuitBreakerOption: Configuration option

func WithThreshold 

func WithThreshold(n int) CircuitBreakerOption

WithThreshold sets the number of consecutive failures before failover.

Parameters:

  • n: Number of failures required

Returns:

  • CircuitBreakerOption: Configuration option

type ConcurrentDualWrite 

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

ConcurrentDualWrite implements a concurrent dual-write strategy.

Writes are executed concurrently on both clusters. Success is defined as at least one cluster succeeding. Failed writes are enqueued for replay.

func NewConcurrentDualWrite 

func NewConcurrentDualWrite(opts ...ConcurrentDualWriteOption) *ConcurrentDualWrite

NewConcurrentDualWrite creates a new ConcurrentDualWrite strategy.

Parameters:

  • opts: Optional configuration options

Returns:

  • *ConcurrentDualWrite: A new concurrent dual-write strategy

func (*ConcurrentDualWrite) Execute 

func (c *ConcurrentDualWrite) Execute(
	ctx context.Context,
	writeA func(context.Context) error,
	writeB func(context.Context) error,
) (resultA, resultB error)

Execute performs concurrent writes to both clusters.

Spawns one goroutine to write to cluster B while cluster A's write runs inline on the calling goroutine; both still execute concurrently, but only one extra goroutine is spawned per call. Returns the errors from both clusters (nil if successful).

Parameters:

  • ctx: Context for the operation
  • writeA: Function to write to cluster A
  • writeB: Function to write to cluster B

Returns:

  • resultA: Error from cluster A (nil if successful)
  • resultB: Error from cluster B (nil if successful)

func (*ConcurrentDualWrite) ExecuteStrict added in v1.4.0 

func (c *ConcurrentDualWrite) ExecuteStrict(
	ctx context.Context,
	writeA func(context.Context) error,
	writeB func(context.Context) error,
) (errA, errB error)

ExecuteStrict performs concurrent writes to both clusters with strict semantics.

For ConcurrentDualWrite, ExecuteStrict is identical to Execute: writes are already synchronous and never fire-and-forget. The method exists to satisfy the [helix.StrictWriter] interface so strict statements can use this strategy.

type ConcurrentDualWriteOption 

type ConcurrentDualWriteOption func(*ConcurrentDualWrite)

ConcurrentDualWriteOption configures a ConcurrentDualWrite strategy.

type LatencyCircuitBreaker 

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

LatencyCircuitBreaker extends CircuitBreaker with latency awareness.

In addition to tracking consecutive errors, this policy also treats slow responses (latency > absoluteMax) as "soft failures". This helps detect degraded clusters that are technically responding but too slow to be useful.

This policy implements the LatencyRecorder interface, which means the Helix client automatically calls RecordLatency() after successful read operations. No manual integration is required.

Example: 

lcb := policy.NewLatencyCircuitBreaker(
    policy.WithLatencyAbsoluteMax(2 * time.Second),
    policy.WithLatencyThreshold(3),
)

client, _ := helix.NewCQLClient(ctx, sessionA, sessionB,
    helix.WithFailoverPolicy(lcb),
)
// Latency is recorded automatically on each read!

Zero value: a bare LatencyCircuitBreaker{} never panics — every method is safe to call — but it is not functionally a circuit breaker: the embedded *CircuitBreaker is nil until one of the constructors runs. CircuitBreaker is embedded by pointer (not by value) so that copying a LatencyCircuitBreaker value only copies the pointer, not CircuitBreaker's mutexes/atomics — copying it, even before first use, would otherwise trip `go vet` copylocks and break existing external composite literals/selectors that depend on the pointer field shape. Because the embed can be nil, every promoted-looking method below (ShouldFailover, RecordFailure, RecordSuccess, Failures, SetClusterNames, MetricsConfigured, SetMetrics, SetEventEmitter, LoggerConfigured, SetLogger) is an explicit wrapper with a nil guard rather than a compiler-promoted method — use one of the constructors (NewLatencyCircuitBreaker / NewLatencyCircuitBreakerChecked) to get a fully configured, functional LatencyCircuitBreaker.

func NewLatencyCircuitBreaker 

func NewLatencyCircuitBreaker(opts ...LatencyCircuitBreakerOption) *LatencyCircuitBreaker

NewLatencyCircuitBreaker creates a new LatencyCircuitBreaker policy.

Defaults:

  • absoluteMax: 2s
  • threshold: 3 (inherited from CircuitBreaker)
  • resetTimeout: 30s (inherited from CircuitBreaker)

Parameters:

  • opts: Optional configuration options

Returns:

  • *LatencyCircuitBreaker: A new latency-aware circuit breaker policy

For production configuration that should fail fast on invalid option values, use NewLatencyCircuitBreakerChecked.

func NewLatencyCircuitBreakerChecked added in v1.4.0 

func NewLatencyCircuitBreakerChecked(opts ...LatencyCircuitBreakerOption) (*LatencyCircuitBreaker, error)

NewLatencyCircuitBreakerChecked creates a new LatencyCircuitBreaker policy and returns a validation error when any option value is invalid.

Parameters:

  • opts: Optional configuration options

Returns:

  • *LatencyCircuitBreaker: A new latency-aware circuit breaker policy
  • error: Joined types.OptionError values when one or more options are invalid

func (*LatencyCircuitBreaker) AbsoluteMax 

func (l *LatencyCircuitBreaker) AbsoluteMax() time.Duration

AbsoluteMax returns the configured latency threshold.

Returns:

  • time.Duration: The absolute maximum latency threshold

func (*LatencyCircuitBreaker) Failures added in v1.5.3 

func (l *LatencyCircuitBreaker) Failures(cluster types.ClusterID) int

Failures returns the current failure count for a cluster. See CircuitBreaker.Failures. A zero-value LatencyCircuitBreaker (nil embedded *CircuitBreaker) safely returns 0.

Parameters:

  • cluster: The cluster to check

Returns:

  • int: Number of consecutive failures

func (*LatencyCircuitBreaker) LoggerConfigured added in v1.5.3 

func (l *LatencyCircuitBreaker) LoggerConfigured() bool

LoggerConfigured reports whether the logger was explicitly set via WithLatencyLogger. See CircuitBreaker.LoggerConfigured. A zero-value LatencyCircuitBreaker (nil embedded *CircuitBreaker) safely returns false.

Returns:

  • bool: true if WithLatencyLogger was called at construction time

func (*LatencyCircuitBreaker) MetricsConfigured added in v1.5.3 

func (l *LatencyCircuitBreaker) MetricsConfigured() bool

MetricsConfigured reports whether the metrics collector was explicitly set via WithLatencyMetrics. See CircuitBreaker.MetricsConfigured. A zero-value LatencyCircuitBreaker (nil embedded *CircuitBreaker) safely returns false.

Returns:

  • bool: true if WithLatencyMetrics was called at construction time

func (*LatencyCircuitBreaker) RecordFailure added in v1.5.3 

func (l *LatencyCircuitBreaker) RecordFailure(cluster types.ClusterID)

RecordFailure increments the failure counter for a cluster. See CircuitBreaker.RecordFailure. A zero-value LatencyCircuitBreaker (nil embedded *CircuitBreaker) safely no-ops.

Parameters:

  • cluster: The cluster that failed

func (*LatencyCircuitBreaker) RecordLatency 

func (l *LatencyCircuitBreaker) RecordLatency(cluster types.ClusterID, latency time.Duration)

RecordLatency checks if the operation latency exceeds the absolute maximum.

If latency > absoluteMax, this is treated as a "soft failure" and RecordFailure() is called internally. Otherwise, RecordSuccess() is called to reset the failure counter.

This method should be called after successful operations to provide latency data for health tracking.

Parameters:

  • cluster: The cluster that was accessed
  • latency: The operation latency

func (*LatencyCircuitBreaker) RecordSuccess added in v1.5.3 

func (l *LatencyCircuitBreaker) RecordSuccess(cluster types.ClusterID)

RecordSuccess resets the failure counter for a cluster. See CircuitBreaker.RecordSuccess. A zero-value LatencyCircuitBreaker (nil embedded *CircuitBreaker) safely no-ops.

Parameters:

  • cluster: The cluster that succeeded

func (*LatencyCircuitBreaker) SetClusterNames added in v1.5.3 

func (l *LatencyCircuitBreaker) SetClusterNames(names types.ClusterNames)

SetClusterNames sets custom display names for clusters in log messages. See CircuitBreaker.SetClusterNames. A zero-value LatencyCircuitBreaker (nil embedded *CircuitBreaker) safely no-ops.

Parameters:

  • names: The cluster names to use in log messages

func (*LatencyCircuitBreaker) SetEventEmitter added in v1.6.0 

func (l *LatencyCircuitBreaker) SetEventEmitter(em types.ClusterEventEmitter)

SetEventEmitter sets the cluster event emitter used for circuit breaker open/close notifications. See CircuitBreaker.SetEventEmitter for the full contract. A zero-value LatencyCircuitBreaker (nil embedded *CircuitBreaker) safely no-ops, mirroring SetMetrics.

Parameters:

  • em: The event emitter; nil disables emission

func (*LatencyCircuitBreaker) SetLogger added in v1.5.3 

func (l *LatencyCircuitBreaker) SetLogger(log types.Logger)

SetLogger replaces the logger. See CircuitBreaker.SetLogger. A zero-value LatencyCircuitBreaker (nil embedded *CircuitBreaker) safely no-ops.

Parameters:

  • log: The logger to use

func (*LatencyCircuitBreaker) SetMetrics added in v1.5.3 

func (l *LatencyCircuitBreaker) SetMetrics(m types.MetricsCollector)

SetMetrics replaces the metrics collector. See CircuitBreaker.SetMetrics. A zero-value LatencyCircuitBreaker (nil embedded *CircuitBreaker) safely no-ops.

Parameters:

  • m: The metrics collector to use

func (*LatencyCircuitBreaker) ShouldFailover added in v1.5.3 

func (l *LatencyCircuitBreaker) ShouldFailover(cluster types.ClusterID, err error) bool

ShouldFailover returns true if the failure threshold has been reached AND the reset timeout has not yet elapsed since the last failure. See CircuitBreaker.ShouldFailover for the full half-open transition semantics. A zero-value LatencyCircuitBreaker (nil embedded *CircuitBreaker) safely returns false.

Parameters:

  • cluster: The cluster that failed
  • err: The error (unused)

Returns:

  • bool: true if failover should occur

type LatencyCircuitBreakerOption 

type LatencyCircuitBreakerOption func(*LatencyCircuitBreaker)

LatencyCircuitBreakerOption configures a LatencyCircuitBreaker policy.

func WithLatencyAbsoluteMax 

func WithLatencyAbsoluteMax(d time.Duration) LatencyCircuitBreakerOption

WithLatencyAbsoluteMax sets the latency threshold for soft failures.

If a successful operation takes longer than this duration, it is treated as a failure for circuit breaker purposes.

Default: 2s

Parameters:

  • d: Maximum acceptable latency

Returns:

  • LatencyCircuitBreakerOption: Configuration option

func WithLatencyLogger added in v1.0.0 

func WithLatencyLogger(log types.Logger) LatencyCircuitBreakerOption

WithLatencyLogger sets the logger for the latency circuit breaker.

Without this option the circuit breaker uses a no-op logger, so circuit trip/close events are never logged. Provide a real logger in production.

Parameters:

  • log: The logger

Returns:

  • LatencyCircuitBreakerOption: Configuration option

func WithLatencyMetrics added in v1.0.0 

func WithLatencyMetrics(m types.MetricsCollector) LatencyCircuitBreakerOption

WithLatencyMetrics sets the metrics collector for the latency circuit breaker.

Without this option the circuit breaker uses a no-op collector, so circuit trip events are never recorded. Provide a real collector in production to observe open/close transitions.

Parameters:

  • m: The metrics collector

Returns:

  • LatencyCircuitBreakerOption: Configuration option

func WithLatencyResetTimeout 

func WithLatencyResetTimeout(d time.Duration) LatencyCircuitBreakerOption

WithLatencyResetTimeout sets the duration after which failure count resets.

Default: 30s

Parameters:

  • d: Reset timeout duration

Returns:

  • LatencyCircuitBreakerOption: Configuration option

func WithLatencyThreshold 

func WithLatencyThreshold(n int) LatencyCircuitBreakerOption

WithLatencyThreshold sets the number of consecutive failures before failover.

This applies to both hard failures (errors) and soft failures (slow responses).

Default: 3

Parameters:

  • n: Number of failures required

Returns:

  • LatencyCircuitBreakerOption: Configuration option

type PrimaryOnlyRead 

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

PrimaryOnlyRead implements a read strategy that always reads from Cluster A.

Cluster B is only used for writes and as a failover target. Once Cluster A fails, reads are redirected to Cluster B until one of:

Without a recovery timeout, failover is permanent until Reset is called or Cluster B also fails — if only ClusterA recovers while B stays healthy, the client stays on ClusterB indefinitely.

func NewPrimaryOnlyRead 

func NewPrimaryOnlyRead(opts ...PrimaryOnlyReadOption) *PrimaryOnlyRead

NewPrimaryOnlyRead creates a new PrimaryOnlyRead strategy.

Parameters:

  • opts: Optional configuration options

Returns:

  • *PrimaryOnlyRead: A new primary-only read strategy

func (*PrimaryOnlyRead) OnFailure 

func (p *PrimaryOnlyRead) OnFailure(cluster types.ClusterID, _ error) (types.ClusterID, bool)

OnFailure handles read failures.

If ClusterA fails and has not yet failed over, routes to ClusterB. If ClusterA fails during an auto-recovery probe, resets the recovery timer so another probe will only occur after another full recovery timeout. If ClusterB fails while in the failed-over state, returns ClusterA as a probe without resetting state. If the caller's retry on ClusterA succeeds, OnSuccess clears the failover flag. If ClusterA is also down, the state stays failed-over to B, avoiding request-level A/B flipping.

Parameters:

  • cluster: The cluster that failed
  • err: The error

Returns:

  • types.ClusterID: Alternative cluster to try, or empty if no failover
  • bool: true if failover should be attempted

func (*PrimaryOnlyRead) OnSuccess 

func (p *PrimaryOnlyRead) OnSuccess(cluster types.ClusterID)

OnSuccess is called when a read succeeds.

If ClusterA succeeds while in failed-over state, the strategy resets back to ClusterA — completing auto-recovery.

Parameters:

  • cluster: The cluster that succeeded

func (*PrimaryOnlyRead) Reset 

func (p *PrimaryOnlyRead) Reset()

Reset resets the failover state back to primary immediately.

func (*PrimaryOnlyRead) Select 

func (p *PrimaryOnlyRead) Select(_ context.Context) types.ClusterID

Select returns ClusterA unless it has failed over.

If a recovery timeout is configured and has elapsed since the last failover, ClusterA is returned as a probe — allowing the caller to re-evaluate whether ClusterA has recovered.

Parameters:

  • ctx: Context (unused)

Returns:

  • types.ClusterID: ClusterA or ClusterB if failed over

type PrimaryOnlyReadOption added in v1.1.0 

type PrimaryOnlyReadOption func(*PrimaryOnlyRead)

PrimaryOnlyReadOption configures a PrimaryOnlyRead strategy.

func WithPrimaryOnlyRecoveryTimeout added in v1.1.0 

func WithPrimaryOnlyRecoveryTimeout(d time.Duration) PrimaryOnlyReadOption

WithPrimaryOnlyRecoveryTimeout sets the duration after which a failed-over PrimaryOnlyRead will automatically attempt to return reads to ClusterA.

When the timeout elapses, the next Select call returns ClusterA as a probe. If that read succeeds (OnSuccess called), the strategy remains on ClusterA. If it fails again (OnFailure called), the failover timer resets.

A zero or negative value disables auto-recovery (default: disabled).

Parameters:

  • d: Recovery timeout duration

Returns:

  • PrimaryOnlyReadOption: Configuration option

type RoundRobinRead 

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

RoundRobinRead implements a read strategy that alternates between clusters.

This provides even load distribution but lower cache efficiency.

func NewRoundRobinRead 

func NewRoundRobinRead() *RoundRobinRead

NewRoundRobinRead creates a new RoundRobinRead strategy.

Returns:

  • *RoundRobinRead: A new round-robin read strategy

func (*RoundRobinRead) OnFailure 

func (r *RoundRobinRead) OnFailure(cluster types.ClusterID, _ error) (types.ClusterID, bool)

OnFailure handles read failures.

Parameters:

  • cluster: The cluster that failed
  • err: The error

Returns:

  • types.ClusterID: The other cluster
  • bool: true (always attempt failover)

func (*RoundRobinRead) OnSuccess 

func (r *RoundRobinRead) OnSuccess(_ types.ClusterID)

OnSuccess is called when a read succeeds.

Parameters:

  • cluster: The cluster that succeeded (unused)

func (*RoundRobinRead) Select 

func (r *RoundRobinRead) Select(_ context.Context) types.ClusterID

Select returns alternating clusters on each call.

Parameters:

  • ctx: Context (unused)

Returns:

  • types.ClusterID: Alternating between ClusterA and ClusterB

type StickyRead 

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

StickyRead implements a sticky read strategy that routes reads to a preferred cluster.

The preferred cluster is randomly selected at initialization and sticks to it to maximize cache hits. On failure, it can fail over to the secondary cluster. Cooldown only gates future state changes; it does not trigger passive probing back to a recovered cluster in the absence of another read failure.

StickyRead has no programmatic way to return the preferred cluster to its original choice once it has failed over (unlike PrimaryOnlyRead.Reset). To force preferred back to a specific cluster, the operator must:

  • wait for the current preferred to fail (failover will swap them again, subject to cooldown), or
  • reconstruct a new StickyRead with WithPreferredCluster and rebuild the CQLClient.

func NewStickyRead 

func NewStickyRead(opts ...StickyReadOption) *StickyRead

NewStickyRead creates a new StickyRead strategy.

By default, the preferred cluster is randomly selected (50/50 between A and B) and the failover cooldown is 5 minutes.

Parameters:

  • opts: Optional configuration options

Returns:

  • *StickyRead: A new sticky read strategy

func (*StickyRead) OnFailure 

func (s *StickyRead) OnFailure(cluster types.ClusterID, _ error) (types.ClusterID, bool)

OnFailure handles read failures and determines failover.

If the failed cluster is the preferred one, returns the alternative cluster for failover. When cooldown has passed, the preferred cluster is also switched. When cooldown is still active, the alternative is returned for the current request but preferred is not changed — this prevents reads from failing entirely while still avoiding rapid preferred-cluster oscillation. Cooldown expiry alone does not change the preferred cluster; a later failure on the current preferred cluster is still required to switch back.

Parameters:

  • cluster: The cluster that failed
  • err: The error (unused)

Returns:

  • types.ClusterID: Alternative cluster to try
  • bool: true if failover should be attempted

func (*StickyRead) OnSuccess 

func (s *StickyRead) OnSuccess(_ types.ClusterID)

OnSuccess is called when a read succeeds.

Parameters:

  • cluster: The cluster that succeeded (unused for sticky reads)

func (*StickyRead) Preferred 

func (s *StickyRead) Preferred() types.ClusterID

Preferred returns the current preferred cluster.

Returns:

  • types.ClusterID: The current preferred cluster

func (*StickyRead) Select 

func (s *StickyRead) Select(_ context.Context) types.ClusterID

Select returns the preferred cluster for reading.

Parameters:

  • ctx: Context (unused but required by interface)

Returns:

  • types.ClusterID: The preferred cluster

type StickyReadOption 

type StickyReadOption func(*StickyRead)

StickyReadOption configures a StickyRead strategy.

func WithPreferredCluster 

func WithPreferredCluster(cluster types.ClusterID) StickyReadOption

WithPreferredCluster sets the initial preferred cluster.

Parameters:

  • cluster: The cluster to prefer initially

Returns:

  • StickyReadOption: Configuration option

func WithStickyReadCooldown 

func WithStickyReadCooldown(d time.Duration) StickyReadOption

WithFailoverCooldown sets the cooldown period after a failover.

Parameters:

  • d: Duration to wait before allowing another failover

Returns:

  • StickyReadOption: Configuration option

type SyncDualWrite 

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

SyncDualWrite implements a sequential dual-write strategy.

Writes are executed sequentially: first to cluster A, then to cluster B. This is useful for debugging or when strict ordering is required.

func NewSyncDualWrite 

func NewSyncDualWrite(opts ...SyncDualWriteOption) *SyncDualWrite

NewSyncDualWrite creates a new SyncDualWrite strategy.

By default, writes go to cluster A first.

Parameters:

  • opts: Optional configuration options

Returns:

  • *SyncDualWrite: A new sync dual-write strategy

func (*SyncDualWrite) Execute 

func (s *SyncDualWrite) Execute(
	ctx context.Context,
	writeA func(context.Context) error,
	writeB func(context.Context) error,
) (resultA, resultB error)

Execute performs sequential writes to both clusters.

Writes to the first cluster, then to the second. If the context is already canceled or deadline-exceeded after the first write returns, the second write is skipped and ctx.Err() is returned for it — avoiding wasted work and an immediate predictable error that the caller would have to handle anyway.

Parameters:

  • ctx: Context for the operation
  • writeA: Function to write to cluster A
  • writeB: Function to write to cluster B

Returns:

  • resultA: Error from cluster A (nil if successful)
  • resultB: Error from cluster B (nil if successful)

func (*SyncDualWrite) ExecuteStrict added in v1.4.0 

func (s *SyncDualWrite) ExecuteStrict(
	ctx context.Context,
	writeA func(context.Context) error,
	writeB func(context.Context) error,
) (errA, errB error)

ExecuteStrict performs sequential writes to both clusters with strict semantics.

For SyncDualWrite, ExecuteStrict is identical to Execute: writes are already synchronous and never fire-and-forget. The method exists to satisfy the [helix.StrictWriter] interface so strict statements can use this strategy.

type SyncDualWriteOption 

type SyncDualWriteOption func(*SyncDualWrite)

SyncDualWriteOption configures a SyncDualWrite strategy.

func WithPrimaryFirst 

func WithPrimaryFirst() SyncDualWriteOption

WithPrimaryFirst configures writes to go to cluster A first.

Returns:

  • SyncDualWriteOption: Configuration option

func WithSecondaryFirst 

func WithSecondaryFirst() SyncDualWriteOption

WithSecondaryFirst configures writes to go to cluster B first.

Returns:

  • SyncDualWriteOption: Configuration option

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.