node

Documentation

Overview

Package node provides core node types and operations for the global node pool.

Index

• func AverageEWMAForDomainsMs(entry *NodeEntry, domains []string) (float64, bool)
• type DomainLatencyStats
• type Hash
  • func HashFromRawOptions(raw []byte) Hash
  • func ParseHex(s string) (Hash, error)
  • func (h Hash) Hex() string
  • func (h Hash) IsZero() bool
  • func (h Hash) String() string
• type LatencyTable
  • func NewLatencyTable(maxEntries int) *LatencyTable
  • func (t *LatencyTable) Close()
  • func (t *LatencyTable) GetDomainStats(domain string) (DomainLatencyStats, bool)
  • func (t *LatencyTable) LoadEntry(domain string, stats DomainLatencyStats)
  • func (t *LatencyTable) LoadEntryClassified(domain string, stats DomainLatencyStats, isAuthority bool) (evictedDomain string, evicted bool)
  • func (t *LatencyTable) Range(fn func(domain string, stats DomainLatencyStats) bool)
  • func (t *LatencyTable) Size() int
  • func (t *LatencyTable) Update(domain string, latency, decayWindow time.Duration) (wasEmpty bool)
  • func (t *LatencyTable) UpdateClassified(domain string, latency, decayWindow time.Duration, isAuthority bool) (wasEmpty bool, evictedDomain string, evicted bool)
• type NodeEntry
  • func NewNodeEntry(hash Hash, rawOptions json.RawMessage, createdAt time.Time, ...) *NodeEntry
  • func (e *NodeEntry) AddSubscriptionID(subID string)
  • func (e *NodeEntry) GetEgressIP() netip.Addr
  • func (e *NodeEntry) GetEgressRegion() string
  • func (e *NodeEntry) GetLastError() string
  • func (e *NodeEntry) GetRegion(geoLookup func(netip.Addr) string) string
  • func (e *NodeEntry) HasEnabledSubscription(subLookup SubLookupFunc) bool
  • func (e *NodeEntry) HasLatency() bool
  • func (e *NodeEntry) HasOutbound() bool
  • func (e *NodeEntry) IsCircuitOpen() bool
  • func (e *NodeEntry) IsDisabledBySubscriptions(subLookup SubLookupFunc) bool
  • func (e *NodeEntry) IsHealthy() bool
  • func (e *NodeEntry) MatchRegexs(regexes []*regexp.Regexp, subLookup SubLookupFunc) bool
  • func (e *NodeEntry) RemoveSubscriptionID(subID string) (empty bool)
  • func (e *NodeEntry) SetEgressIP(ip netip.Addr)
  • func (e *NodeEntry) SetEgressRegion(region string)
  • func (e *NodeEntry) SetLastError(msg string)
  • func (e *NodeEntry) SubscriptionCount() int
  • func (e *NodeEntry) SubscriptionIDs() []string
• type SubLookupFunc

Functions

func AverageEWMAForDomainsMs

func AverageEWMAForDomainsMs(entry *NodeEntry, domains []string) (float64, bool)

AverageEWMAForDomainsMs returns the average EWMA latency in milliseconds across domains that exist in the node's latency table.

Types

type DomainLatencyStats

type DomainLatencyStats struct {
	Ewma        time.Duration
	LastUpdated time.Time
}

DomainLatencyStats holds the TD-EWMA latency statistics for a single domain.

type Hash

type Hash [16]byte

Hash is a 128-bit node identity derived from canonical JSON of the node's raw options (with the "tag" field removed). Two nodes with identical configuration (ignoring tag) produce the same Hash.

var Zero Hash

Zero is the zero-value Hash.

func HashFromRawOptions

func HashFromRawOptions(raw []byte) Hash

HashFromRawOptions computes a node Hash from raw JSON options. It unmarshals the JSON, removes the "tag" key, and re-marshals. Go's encoding/json sorts map keys at all nesting levels, so the output is deterministic without any manual sorting. If JSON parsing fails, it falls back to hashing the raw bytes directly.

func ParseHex

func ParseHex(s string) (Hash, error)

ParseHex decodes a 32-character hex string into a Hash.

func (Hash) Hex

func (h Hash) Hex() string

Hex returns the lowercase hex encoding of the hash.

func (Hash) IsZero

func (h Hash) IsZero() bool

