Documentation
¶
Overview ¶
Package source provides built-in partition source implementations.
Partition sources discover available partitions for assignment. The package includes:
- Static: Fixed list of partitions defined at compile time.
- NatsKV: Dynamic partition list backed by a NATS KeyValue bucket, supporting live updates via KV watches (implements PartitionSource, PartitionUpdater, and WatchablePartitionSource).
Custom sources can be implemented by satisfying the types.PartitionSource interface.
Index ¶
- Variables
- type NatsKV
- func (s *NatsKV) AddPartitions(ctx context.Context, partitions ...types.Partition) error
- func (s *NatsKV) List(_ context.Context) ([]types.Partition, error)
- func (s *NatsKV) Modify(ctx context.Context, fn func([]types.Partition) []types.Partition) error
- func (s *NatsKV) RemovePartitions(ctx context.Context, partitions ...types.Partition) error
- func (s *NatsKV) Snapshot(_ context.Context) ([]types.Partition, uint64, bool, error)
- func (s *NatsKV) Start(ctx context.Context) error
- func (s *NatsKV) Stop(_ context.Context) error
- func (s *NatsKV) Update(ctx context.Context, partitions []types.Partition) error
- func (s *NatsKV) Watch(ctx context.Context) <-chan struct{}
- type NatsKVOption
- type SourceUnavailableHook
- type Static
Constants ¶
This section is empty.
Variables ¶
var ErrUpdateRetryExhausted = errors.New("update retry budget exhausted")
ErrUpdateRetryExhausted is returned by Update and Modify when all CAS retry attempts are exhausted due to persistent concurrent conflicts.
Functions ¶
This section is empty.
Types ¶
type NatsKV ¶
type NatsKV struct {
// contains filtered or unexported fields
}
NatsKV implements a partition source backed by a NATS KeyValue bucket.
It watches a specific key in the KV bucket for updates to the partition list, supplemented by a periodic reconcile loop that recovers from missed watcher events. The combination of watch + reconcile guarantees eventual convergence with KV state.
Writes use CAS (compare-and-swap) to prevent silent lost updates from concurrent callers. Use Modify, AddPartitions, or RemovePartitions for read-modify-write operations; Update is a CAS-fenced authoritative replace.
The revision and known fields enable downstream audit logic to distinguish a never-written source (known=false) from a written-then-deleted source (known=true, empty partitions), which have different implications for coverage invariants.
func NewNatsKV ¶
func NewNatsKV(kv jetstream.KeyValue, key string, logger types.Logger, opts ...NatsKVOption) *NatsKV
NewNatsKV creates a new NATS KV-based partition source.
Existing callers that omit opts receive a 30-second reconcile interval by default. Pass WithReconcileInterval(0) to disable polling (useful in tests that want deterministic behaviour).
Parameters:
- kv: The NATS KeyValue bucket to use
- key: The key where partitions are stored (JSON or gzip-compressed JSON)
- logger: Optional structured logger (may be nil)
- opts: Functional options
Returns:
- *NatsKV: Configured source (call Start before use)
func (*NatsKV) AddPartitions ¶ added in v2.4.0
AddPartitions adds one or more partitions to the source, preserving concurrent additions. Duplicate partitions (by CanonicalID) are silently ignored; calling AddPartitions twice with the same partition is a no-op, not an error.
Internally implemented via Modify, so it is safe for concurrent callers.
Parameters:
- ctx: Context for the operation
- partitions: Partitions to add (validated before any KV round-trip)
Returns:
- error: Validation error, or ErrUpdateRetryExhausted if CAS budget exhausted
func (*NatsKV) List ¶
List returns the current list of partitions (deep copy).
Returns:
- []types.Partition: Current partition list
- error: Always nil
func (*NatsKV) Modify ¶ added in v2.4.0
Modify atomically transforms the partition list by applying fn, retrying on concurrent writes until the CAS succeeds or the retry budget is exhausted.
fn receives a fresh snapshot read directly from KV (not the local cache) on every attempt and must be deterministic and side-effect-free — it may be called multiple times. The function signature is:
fn(current []types.Partition) []types.Partition
On ErrKeyNotFound (key never written) fn receives an empty list. Returns ErrUpdateRetryExhausted if all attempts fail.
Example:
err := src.Modify(ctx, func(current []types.Partition) []types.Partition {
return append(current, types.Partition{Keys: []string{"new"}})
})
Parameters:
- ctx: Context for the operation
- fn: Transform function (may be called multiple times; must be side-effect-free)
Returns:
- error: Modify error or ErrUpdateRetryExhausted
func (*NatsKV) RemovePartitions ¶ added in v2.4.0
RemovePartitions removes one or more partitions from the source, matching by CanonicalID. Partitions not found are silently ignored. Concurrent mutations are preserved; internally implemented via Modify.
Parameters:
- ctx: Context for the operation
- partitions: Partitions to remove (validated before any KV round-trip)
Returns:
- error: Validation error, or ErrUpdateRetryExhausted if CAS budget exhausted
func (*NatsKV) Snapshot ¶ added in v2.4.0
Snapshot returns the current partition list together with the last observed KV revision and the known flag. This implements types.RevisionedPartitionSource.
The known flag distinguishes a never-written source (known=false, revision=0) from a written-then-deleted source (known=true, revision=deleteRevision, empty partitions).
Parameters:
- ctx: Context for cancellation (unused; snapshot is served from memory)
Returns:
- partitions: Deep copy of current partition list
- revision: Last observed KV revision
- known: True once any KV event has been observed
- error: Always nil
func (*NatsKV) Start ¶
Start initializes the source and starts watching for updates.
It fetches the initial partition list from KV, seeds revision and known state, and spawns background watcher and reconcile goroutines.
Returns:
- error: Initialization error (invalid initial data, watcher setup failure)
func (*NatsKV) Stop ¶
Stop stops the watcher and reconcile loop and waits for all source-owned goroutines to exit.
Returns:
- error: Cleanup error (nil on success)
func (*NatsKV) Update ¶
Update replaces the entire partition list in the KV bucket using CAS.
Update is an authoritative-replace primitive: it replaces whatever is currently in KV with exactly the provided list. It is NOT safe for concurrent read-modify-write patterns — use Modify for those. Two callers issuing Update with different lists will serialize through CAS retry; last-writer-wins semantics apply.
On success, the local cache is updated immediately so a subsequent List() returns the new value without waiting for the watcher round-trip.
Returns ErrUpdateRetryExhausted if all CAS retries are exhausted.
Note: NATS KV buckets have a default value size limit of 1MB (MaxMsgSize). Large partition lists are automatically compressed with Gzip. If the compressed size still exceeds the limit, the update will fail.
Parameters:
- ctx: Context for the operation
- partitions: New partition list (replaces existing content)
Returns:
- error: Update error or ErrUpdateRetryExhausted
func (*NatsKV) Watch ¶
Watch returns a channel that emits a signal when the partition list changes.
The channel is buffered (capacity 1). If the caller is not keeping up, signals are dropped rather than blocking the source. The channel is closed when ctx is cancelled.
Parameters:
- ctx: Context whose cancellation deregisters the watcher
Returns:
- <-chan struct{}: Signal channel
type NatsKVOption ¶ added in v2.4.0
type NatsKVOption func(*NatsKV)
NatsKVOption is a functional option for configuring a NatsKV source.
func WithLeadershipProbe ¶ added in v2.4.0
func WithLeadershipProbe(fn func() bool) NatsKVOption
WithLeadershipProbe sets a callback that the reconcile loop calls on each tick to choose between leader cadence (30s) and follower cadence (5min). When set, the fixed reconcile interval from WithReconcileInterval is ignored.
The provided function must be safe to call from any goroutine.
Parameters:
- fn: Returns true if this instance is currently the leader
Returns:
- NatsKVOption: Option function
func WithMetrics ¶ added in v2.5.0
func WithMetrics(m types.SourceMetrics) NatsKVOption
WithMetrics wires a types.SourceMetrics collector. The source uses the collector to expose its `parti_source_bucket_missing` gauge (set to 1 on the first unavailable-source observation — see SourceUnavailableHook for the empirical error surface — cleared on the next successful operation). The default is types.NopSourceMetrics.
Parameters:
- m: SourceMetrics collector
Returns:
- NatsKVOption: Option function
func WithReconcileInterval ¶ added in v2.4.0
func WithReconcileInterval(d time.Duration) NatsKVOption
WithReconcileInterval sets the periodic-reconcile ticker cadence. A value of 0 disables polling entirely. The default is 30s. When WithLeadershipProbe is also set the fixed interval is ignored in favour of the leadership-driven cadence (leader=30s / follower=5min).
Disabling the reconciler (d <= 0 with no leadership probe wired) disables the load-bearing recovery path for the source watcher: the nats.go KV watcher's Updates() channel does NOT close on a NATS server restart, and the periodic reconcile is the only mechanism that re-syncs source state after a silently stalled watcher. Disable polling only with full awareness — Start emits a one-shot WARN log line when this state is detected so the operational consequence is visible at first deploy.
Disabling the reconciler also disables bucket delete+recreate recovery: see reconcileOnce for the recovery boundary.
Parameters:
- d: Reconcile interval (0 disables; default 30s)
Returns:
- NatsKVOption: Option function
func WithUnavailableHook ¶ added in v2.5.0
func WithUnavailableHook(h SourceUnavailableHook) NatsKVOption
WithUnavailableHook registers a SourceUnavailableHook that fires when the partition-source bucket is observed to be unavailable. See SourceUnavailableHook for the deadlock contract and rate-limiting behavior. Without a hook, the loss is still logged and the metric is set (when WithMetrics is configured), but no escalation runs — the library cannot recreate a user-owned bucket on its own.
Parameters:
- h: Callback invoked on bucket-loss detection
Returns:
- NatsKVOption: Option function
func WithUpdateRetries ¶ added in v2.4.0
func WithUpdateRetries(n int) NatsKVOption
WithUpdateRetries sets the maximum number of CAS attempts for Update and Modify. A value <= 0 falls back to the default of 5. After exhausting all attempts, Update and Modify return ErrUpdateRetryExhausted.
Parameters:
- n: Maximum number of CAS attempts (default 5)
Returns:
- NatsKVOption: Option function
type SourceUnavailableHook ¶ added in v2.5.0
type SourceUnavailableHook func(err error)
SourceUnavailableHook fires when the partition-source bucket is unavailable on the live connection. The reconciler and watcher-restart paths each see a distinct error from the nats.go surface — jetstream.ErrBucketNotFound from a fresh jetstream.JetStream.KeyValue lookup, jetstream.ErrStreamNotFound from a cached watcher rebind, nats.ErrNoResponders from a cached jetstream.KeyValue.Get, or context.DeadlineExceeded when the source bucket cannot answer in time. The err argument hands the original sentinel to the hook so implementations can log or surface the raw cause, but **do not gate readiness on a single sentinel type**: when the hook fires, the source is unavailable.
The library does not recreate the bucket — it is caller-owned, and rebuilding it requires the operator's provisioning flow. Wire this hook into your readiness logic (e.g. flip a readiness flag and let the orchestrator rotate the pod) so the loss becomes an actionable signal rather than a silent stall.
Concurrency contract: the hook is invoked synchronously from the source's reconcile / restart goroutine. Implementations MUST be non-blocking and MUST NOT call back into the source (recursive Get / Update / Stop will deadlock against the watcher's restart path).
Rate-limiting: repeated observations within a cooldown window (default 30s, matching the default reconcile cadence) suppress the hook so the callback is not flooded. The companion [SourceMetrics] gauge stays set for the full duration of the outage and clears once a subsequent operation succeeds.
type Static ¶
type Static struct {
// contains filtered or unexported fields
}
Static implements a partition source with a fixed list of partitions.
func NewStatic ¶
NewStatic creates a new static partition source.
The source returns a fixed list of partitions that never changes. Useful for testing and scenarios where partitions are known at startup.
Parameters:
- partitions: Fixed list of partitions
Returns:
- *Static: Initialized static source
Example:
partitions := []types.Partition{
{Keys: []string{"tool001", "chamber1"}, Weight: 100},
{Keys: []string{"tool001", "chamber2"}, Weight: 150},
}
src := source.NewStatic(partitions)
js, _ := jetstream.New(conn)
mgr, err := parti.NewManager(&cfg, js, src, strategy.NewConsistentHash())
if err != nil { /* handle */ }
func (*Static) List ¶
List returns the static list of partitions.
Returns:
- []types.Partition: The fixed list of partitions
- error: Always nil (never fails)
func (*Static) Start ¶
Start implements PartitionSource.Start. For Static source, it validates the static partitions.
func (*Static) Update ¶
Update updates the partition list.
This allows the static source to simulate dynamic partition changes, which is useful for testing partition refresh scenarios.
Parameters:
- ctx: Context for the operation (unused)
- partitions: New list of partitions
Returns:
- error: Always nil
Example:
src := source.NewStatic(initialPartitions) // Later: add more partitions src.Update(context.Background(), expandedPartitions)
Source Files