dht

package
v0.3.0 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const (
	// PunchPriorityHigh is used by the replication maintainer (过程二) when
	// a ReplicationSet member goes Bad — this directly affects record availability.
	PunchPriorityHigh = 2
	// PunchPriorityLow is used by the health probe (过程三) when persistent
	// UDP failures indicate the peer is behind a restrictive NAT.
	PunchPriorityLow = 1
	// PunchPriorityLowest is used by the query engine when it encounters an
	// unreachable node while iterating — speculative, best-effort.
	PunchPriorityLowest = 0
)

Punch priority levels for PunchTransport.Punch (§11 trigger table).

View Source 
const (
	// DefaultAlpha is exported for callers that inspect it; the slot engine uses
	// queryAlpha internally.
	DefaultAlpha = 5

	// DefaultStagger is retained for API compatibility but is not used by the
	// slot-based engine.
	DefaultStagger = 200 * time.Millisecond
)
View Source 
const DebugHTTPAddr = "127.0.0.1:2634"

DebugHTTPAddr is a suggested address for StartDebugHTTP when embedding the dht package directly (without a2ald). The daemon serves /debug/* on its own API port.

View Source 
const (
	DefaultMaxTotalKeys = 100_000
)

Variables

View Source 
var ErrNoEndpoint = errors.New("dht: no endpoint record")

ErrNoEndpoint is returned when iterative FIND_VALUE does not yield a valid endpoint record.

View Source 
var ErrNoMatchingRecords = errors.New("dht: no matching records")

ErrNoMatchingRecords is returned when FindRecords / AggregateRecords find nothing for the filter.

View Source 
var ErrStaleRecord = errors.New("dht: stale record")

ErrStaleRecord means an equal or older record already exists for the same slot.

View Source 
var ErrStorePolicy = errors.New("dht: policy rejected")

ErrStorePolicy signals a node-side policy rejection that is independent of record validity. Auth functions (RecordAuthFunc) should wrap this sentinel when the refusal is due to ACL, whitelist, or other node-level rules so that callers can distinguish a stable policy refusal from a record-level error. 

return fmt.Errorf("address blacklisted: %w", dht.ErrStorePolicy)

Functions

func ObservedAddr 

func ObservedAddr(from net.Addr) []byte

ObservedAddr encodes remote IP:port for PONG / FIND_*_RESP (spec §7.6).

Types

type BootstrapSeed 

type BootstrapSeed struct {
	Addr net.Addr
	Info protocol.NodeInfo
}

BootstrapSeed is a known dial address plus wire NodeInfo (legacy; prefer BootstrapAddrs).

type Config 

type Config struct {
	Transport transport.Transport
	Keystore  crypto.KeyStore
	// OnObservedAddr is called whenever a DHT response carries an observed_addr
	// (PONG, FIND_NODE_RESP, FIND_VALUE_RESP). reporter is the responding node's
	// NodeID; wire is the raw observed_addr bytes (6 or 18 bytes).
	// May be nil.
	OnObservedAddr func(reporter a2al.NodeID, wire []byte)
	// RecordAuth is an optional authority policy called by Store.Put after
	// signature/expiry verification passes (Phase 4: includes DHT key).
	// If nil, no authority check is performed (useful in tests).
	RecordAuth RecordAuthFunc
	// MaxStoreKeys limits the number of distinct DHT keys in the local store.
	// 0 uses DefaultMaxTotalKeys. Configurable per-node soft limit.
	MaxStoreKeys int
	// Logger is used for DHT-level diagnostics (send failures, RPC retries).
	// If nil, slog.Default() is used.
	Logger *slog.Logger
	// SeenPeersPath is the file path for persisting the seenPeers sliding-window
	// table across restarts (spec §7.3). Empty disables persistence (default in
	// tests). The file is written with mode 0600.
	SeenPeersPath string
	// PunchTransport is an optional ICE hole-punch integration.
	// When nil (default) the node operates in direct-UDP-only mode, which is
	// fully backward-compatible. Set by the host layer at startup via the same
	// injection pattern as OnObservedAddr.
	PunchTransport PunchTransport

	// LearnedPathFirst enables learned-path outbound selection (lastInbound,
	// skipCold, DeferICE). Default false preserves legacy direct-UDP behaviour.
	LearnedPathFirst bool

	// PushHandler is called when this node receives a MsgDHTPush (oneShot delivery).
	// Returns true if the record was new (daemon renews subscription via ACK).
	// Optional; nil disables push reception (node still participates as a pusher).
	PushHandler func(key a2al.NodeID, rec protocol.SignedRecord) bool
}

Config holds runtime dependencies for a DHT node (spec Step 7).

type DebugStats added in v0.1.4 

type DebugStats struct {
	RxPackets             uint64  `json:"rx_packets_verified"`
	TxPackets             uint64  `json:"tx_packets"`
	RPCOK                 uint64  `json:"rpc_completed"`
	TotalPeers            int     `json:"total_peers"`
	Reach1h               int     `json:"reach_1h"`
	Reach24h              int     `json:"reach_24h"`
	Reach7d               int     `json:"reach_7d"`
	EstimatedNetworkSize  int     `json:"estimated_network_size"`
	EstimateConfidence    float64 `json:"estimate_confidence,omitempty"`
	VerifiedPeers1h       int     `json:"verified_peers_1h,omitempty"`
	UniqueNodesSinceStart uint64  `json:"unique_nodes_since_start"`
}

DebugStats is the DHT portion of GET /debug/stats (spec §3.6, §7).

type FakePunchNetwork added in v0.1.8 

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

FakePunchNetwork is a shared peer registry for FakePunchTransport. Create once and share across all nodes that should be able to "punch" each other.

Each registered node is assigned a unique fake *net.UDPAddr so that OnPunchComplete can populate NodeInfo.IP/Port correctly and protocol marshalling succeeds. The actual in-process delivery still uses the MemTransport address via InjectReceived.

func NewFakePunchNetwork added in v0.1.8 

func NewFakePunchNetwork() *FakePunchNetwork

NewFakePunchNetwork creates an empty in-memory punch network.

func (*FakePunchNetwork) NewTransport added in v0.1.8 

func (fn *FakePunchNetwork) NewTransport(nodeID a2al.NodeID, addr net.Addr) *FakePunchTransport

NewTransport registers nodeID at addr and returns a FakePunchTransport for that node. Call Bind(node) on the returned transport after creating the Node so that Punch and SendTo can call OnPunchComplete / InjectReceived.

type FakePunchTransport added in v0.1.8 

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

FakePunchTransport implements dht.PunchTransport using in-memory delivery. Obtain via FakePunchNetwork.NewTransport; call Bind before use.

func (*FakePunchTransport) Bind added in v0.1.8 

func (t *FakePunchTransport) Bind(node *Node)

Bind wires the transport to its owning Node. Must be called after NewNode so that OnPunchComplete and InjectReceived are available. Also records the node's logical Address so Punch can supply it to OnPunchComplete (required to build a valid NodeInfo for routing).

func (*FakePunchTransport) HasConn added in v0.1.8 

func (t *FakePunchTransport) HasConn(nodeID a2al.NodeID) bool

HasConn implements dht.PunchTransport. FakePunchTransport does not maintain a pool; SendTo delivers directly via InjectReceived so every registered peer is always "connected". Returns true when the remote nodeID is registered in the network.

func (*FakePunchTransport) InvalidateConn added in v0.2.8 

func (t *FakePunchTransport) InvalidateConn(_ a2al.NodeID)

InvalidateConn implements dht.PunchTransport. FakePunchTransport has no real connection pool; this is a no-op.

func (*FakePunchTransport) Punch added in v0.1.8 

func (t *FakePunchTransport) Punch(nodeID a2al.NodeID, _ *protocol.EndpointRecord, _ int)

Punch implements dht.PunchTransport.

Looks up nodeID in the network. If the remote is registered, simulates immediate ICE success on both sides:

  • Calls self.OnPunchComplete(nodeID, remote.fakeAddr, true)
  • Calls remote.OnPunchComplete(selfNID, self.fakeAddr, true)

Both calls run in separate goroutines to match the async contract of a real ICE scheduler. The fakeAddr values are synthesised *net.UDPAddr entries with valid IP/Port, ensuring NodeInfo can be marshalled correctly in FIND_NODE responses.

If the remote is not registered, calls self.OnPunchComplete(nodeID, nil, false) to clear isPunching.

func (*FakePunchTransport) SendTo added in v0.1.8 

func (t *FakePunchTransport) SendTo(_ context.Context, nodeID a2al.NodeID, msg []byte) (bool, error)

SendTo implements dht.PunchTransport.

Looks up nodeID in the network. If found, delivers msg by calling InjectReceived on the remote node (from = self.selfAddr, simulating a QUIC stream receive on the remote side). Returns ok=true on success, ok=false when the remote is not registered (triggers UDP fallback in the caller).

type Node 

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

Node is a single DHT participant (routing + local store + wire handler).

func NewNode 

func NewNode(cfg Config) (*Node, error)

NewNode builds a node; keystore must list exactly one address (Phase 1).

func (*Node) AddContact 

func (n *Node) AddContact(addr net.Addr, ni protocol.NodeInfo)

AddContact pins a peer's dial address and seeds the routing table. Treated as a trusted (user-configured) contact: VerifiedAt is set to now.

func (*Node) Address 

func (n *Node) Address() a2al.Address

Address returns the agent address.

func (*Node) BindPeerAddr 

func (n *Node) BindPeerAddr(id a2al.NodeID, addr net.Addr)

BindPeerAddr registers a Verified live dial address. Deprecated: prefer NotePeerDialSuccess which additionally updates health counters. Retained for test helpers and internal callers that do not have RTT information.

func (*Node) BindPeerAnchor added in v0.2.6 

func (n *Node) BindPeerAnchor(id a2al.NodeID, addr net.Addr)

BindPeerAnchor registers a long-lived advertised/infra dial address. Anchor is never overwritten by hearsay or ephemeral sources.

func (*Node) Bootstrap 

func (n *Node) Bootstrap(ctx context.Context, seeds []BootstrapSeed) error

Bootstrap registers seeds with pre-known identity and runs FIND_NODE(self). For seeds where only ip:port is known, use BootstrapAddrs instead.

func (*Node) BootstrapAddrs 

func (n *Node) BootstrapAddrs(ctx context.Context, addrs []net.Addr) error

BootstrapAddrs connects to seed nodes by raw network addresses (ip:port only). All pings run concurrently; FindNode(self) is launched as soon as the first peer responds, without waiting for slower or failing seeds. This ensures the routing table starts widening at the earliest possible moment.

func (*Node) BootstrapCandidateAddrs 

func (n *Node) BootstrapCandidateAddrs(max int) []net.Addr

BootstrapCandidateAddrs returns up to max UDP addresses for cold-start bootstrap (routing table + remembered peer addrs). Best-effort for persisting peers.cache.

Candidates are sorted by observed health: Good → Unknown → Bad. The max cap therefore naturally favours peers we have successfully communicated with before, so that the next cold-start spends its bootstrap window on the most promising contacts rather than on known-dead nodes.

func (*Node) BuildFindNodeRequest added in v0.2.7 

func (n *Node) BuildFindNodeRequest(target a2al.NodeID) ([]byte, error)

BuildFindNodeRequest creates a signed FIND_NODE request targeting the given NodeID, suitable for sending via the signaling WebSocket DHT proxy.

func (*Node) BuildFindValueRequest added in v0.2.7 

func (n *Node) BuildFindValueRequest(target a2al.NodeID, recType uint8) ([]byte, error)

BuildFindValueRequest creates a signed FIND_VALUE request for the given key and record type, suitable for sending via the signaling WebSocket DHT proxy. recType 0 requests all record types.

func (*Node) ClearReachabilityHints added in v0.2.6 

func (n *Node) ClearReachabilityHints()

ClearReachabilityHints resets learned-path hints after network topology changes. PeerHealth backoff and failCount are preserved; lastInbound evidence is cleared so stale NAT-mapped addresses are not preferred on the new network.

skipColdUDP is intentionally NOT cleared here: it represents "UDP anchor has been confirmed persistently unreachable for this peer", which is a property of the remote peer rather than a local path observation. Changing networks does not invalidate that conclusion — the anchor is still likely unreachable. Its only correct reset path is recordSuccess, which fires as soon as any RPC to the peer succeeds (proving the anchor or ICE path is now working).

func (*Node) Close 

func (n *Node) Close() error

Close stops the node, closes the transport, and waits for the receive loop to exit.

func (*Node) DebugHTTPHandler 

func (n *Node) DebugHTTPHandler() http.Handler

DebugHTTPHandler returns read-only /debug/* handlers for mounting on an existing server (spec §3.6).

func (*Node) DebugStatsData added in v0.1.4 

func (n *Node) DebugStatsData() DebugStats

DebugStatsData returns a snapshot for embedding in host-level /debug/stats.

func (*Node) EstimatedNetworkSize added in v0.1.7 

func (n *Node) EstimatedNetworkSize() int

EstimatedNetworkSize returns the bucket-density estimate of the current number of active nodes in the DHT (includes all nodes; for freshness-filtered estimate use EstimatedNetworkSizeFiltered).

func (*Node) EstimatedNetworkSizeFiltered added in v0.1.7 

func (n *Node) EstimatedNetworkSizeFiltered(cutoff time.Time) (int, float64)

EstimatedNetworkSizeFiltered returns the network size estimate restricted to nodes verified within the past 30 minutes, along with a confidence score in [0, 1]. Higher confidence means more sample buckets contributed to the median.

func (*Node) FamilyStableAddr added in v0.3.0 

func (n *Node) FamilyStableAddr(id a2al.NodeID, v6 bool) (net.Addr, bool)

FamilyStableAddr returns the known stable dial address for id restricted to the requested address family, with no cross-family fallback. Returns false when the requested family has no known address or is currently in back-off.

This is the correct primitive for read-only consumers (NAT probe, passive observe) that must not silently substitute the other family. For outbound send decisions that may fall back across families, use outboundPlan instead.

func (*Node) FindNode 

func (n *Node) FindNode(ctx context.Context, peer net.Addr, target a2al.NodeID) ([]protocol.NodeInfo, error)

FindNode asks peer for closest nodes to target NodeID.

func (*Node) FindValue 

func (n *Node) FindValue(ctx context.Context, peer net.Addr, key a2al.NodeID) (*protocol.SignedRecord, error)

FindValue queries peer for the best endpoint record at key NodeID (legacy helper).

func (*Node) FindValueServedCount added in v0.2.2 

func (n *Node) FindValueServedCount() uint64

FindValueServedCount returns the cumulative number of FIND_VALUE RPCs answered with at least one record by this node.

func (*Node) FindValueWithNodes 

func (n *Node) FindValueWithNodes(ctx context.Context, peer net.Addr, key a2al.NodeID, recType uint8, subscribe bool) ([]protocol.SignedRecord, []protocol.NodeInfo, error)

FindValueWithNodes queries peer. recType 0 requests all record types in the response. subscribe=true sets OneShotSubscribe in the request, asking the peer to push future records. Only pass subscribe=true in AggregateRecords mode (not point-in-time FindRecords lookups).

func (*Node) HandleReadRequest added in v0.2.7 

func (n *Node) HandleReadRequest(raw []byte) []byte

HandleReadRequest processes a raw signed DHT request received over a non-UDP channel (e.g. the signaling WebSocket) and returns the signed response bytes. Only read-only operations (PING, FIND_NODE, FIND_VALUE) are served; STORE and any other write operations are rejected by returning nil.

func (*Node) InjectReceived added in v0.1.8 

func (n *Node) InjectReceived(data []byte, from net.Addr)

InjectReceived processes a pre-received DHT message from an external transport (e.g. a punched QUIC stream managed by the host layer).

It follows the same decode→dispatch path as recvLoop without reading from the UDP transport. from should be the peer's reachable net.Addr so that outbound responses can be addressed correctly; the host layer supplies the ICE-negotiated address or the peer's reflexive candidate.

Safe to call from any goroutine concurrently with the normal UDP receive loop.

func (*Node) IsPunching added in v0.1.8 

func (n *Node) IsPunching(id a2al.NodeID) bool

IsPunching reports whether an ICE hole-punch attempt is currently in flight for the given peer. Used by the query engine's addCand to exclude punching nodes from all track slots (§12 of strategy doc).

func (*Node) LocalAddr 

func (n *Node) LocalAddr() net.Addr

LocalAddr returns the underlying transport address.

func (*Node) LocalStoreGet added in v0.1.6 

func (n *Node) LocalStoreGet(key a2al.NodeID, recType uint8) []protocol.SignedRecord

LocalStoreGet returns verified non-expired records at the given DHT key, with optional RecType filter (0 = all types).

func (*Node) LocalStoreGetByAddress added in v0.1.6 

func (n *Node) LocalStoreGetByAddress(addr a2al.Address, recType uint8) []protocol.SignedRecord

LocalStoreGetByAddress returns verified non-expired records where sr.Address matches addr, with optional RecType filter (0 = all types). Scans all store buckets; intended for low-frequency paths such as the QUIC control exchange.

func (*Node) LocalStoreInvalidate added in v0.1.6 

func (n *Node) LocalStoreInvalidate(key a2al.NodeID, recType uint8)

LocalStoreInvalidate removes locally-cached records for key and recType (0 = all types). Used internally by the host layer to clear stale endpoint records when a connection attempt fails, so the next Resolve fetches fresh data from the network.

func (*Node) LocalStorePut added in v0.1.6 

func (n *Node) LocalStorePut(storeKey a2al.NodeID, rec protocol.SignedRecord) error

LocalStorePut writes rec into the local store without triggering replication. Use this to seed records received via an out-of-band channel (e.g. QUIC control plane AgentInfo push) so that subsequent AggregateRecords queries return the fresh data immediately.

If rec is an endpoint record, any negative-cache entry for the same nodeID is cleared so that the replication gap-fill NAT track can use the new signal URL immediately rather than waiting for the suppression window to expire.

func (*Node) NodeID 

func (n *Node) NodeID() a2al.NodeID

NodeID returns the DHT key for this node.

func (*Node) NotePeerDialFailure added in v0.2.7 

func (n *Node) NotePeerDialFailure(id a2al.NodeID, addr net.Addr)

NotePeerDialFailure records a failed direct connection attempt to addr. It is a hint to the health subsystem that this address is not currently reachable via the upper-layer transport.

Callers MUST only invoke this for genuine transport failures, not for context cancellations caused by a competing dial winning (happy eyeballs).

func (*Node) NotePeerDialSuccess added in v0.2.7 

func (n *Node) NotePeerDialSuccess(id a2al.NodeID, addr net.Addr, rtt time.Duration)

NotePeerDialSuccess records a verified direct connection (e.g. QUIC handshake) to addr from the upper-layer transport. It updates both the live address slot and the health counters — equivalent in effect to a successful DHT RPC, allowing a peer that was in Bad state to recover without waiting for a UDP round-trip.

addr should be the actual remote address observed on the connection (e.g. conn.RemoteAddr()), which may differ from the dialled address due to NAT. rtt is the round-trip time of the handshake; pass 0 if unknown.

func (*Node) OnPunchComplete added in v0.1.8 

func (n *Node) OnPunchComplete(nodeID a2al.NodeID, peerLogicalAddr a2al.Address, peerNetAddr net.Addr, success bool, isDirect bool, failReason PunchFailReason)

OnPunchComplete is the callback invoked by the PunchTransport implementation when an ICE attempt for nodeID finishes (success or failure).

peerLogicalAddr is the peer's a2al.Address (21-byte identity address derived from their public key). Used to populate NodeInfo.Address so that the entry can be included in FIND_NODE responses.

peerNetAddr is the peer's reachable net.Addr as determined by ICE (typically the winning candidate pair's remote UDP address). Must be non-nil on success.

success=true: a punched QUIC connection is now available via SendTo.

isDirect=true (Phase 8 误分类纠正): the ICE-selected path is host or server-reflexive, meaning the remote is directly reachable via plain UDP. In this case the node is admitted to the standard routing bucket (tabAdd) rather than the punched zone; the Mode B QUIC connection remains in the pool until it idle-times-out (5 min per modeBQUICConfig) — no special teardown.

success=false: failReason identifies the cause. PunchFailNoAgent indicates the remote is definitively offline; the health system marks the peer Bad permanently until the next probe cycle clears it, skipping further ICE.

Safe to call from any goroutine; uses healthMu/tabMu/peerMu.

func (*Node) PeerAllowContact added in v0.1.6 

func (n *Node) PeerAllowContact(id a2al.NodeID) bool

PeerAllowContact returns true if at least one address family's back-off for this peer has expired (or was never set). Contact is permitted when any family is available — callers use whichever family succeeds.

func (*Node) PeerHealthOf added in v0.1.3 

func (n *Node) PeerHealthOf(id a2al.NodeID) PeerHealthState

PeerHealthOf returns the aggregate health state across both address families.

Aggregation rules (designed to preserve exact backward-compatibility for single-family v4-only nodes during the migration period):

  • Any active family Good → PeerHealthGood
  • All active families Bad → PeerHealthBad
  • No active family at all → PeerHealthUnknown

Exception: if any family has a fresh lastInbound (within lastInboundFreshTTL), Bad is suppressed to Unknown. A peer that recently sent us an inbound message is demonstrably alive; the reply path (back to their source address) is available via the active NAT mapping, so treating them as fully unreachable would discard a working path and waste the inbound evidence.

func (*Node) PeerRTT added in v0.1.7 

func (n *Node) PeerRTT(addr net.Addr) time.Duration

PeerRTT returns the last measured round-trip time for addr, or 0 if the address is not yet known or has never completed a successful exchange.

func (*Node) Ping 

func (n *Node) Ping(ctx context.Context, peer net.Addr) error

Ping sends PING and waits for PONG. Start() must be running.

func (*Node) PingIdentity 

func (n *Node) PingIdentity(ctx context.Context, peer net.Addr) (*PeerIdentity, error)

PingIdentity sends PING, waits for PONG, and returns the remote peer's identity extracted from the response. The peer is automatically registered into the routing table and dial address map.

func (*Node) PublicHubCandidates added in v0.2.6 

func (n *Node) PublicHubCandidates(max int) []net.Addr

PublicHubCandidates returns up to max routing-table peers whose endpoint record indicates high inbound TCP reachability: NATFullCone, or any endpoint contains a publicly-routable IPv6 GUA address. Results are selected with bucket-index diversity (at most one per bucket) to spread candidates across the key space.

func (*Node) PublishEndpointRecord 

func (n *Node) PublishEndpointRecord(ctx context.Context, rec protocol.SignedRecord) error

PublishEndpointRecord stores the record locally and immediately returns (过程一). Replication to remote peers is handled asynchronously by scheduleReplicate → renewBackground (FindNode + staggered StoreAt).

func (*Node) PublishMailboxRecord 

func (n *Node) PublishMailboxRecord(ctx context.Context, storeKey a2al.NodeID, rec protocol.SignedRecord) error

PublishMailboxRecord stores the mailbox record at storeKey (recipient NodeID) locally and replicates to k-closest reachable peers asynchronously.

func (*Node) PublishTopicRecord 

func (n *Node) PublishTopicRecord(ctx context.Context, storeKey a2al.NodeID, rec protocol.SignedRecord) error

PublishTopicRecord stores the topic record at storeKey (SHA-256("topic:"+topic)) locally and replicates asynchronously.

func (*Node) RecoveryNotify added in v0.2.7 

func (n *Node) RecoveryNotify() <-chan struct{}

RecoveryNotify returns the channel that receives a token whenever the node recovers from a suspected-offline period (first successful RPC after ≥ offlineSuspectThreshold without any success). The channel has capacity 1; the caller should consume tokens promptly to avoid missing events.

func (*Node) RelaxHealthThrottle added in v0.2.7 

func (n *Node) RelaxHealthThrottle(dur time.Duration)

RelaxHealthThrottle caps the nextRetryAt of all known peers to at most now+cap. Peers whose backoff expires sooner, or who have no backoff, are unaffected. failCount is intentionally preserved so the exponential-backoff schedule is not fully discarded, only shortened.

Call this after a confirmed network recovery to let probe and replication loops reach throttled peers within cap, rather than waiting out potentially long backlogs accumulated during the outage.

func (*Node) RemoveRepSetsForPublisher added in v0.1.3 

func (n *Node) RemoveRepSetsForPublisher(publisher a2al.NodeID)

RemoveRepSetsForPublisher removes all repSets whose publisher matches the given NodeID. Call when an agent is deleted so background probes and refill tasks cease.

func (*Node) RepSetCounts added in v0.2.9 

func (n *Node) RepSetCounts(storeKey, publisher a2al.NodeID) (active, total int)

RepSetCounts returns (active, total) replica counts for (storeKey, publisher).

  • active: confirmed in the most recent renewBackground cycle — these are the replicas we have 100% confidence in right now.
  • total: all entries currently tracked (includes entries whose renewal has not yet succeeded in the current cycle).

Before the first renewal cycle completes, active == 0.

func (*Node) RepSetSize added in v0.1.8 

func (n *Node) RepSetSize(storeKey, publisher a2al.NodeID) int

RepSetSize returns the total number of tracked remote replicas for a record. Deprecated: prefer RepSetCounts which also returns the active (confirmed this cycle) count. RepSetSize is kept for callers that only need the total.

func (*Node) RoutingPeerIDs added in v0.3.0 

func (n *Node) RoutingPeerIDs() []a2al.NodeID

RoutingPeerIDs returns the NodeIDs of all peers currently in the routing table. This is a topology enumeration — it carries no address or dial decision. Callers that need a dial address for a specific family should use FamilyStableAddr.

func (*Node) SeedRecord added in v0.2.7 

func (n *Node) SeedRecord(rec protocol.SignedRecord) error

SeedRecord writes a pre-validated SignedRecord directly into the local store without triggering replication. For bootstrap seeding only: used when a record is obtained out-of-band (e.g. via the signaling WebSocket DHT proxy) and needs to be available locally before the DHT layer is operational.

func (*Node) SelfExtIP added in v0.1.7 

func (n *Node) SelfExtIP() net.IP

SelfExtIP returns the node's current public IP as seen by STUN/HTTP probe, or nil if not yet known.

func (*Node) SelfExtIPv6 added in v0.2.6 

func (n *Node) SelfExtIPv6() net.IP

SelfExtIPv6 returns the node's public IPv6 GUA set via SetSelfExtIPv6.

func (*Node) SendNATProbeReq added in v0.1.4 

func (n *Node) SendNATProbeReq(ctx context.Context, probeAddr net.Addr, claimedAddr []byte) (bool, error)

SendNATProbeReq asks probeAddr to send a NATProbeEcho to claimedAddr. claimedAddr is the wire-encoded public UDP address (6 or 18 bytes). Returns true if an echo arrived within the context deadline, nil error on timeout.

func (*Node) SetLearnedPathFirst added in v0.2.6 

func (n *Node) SetLearnedPathFirst(on bool)

SetLearnedPathFirst toggles learned-path outbound selection at runtime (tests / rollout).

func (*Node) SetMaxStoreKeys added in v0.1.7 

func (n *Node) SetMaxStoreKeys(max int)

SetMaxStoreKeys updates the maximum number of distinct keys in the local store.

func (*Node) SetOfflineSuspect added in v0.2.8 

func (n *Node) SetOfflineSuspect(v bool)

SetOfflineSuspect lets the daemon signal that local network connectivity is known to be unavailable (e.g. sleep/wake detected, no default route). recordSuccess clears the flag automatically when the first RPC succeeds. When v is true, any accumulated pending failures are discarded immediately so they cannot be flushed into failCount on the next success.

func (*Node) SetPassiveRouting added in v0.1.7 

func (n *Node) SetPassiveRouting(passive bool)

SetPassiveRouting controls whether this node suppresses proactive FindNode queries. When true (passive mode), the node fills its routing table naturally through incoming traffic and skips active bucket-refill and topology scans.

func (*Node) SetPushHandler added in v0.2.4 

func (n *Node) SetPushHandler(fn func(key a2al.NodeID, rec protocol.SignedRecord) bool)

SetPushHandler registers fn as the handler for incoming MsgDHTPush messages. fn(key, record) should return true if the record was new (causing ACK to renew the subscription). Thread-safe; can be called at any time.

func (*Node) SetSelfExtIP 

func (n *Node) SetSelfExtIP(ip net.IP)

SetSelfExtIP records our own public IPv4 (from STUN/HTTP probe). Used to detect NAT hairpin peers: nodes behind the same NAT share the same public IP and typically cannot reach each other via that IP (router hairpinning not supported).

func (*Node) SetSelfExtIPv6 added in v0.2.6 

func (n *Node) SetSelfExtIPv6(ip net.IP)

SetSelfExtIPv6 records our own public IPv6 GUA (from STUN/HTTP probe). Unlike the v4 counterpart this is not used for hairpin detection (v6 GUA nodes are directly reachable); it is used to self-identify when the node's v6 address appears in the well-known DNS list.

func (*Node) Start 

func (n *Node) Start()

Start begins the inbound packet loop and background maintenance workers.

func (*Node) StartDebugHTTP 

func (n *Node) StartDebugHTTP(addr string) (stop func(), err error)

StartDebugHTTP listens on addr and serves read-only /debug/* JSON (spec §3.6). When using the a2ald daemon the /debug/* routes are served on the API port (default 127.0.0.1:2121) and this method is not needed. Use it only when embedding the dht package directly without the daemon. stop shuts the server down (idempotent).

func (*Node) StartWithBootstrap 

func (n *Node) StartWithBootstrap(ctx context.Context, addrs []net.Addr) error

StartWithBootstrap starts the receive loop then bootstraps with raw addresses.

func (*Node) StoreAt 

func (n *Node) StoreAt(ctx context.Context, peer net.Addr, storeKey a2al.NodeID, rec protocol.SignedRecord) (stored bool, peerID a2al.NodeID, meta deliverMeta, reason protocol.StoreReason, err error)

StoreAt sends STORE to peer. storeKey zero omits BodyStore.Key (receiver derives key from rec.Address). On success peerID is the remote node's identity from the STORE response. meta describes the outbound path actually used (for diagnostic logging). StoreAt sends a STORE RPC to peer and returns the outcome.

stored is true when the remote node confirmed holding the record — either by accepting the write (resp.Stored) or because it already had an equal-or-newer copy (resp.AlreadyHad). Both mean the record is safe on that node.

reason carries the machine-readable rejection cause when stored is false. It is StoreReasonOK (0) on success or when the remote is an older node that does not populate the field.

func (*Node) StoreRxCount added in v0.2.2 

func (n *Node) StoreRxCount() uint64

StoreRxCount returns the cumulative number of STORE RPCs accepted by this node.

type PeerHealthState added in v0.1.3 

type PeerHealthState uint8

PeerHealthState classifies a peer's reachability based on observed RPC outcomes. 

const (
	PeerHealthUnknown PeerHealthState = iota // no RPC history yet
	PeerHealthGood                           // last RPC succeeded, failCount == 0
	PeerHealthBad                            // consecutive failures >= badHealthThreshold
)

type PeerIdentity 

type PeerIdentity struct {
	Address      a2al.Address
	NodeID       a2al.NodeID
	ObservedWire []byte // BodyPong.observed_addr (how reporter sees us); may be nil
}

PeerIdentity holds the identity extracted from a PONG response.

type PunchFailReason added in v0.1.8 

type PunchFailReason int

PunchTransport is the optional ICE hole-punch integration layer.

When non-nil (injected via Config.PunchTransport), the DHT can reach NAT-bound peers that are not directly accessible over UDP. When nil (default), all DHT behaviour is identical to the current direct-UDP-only mode — no code paths change.

The interface is intentionally narrow: DHT never sees QUIC connections, streams, or ICE candidates. It only asks "can I send this byte slice to that NodeID?" and "please start punching that NodeID in the background".

Implementations live in the host layer (host/dht_punch_pool.go) and are wired in by the daemon at startup, following the same pattern as OnObservedAddr injection. PunchFailReason enumerates the reasons an ICE punch attempt can fail. Used by OnPunchComplete to enable differentiated handling in the DHT layer. 

const (
	// PunchFailNone indicates success (used internally; callers check success bool).
	PunchFailNone PunchFailReason = 0

	// PunchFailNoAgent means the signaling hub reported the remote has no
	// registered callee (hub returned "noagent"). The remote is likely offline
	// or does not support ICE. The DHT can stop probing this peer.
	PunchFailNoAgent PunchFailReason = 1

	// PunchFailICETimeout means ICE negotiation started but connectivity checks
	// timed out. The remote may be temporarily offline or behind a symmetric NAT.
	PunchFailICETimeout PunchFailReason = 2

	// PunchFailOther covers all other errors (TLS, network I/O, dial errors).
	PunchFailOther PunchFailReason = 3
)

type PunchTransport added in v0.1.8 

type PunchTransport interface {
	// SendTo attempts to deliver msg to nodeID via an existing punched QUIC
	// connection. ok=false means no active punched connection is available
	// for this peer; the caller must fall back to the UDP transport.
	// Called on the hot RPC path — must not block.
	SendTo(ctx context.Context, nodeID a2al.NodeID, msg []byte) (ok bool, err error)

	// Punch enqueues an asynchronous ICE hole-punch attempt for nodeID.
	// er supplies the signal URL and NAT classification from the peer's
	// signed endpoint record. priority is one of the PunchPriority*
	// constants. Non-blocking: returns immediately after enqueue.
	Punch(nodeID a2al.NodeID, er *protocol.EndpointRecord, priority int)

	// HasConn reports whether an active Mode B QUIC connection already exists
	// for nodeID. Used by triggerPunch to skip redundant punch attempts when
	// the pool already has a healthy connection.
	HasConn(nodeID a2al.NodeID) bool

	// InvalidateConn closes and evicts the Mode B QUIC connection for nodeID.
	// Called by sendAndWait when a QUIC-routed RPC times out, indicating the
	// connection is no longer viable. The next outboundPlan call will see
	// HasConn=false and fall back to UDP or deferICE naturally.
	// No-op when no connection exists for nodeID.
	InvalidateConn(nodeID a2al.NodeID)
}

type Query 

type Query struct {
	Alpha   int           // retained for API compatibility; slot engine uses queryAlpha
	Stagger time.Duration // retained for API compatibility; slot engine uses querySlotStagger
	// contains filtered or unexported fields
}

Query runs iterative FIND_NODE / FIND_VALUE (spec Step 8).

func NewQuery 

func NewQuery(n *Node) *Query

NewQuery builds a querier backed by n.

func (*Query) AggregateRecords 

func (q *Query) AggregateRecords(ctx context.Context, target a2al.NodeID, recType uint8) ([]protocol.SignedRecord, error)

AggregateRecords queries the network until the good+unknown candidate pools are exhausted, then merges and deduplicates all discovered records (Phase 4 Topic/Mailbox). Unlike FindRecords, there is no local-cache fast path: the network query always runs so that newly-joined publishers are discovered. Locally-cached records are seeded into the result set and merged with network results.

func (*Query) FindNode 

func (q *Query) FindNode(ctx context.Context, target a2al.NodeID) ([]protocol.NodeInfo, error)

FindNode runs iterative FIND_NODE until the good+unknown candidate pools are exhausted. Returns the K XOR-closest nodes discovered.

func (*Query) FindRecords 

func (q *Query) FindRecords(ctx context.Context, target a2al.NodeID, recType uint8) ([]protocol.SignedRecord, error)

FindRecords runs iterative FIND_VALUE (recType 0 = all). Returns on the first valid record found (optimistic strategy). Local store is checked first; if a valid cached record exists the network is not queried. The cache is transparently invalidated by the host layer when a subsequent connection attempt using the cached data fails.

Network-discovered records are written into the local store so that subsequent lookupEndpointRecord calls (e.g. from the replication NAT track) can find them without another round-trip.

func (*Query) Resolve 

func (q *Query) Resolve(ctx context.Context, target a2al.NodeID) (*protocol.EndpointRecord, error)

Resolve runs iterative FIND_VALUE for target NodeID and returns a verified endpoint record.

type ReachProfile added in v0.2.6 

type ReachProfile uint8

ReachProfile is a derived node-level reachability classification. It is not persisted; it guides probe/outbound/health gating without replacing AddrRank. 

const (
	ReachUnknown ReachProfile = iota
	ReachNAT                  // symmetric NAT or ICE-dependent peer
	ReachPublic               // stable advertised/public anchor (NatType < Symmetric)
)

type RecordAuthFunc 

type RecordAuthFunc func(key a2al.NodeID, rec protocol.SignedRecord, now time.Time) error

RecordAuthFunc decides whether a record may be stored at the given DHT key after signature and expiry checks (Phase 4: includes key binding for sovereign).

type Store 

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

Store is an in-memory record map keyed by DHT NodeID (Phase 4 multi-category).

func NewStore 

func NewStore(auth RecordAuthFunc, maxKeys int) *Store

NewStore creates an empty Store. auth is optional (see Config.RecordAuth). maxKeys limits distinct key count; 0 uses DefaultMaxTotalKeys.

func (*Store) AddOneShotSub added in v0.2.4 

func (s *Store) AddOneShotSub(key a2al.NodeID, addr net.UDPAddr)

AddOneShotSub registers addr to receive a DHT_PUSH when a new record is stored at key. Duplicate addr entries for the same key are deduplicated; existing TTL is refreshed.

func (*Store) ClearSoftExpiry added in v0.2.9 

func (s *Store) ClearSoftExpiry(key a2al.NodeID, recType uint8)

ClearSoftExpiry removes a soft-expiry deadline for a record slot. Called when the publisher's direct StoreAt confirms the record is authoritative. Thread-safe.

func (*Store) ConsumeOneShotSubs added in v0.2.4 

func (s *Store) ConsumeOneShotSubs(key a2al.NodeID) []oneShotSub

ConsumeOneShotSubs atomically removes and returns all valid (non-expired) subscriptions for key. The caller is responsible for sending DHT_PUSH to each returned address.

func (*Store) DebugRecords 

func (s *Store) DebugRecords(now time.Time) []StoreRecordDebug

DebugRecords lists non-expired verified records (spec §3.6).

func (*Store) Get 

func (s *Store) Get(key a2al.NodeID, now time.Time) *protocol.SignedRecord

Get returns the newest valid endpoint record (RecTypeEndpoint) at key, or nil.

func (*Store) GetAll 

func (s *Store) GetAll(key a2al.NodeID, recType uint8, now time.Time) []protocol.SignedRecord

GetAll returns verified non-expired records at key, optionally filtered by RecType (0 = all).

func (*Store) GetAllByAddress added in v0.1.6 

func (s *Store) GetAllByAddress(addr a2al.Address, recType uint8, now time.Time) []protocol.SignedRecord

GetAllByAddress returns verified non-expired records where sr.Address == addr, with optional RecType filter (0 = all types). Scans all key buckets; intended for low-frequency use (e.g. QUIC control plane exchange on connection setup).

func (*Store) Invalidate added in v0.1.6 

func (s *Store) Invalidate(key a2al.NodeID, recType uint8)

Invalidate removes locally-cached records for key, optionally filtered by recType (0 = all types). Called internally when a connection attempt to the peer fails, ensuring the next Resolve goes to the network for fresh data.

func (*Store) Len 

func (s *Store) Len() int

Len is the number of distinct key buckets with at least one record.

func (*Store) PruneExpiredOneShotSubs added in v0.2.4 

func (s *Store) PruneExpiredOneShotSubs()

PruneExpiredOneShotSubs removes all expired one-shot subscriptions across all keys. Should be called periodically (e.g. every 30s).

func (*Store) Put 

func (s *Store) Put(key a2al.NodeID, rec protocol.SignedRecord, now time.Time) error

Put stores rec at key. Zero key derives NodeID(rec.Address).

func (*Store) SetMaxKeys added in v0.1.7 

func (s *Store) SetMaxKeys(n int)

SetMaxKeys updates the maximum number of distinct DHT keys stored in-place. Safe to call concurrently; takes effect on the next eviction cycle.

func (*Store) SetSoftExpiry added in v0.2.9 

func (s *Store) SetSoftExpiry(key a2al.NodeID, recType uint8, deadline time.Time)

SetSoftExpiry marks a path-cached record slot with a node-local expiry deadline. After deadline, GetAll / Get suppress the record even if its signed TTL is still valid. Thread-safe.

type StoreRecordDebug 

type StoreRecordDebug struct {
	KeyNodeIDHex string `json:"key_node_id_hex"`
	AddressHex   string `json:"address_hex"`
	RecType      uint8  `json:"rec_type"`
	Seq          uint64 `json:"seq"`
	Timestamp    uint64 `json:"timestamp"`
	TTL          uint32 `json:"ttl_seconds"`
	PayloadLen   int    `json:"payload_cbor_len"`
}

StoreRecordDebug is a JSON-friendly store row (spec §3.6).

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.