IsZero reports whether h is the zero hash.

func (Hash) String

func (h Hash) String() string

String implements fmt.Stringer.

type LatencyTable

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

LatencyTable is a bounded, thread-safe per-domain latency table. It keeps authority domains in a resident partition and regular domains in a bounded partition evicted by least-recently-accessed timestamp. Domain lookup inside the table uses a 64-bit xxh3 hash key for compactness. This intentionally accepts extremely low-probability hash collisions.

func NewLatencyTable

func NewLatencyTable(maxEntries int) *LatencyTable

NewLatencyTable creates a new LatencyTable whose regular partition is bounded to maxEntries.

func (*LatencyTable) Close

func (t *LatencyTable) Close()

Close is kept for lifecycle symmetry. LatencyTable currently has no background resources and Close is a no-op.

func (*LatencyTable) GetDomainStats

func (t *LatencyTable) GetDomainStats(domain string) (DomainLatencyStats, bool)

GetDomainStats returns the latency stats for a domain, if present. Read touches are write-throttled: last-access timestamp is updated only when the last update is older than latencyReadTouchMinInterval.

func (*LatencyTable) LoadEntry

func (t *LatencyTable) LoadEntry(domain string, stats DomainLatencyStats)

LoadEntry stores a bootstrap-recovered entry directly (no TD-EWMA).

func (*LatencyTable) LoadEntryClassified

func (t *LatencyTable) LoadEntryClassified(
	domain string,
	stats DomainLatencyStats,
	isAuthority bool,
) (evictedDomain string, evicted bool)

LoadEntryClassified stores a bootstrap-recovered entry directly (no TD-EWMA) into either authority or regular partition.

func (*LatencyTable) Range

func (t *LatencyTable) Range(fn func(domain string, stats DomainLatencyStats) bool)

Range iterates all domain entries. Returning false stops iteration.

func (*LatencyTable) Size

func (t *LatencyTable) Size() int

Size returns the number of domains with latency data.

func (*LatencyTable) Update

func (t *LatencyTable) Update(domain string, latency, decayWindow time.Duration) (wasEmpty bool)

Update records a latency observation for the given domain using TD-EWMA. wasEmpty is true if the table had no entries before this update.

TD-EWMA formula:

weight = exp(-Δt / decayWindow)
newEwma = oldEwma * weight + latency * (1 - weight)

For the first observation of a domain, Ewma is set to the raw latency.

func (*LatencyTable) UpdateClassified

func (t *LatencyTable) UpdateClassified(
	domain string,
	latency, decayWindow time.Duration,
	isAuthority bool,
) (wasEmpty bool, evictedDomain string, evicted bool)

UpdateClassified records a latency observation and writes into either authority-resident partition or regular partition. It returns evicted=true with evictedDomain when a regular-domain eviction happens due to bounded capacity.

type NodeEntry

type NodeEntry struct {
	// --- Static (immutable after creation) ---
	Hash       Hash
	RawOptions json.RawMessage
	CreatedAt  time.Time

	LastError string

	// Atomic dynamic fields for concurrent hot-path reads.
	FailureCount     atomic.Int32
	CircuitOpenSince atomic.Int64 // unix-nano; 0 = not open

	LastEgressUpdate atomic.Int64 // unix-nano of last successful egress-IP sample
	// Probe-attempt timestamps (unix-nano). These are updated regardless of
	// probe success/failure, and are used by probe schedulers.
	LastLatencyProbeAttempt          atomic.Int64
	LastAuthorityLatencyProbeAttempt atomic.Int64
	LastEgressUpdateAttempt          atomic.Int64
	LatencyTable                     *LatencyTable // per-domain latency stats; nil if not initialized

	// Outbound instance for this node.
	Outbound atomic.Pointer[adapter.Outbound]
	// contains filtered or unexported fields
}

NodeEntry represents a node in the global pool. Static fields are set at creation; dynamic fields use atomics or mutex.

func NewNodeEntry

func NewNodeEntry(hash Hash, rawOptions json.RawMessage, createdAt time.Time, maxLatencyTableEntries int) *NodeEntry

NewNodeEntry creates a NodeEntry with the given static fields. maxLatencyTableEntries controls the bounded size of the regular-domain LRU partition in the per-domain latency table. Pass 0 to skip latency table initialization (e.g. in tests that don't need it).

func (*NodeEntry) AddSubscriptionID

func (e *NodeEntry) AddSubscriptionID(subID string)

AddSubscriptionID adds subID to the subscription set if not already present. Must be called under external synchronization (e.g. xsync.Compute).

func (*NodeEntry) GetEgressIP

func (e *NodeEntry) GetEgressIP() netip.Addr

GetEgressIP returns the node's egress IP, or the zero Addr if unknown.

func (*NodeEntry) GetEgressRegion

func (e *NodeEntry) GetEgressRegion() string

GetEgressRegion returns the node's stored region from probe metadata, or empty string if unknown.

func (*NodeEntry) GetLastError

func (e *NodeEntry) GetLastError() string

GetLastError returns the node's error string (thread-safe).

func (*NodeEntry) GetRegion

func (e *NodeEntry) GetRegion(geoLookup func(netip.Addr) string) string

GetRegion resolves a node region using explicit probe metadata first, then GeoIP fallback from egress IP.

func (*NodeEntry) HasEnabledSubscription

func (e *NodeEntry) HasEnabledSubscription(subLookup SubLookupFunc) bool

HasEnabledSubscription reports whether the node currently has at least one enabled subscription reference, based on subLookup.

subLookup must apply the caller's definition of "subscription still holds this node" (for example, excluding evicted managed-node entries).

func (*NodeEntry) HasLatency

func (e *NodeEntry) HasLatency() bool

HasLatency returns true if the node has at least one latency record.

func (*NodeEntry) HasOutbound

func (e *NodeEntry) HasOutbound() bool

HasOutbound returns true if the node has a valid outbound instance.

func (*NodeEntry) IsCircuitOpen

func (e *NodeEntry) IsCircuitOpen() bool

IsCircuitOpen returns true if the node is currently circuit-broken.

func (*NodeEntry) IsDisabledBySubscriptions

func (e *NodeEntry) IsDisabledBySubscriptions(subLookup SubLookupFunc) bool

IsDisabledBySubscriptions reports whether the node should be treated as disabled: all referencing subscriptions are disabled (or missing/inapplicable by subLookup semantics).

func (*NodeEntry) IsHealthy

func (e *NodeEntry) IsHealthy() bool

IsHealthy returns true when the node can be treated as healthy for routing/statistics: outbound is ready and circuit is not open.

func (*NodeEntry) MatchRegexs

func (e *NodeEntry) MatchRegexs(regexes []*regexp.Regexp, subLookup SubLookupFunc) bool

MatchRegexs tests whether the node matches ALL given regex filters. A match means any tag from any enabled subscription satisfies all regexes. Tags are tested in the format "<subscriptionName>/<tag>". For an empty regex list:

• if subLookup is nil, it matches everything (compatibility fallback);
• otherwise, it matches only when at least one enabled subscription exists.

func (*NodeEntry) RemoveSubscriptionID

func (e *NodeEntry) RemoveSubscriptionID(subID string) (empty bool)

RemoveSubscriptionID removes subID from the subscription set. Returns true if the set is now empty (node should be deleted). Must be called under external synchronization (e.g. xsync.Compute).

func (*NodeEntry) SetEgressIP

func (e *NodeEntry) SetEgressIP(ip netip.Addr)

SetEgressIP stores the node's egress IP.

func (*NodeEntry) SetEgressRegion

func (e *NodeEntry) SetEgressRegion(region string)

SetEgressRegion stores the node's explicit probe region. Empty input clears the stored value.

func (*NodeEntry) SetLastError

func (e *NodeEntry) SetLastError(msg string)

SetLastError sets the node's error string (thread-safe).

func (*NodeEntry) SubscriptionCount

func (e *NodeEntry) SubscriptionCount() int

SubscriptionCount returns the number of subscriptions referencing this node.

func (*NodeEntry) SubscriptionIDs

func (e *NodeEntry) SubscriptionIDs() []string

SubscriptionIDs returns a copy of the subscription ID slice (thread-safe).

type SubLookupFunc

type SubLookupFunc func(subID string, hash Hash) (name string, enabled bool, tags []string, ok bool)

SubLookupFunc resolves a subscription ID + node hash to the subscription's name, enabled status, and the tags for that node in that subscription. Returns ok=false if the subscription does not exist.