xorbie

package module
v0.0.2 Latest Latest
Warning

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

 Go to latest Published: Aug 15, 2026 License: Apache-2.0, MIT Imports: 17 Imported by: 0

README

xorbie

A network neutral Kademlia library.

Extracted from zikade.

License

SPDX-License-Identifier: Apache-2.0 OR MIT

Documentation

Index

Constants

View Source 
const (
	// IncludeQueryID is the id for connectivity checks performed by the include state machine.
	// This identifier is used for routing network responses to the state machine.
	IncludeQueryID = coordt.QueryID("include")

	// ProbeQueryID is the id for connectivity checks performed by the probe state machine
	// This identifier is used for routing network responses to the state machine.
	ProbeQueryID = coordt.QueryID("probe")
)

Variables

View Source 
var ErrEventDropped = errors.New("event dropped")

ErrEventDropped is the error reported to the caller of an operation whose event was dropped because the behaviour that would have carried it out had no queue space.

View Source 
var ErrRequestDropped = errors.New("request dropped")

ErrRequestDropped is the error reported for a request dropped because no capacity was available for it, either for the node it was addressed to or across all nodes.

Functions

This section is empty.

Types

type Behaviour 

type Behaviour[I BehaviourEvent, O BehaviourEvent] interface {
	// Ready returns a channel that signals when the behaviour is ready to perform work.
	// A behaviour must signal whenever it has work available, including work that has
	// become available through the passage of time rather than through an event. It may
	// signal when it has none.
	Ready() <-chan struct{}

	// Notify informs the behaviour of an event. The behaviour may perform the event
	// immediately and queue the result, causing the behaviour to become ready.
	// It is safe to call Notify from the Perform method.
	Notify(ctx context.Context, ev I)

	// Perform gives the behaviour the opportunity to perform work or to return a queued
	// result as an event.
	Perform(ctx context.Context) (O, bool)
}

type BehaviourEvent 

type BehaviourEvent interface {
	// contains filtered or unexported methods
}

type BrdcstCommand 

type BrdcstCommand interface {
	BehaviourEvent
	// contains filtered or unexported methods
}

BrdcstCommand is a type of BehaviourEvent that instructs a [BrdcstBehaviour] to perform an action.

type BroadcastConfig 

type BroadcastConfig[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	// Logger is a structured logger that will be used when logging.
	Logger *slog.Logger

	// Tracer is the tracer that should be used to trace execution.
	Tracer trace.Tracer

	// Meter is the meter that should be used to record metrics.
	Meter metric.Meter

	// QueueCapacity is the maximum number of events that may be waiting to be processed by
	// the behaviour. Events arriving when the queue is full are dropped. It must be larger
	// than [NetworkConfig.Capacity], since a node handler queues a response here before
	// releasing the capacity it held, so that many responses can be waiting at once.
	QueueCapacity int

	// VerifyResponse reports whether a node's reply to a stored record shows that it stored
	// the record, returning a nil error when it did. A nil VerifyResponse takes every reply
	// that is not itself an error as a success.
	VerifyResponse func(req, resp M) error
}

func DefaultBroadcastConfig 

func DefaultBroadcastConfig[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]]() *BroadcastConfig[K, N, M]

func (*BroadcastConfig[K, N, M]) Validate 

func (cfg *BroadcastConfig[K, N, M]) Validate() error

Validate checks the configuration options and returns an error if any have invalid values.

type BroadcastWaiter 

type BroadcastWaiter[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	// contains filtered or unexported fields
}

A BroadcastWaiter implements QueryMonitor for broadcasts

func NewBroadcastWaiter 

func NewBroadcastWaiter[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]](n int) *BroadcastWaiter[K, N, M]

func (*BroadcastWaiter[K, N, M]) Finished 

func (w *BroadcastWaiter[K, N, M]) Finished() <-chan CtxEvent[*EventBroadcastFinished[K, N]]

func (*BroadcastWaiter[K, N, M]) NotifyFinished 

func (w *BroadcastWaiter[K, N, M]) NotifyFinished() chan<- CtxEvent[*EventBroadcastFinished[K, N]]

func (*BroadcastWaiter[K, N, M]) NotifyProgressed 

func (w *BroadcastWaiter[K, N, M]) NotifyProgressed() chan<- CtxEvent[*EventQueryProgressed[K, N, M]]

func (*BroadcastWaiter[K, N, M]) Progressed 

func (w *BroadcastWaiter[K, N, M]) Progressed() <-chan CtxEvent[*EventQueryProgressed[K, N, M]]

type BufferedRoutingNotifier 

type BufferedRoutingNotifier[K kad.Key[K], N kad.NodeID[K]] struct {
	// contains filtered or unexported fields
}

A BufferedRoutingNotifier is a RoutingNotifier that buffers RoutingNotification events and provides methods to expect occurrences of specific events. It is designed for use in a test environment.

func NewBufferedRoutingNotifier 

func NewBufferedRoutingNotifier[K kad.Key[K], N kad.NodeID[K]]() *BufferedRoutingNotifier[K, N]

func (*BufferedRoutingNotifier[K, N]) Expect 

func (w *BufferedRoutingNotifier[K, N]) Expect(ctx context.Context, expected RoutingNotification) (RoutingNotification, error)

func (*BufferedRoutingNotifier[K, N]) ExpectRoutingRemoved 

func (w *BufferedRoutingNotifier[K, N]) ExpectRoutingRemoved(ctx context.Context, id N) (*EventRoutingRemoved[K, N], error)

ExpectRoutingRemoved blocks until an EventRoutingRemoved event is seen for the specified node id

func (*BufferedRoutingNotifier[K, N]) ExpectRoutingUpdated 

func (w *BufferedRoutingNotifier[K, N]) ExpectRoutingUpdated(ctx context.Context, id N) (*EventRoutingUpdated[K, N], error)

ExpectRoutingUpdated blocks until an EventRoutingUpdated event is seen for the specified node id

func (*BufferedRoutingNotifier[K, N]) Notify 

func (w *BufferedRoutingNotifier[K, N]) Notify(ctx context.Context, ev RoutingNotification)

type Coordinator 

type Coordinator[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	// contains filtered or unexported fields
}

A Coordinator coordinates the state machines that comprise a Kademlia DHT

func NewCoordinator 

func NewCoordinator[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]](
	self N,
	rtr coordt.Router[K, N, M],
	rt routing.RoutingTableCpl[K, N],
	cplFn routing.NodeIDForCplFunc[K, N],
	cfg *CoordinatorConfig[K, N, M],
) (*Coordinator[K, N, M], error)

func (*Coordinator[K, N, M]) AddNodes 

func (c *Coordinator[K, N, M]) AddNodes(ctx context.Context, ids []N) error

AddNodes suggests new DHT nodes to be added to the routing table. If the routing table is updated as a result of this operation an EventRoutingUpdated notification is emitted on the routing notification channel.

func (*Coordinator[K, N, M]) Bootstrap 

func (c *Coordinator[K, N, M]) Bootstrap(ctx context.Context) error

Bootstrap instructs the dht to begin bootstrapping the routing table from the nodes configured as RoutingConfig.BootstrapPeers. A bootstrap also starts automatically whenever the routing table holds fewer than RoutingConfig.BootstrapMinimumPopulation nodes.

func (*Coordinator[K, N, M]) BroadcastRecord 

func (c *Coordinator[K, N, M]) BroadcastRecord(ctx context.Context, msg M) error

func (*Coordinator[K, N, M]) BroadcastStatic 

func (c *Coordinator[K, N, M]) BroadcastStatic(ctx context.Context, msg M, seeds []N) error

func (*Coordinator[K, N, M]) Close 

func (c *Coordinator[K, N, M]) Close() error

Close cleans up all resources associated with this Coordinator.

func (*Coordinator[K, N, M]) GetClosestNodes 

func (c *Coordinator[K, N, M]) GetClosestNodes(ctx context.Context, k K, n int) ([]N, error)

GetClosestNodes requests the n closest nodes to the key from the node's local routing table.

func (*Coordinator[K, N, M]) ID 

func (c *Coordinator[K, N, M]) ID() N

func (*Coordinator[K, N, M]) IsRoutable 

func (c *Coordinator[K, N, M]) IsRoutable(ctx context.Context, id N) bool

IsRoutable reports whether the supplied node is present in the local routing table.

func (*Coordinator[K, N, M]) NotifyConnectivity 

func (c *Coordinator[K, N, M]) NotifyConnectivity(ctx context.Context, id N)

NotifyConnectivity notifies the coordinator that a node has passed a connectivity check which means it is connected and supports finding closer nodes

func (*Coordinator[K, N, M]) NotifyNonConnectivity 

func (c *Coordinator[K, N, M]) NotifyNonConnectivity(ctx context.Context, id N)

NotifyNonConnectivity notifies the coordinator that a node has failed a connectivity check which means it is not connected and/or it doesn't support finding closer nodes

func (*Coordinator[K, N, M]) QueryClosest 

func (c *Coordinator[K, N, M]) QueryClosest(ctx context.Context, target K, fn coordt.QueryFunc[K, N, M], numResults int) ([]N, coordt.QueryStats, error)

QueryClosest starts a query that attempts to find the closest nodes to the target key. It returns the closest nodes found to the target key and statistics on the actions of the query.

The supplied [QueryFunc] is called after each successful request to a node with the ID of the node, the response received from the find nodes request made to the node and the current query stats. The query terminates when [QueryFunc] returns an error or when the query has visited the configured minimum number of closest nodes (default 20)

numResults specifies the minimum number of nodes to successfully contact before considering iteration complete. The query is considered to be exhausted when it has received responses from at least this number of nodes and there are no closer nodes remaining to be contacted. A default of 20 is used if this value is less than 1.

func (*Coordinator[K, N, M]) QueryMessage 

func (c *Coordinator[K, N, M]) QueryMessage(ctx context.Context, msg M, fn coordt.QueryFunc[K, N, M], numResults int) ([]N, coordt.QueryStats, error)

QueryMessage starts a query that iterates over the closest nodes to the target key in the supplied message. The message is sent to each node that is visited.

The supplied [QueryFunc] is called after each successful request to a node with the ID of the node, the response received from the find nodes request made to the node and the current query stats. The query terminates when [QueryFunc] returns an error or when the query has visited the configured minimum number of closest nodes (default 20)

numResults specifies the minimum number of nodes to successfully contact before considering iteration complete. The query is considered to be exhausted when it has received responses from at least this number of nodes and there are no closer nodes remaining to be contacted. A default of 20 is used if this value is less than 1.

func (*Coordinator[K, N, M]) SetRoutingNotifier 

func (c *Coordinator[K, N, M]) SetRoutingNotifier(rn RoutingNotifier)

type CoordinatorConfig 

type CoordinatorConfig[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	// Logger is a structured logger that will be used when logging.
	Logger *slog.Logger

	// MeterProvider is the the meter provider to use when initialising metric instruments.
	MeterProvider metric.MeterProvider

	// TracerProvider is the tracer provider to use when initialising tracing
	TracerProvider trace.TracerProvider

	// Network is the configuration used for the [NetworkBehaviour] which sends requests to other nodes.
	Network NetworkConfig

	// Routing is the configuration used for the [RoutingBehaviour] which maintains the health of the routing table.
	Routing RoutingConfig[K, N]

	// Query is the configuration used for the [PooledQueryBehaviour] which manages the execution of user queries.
	Query QueryConfig

	// Brdcst is the configuration used for the [PooledBroadcastBehaviour] which manages the storing of records with other nodes.
	Brdcst BroadcastConfig[K, N, M]
}

func DefaultCoordinatorConfig 

func DefaultCoordinatorConfig[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]]() *CoordinatorConfig[K, N, M]

func (*CoordinatorConfig[K, N, M]) Validate 

func (cfg *CoordinatorConfig[K, N, M]) Validate() error

Validate checks the configuration options and returns an error if any have invalid values.

type CtxEvent 

type CtxEvent[E any] struct {
	Ctx   context.Context
	Event E
}

CtxEvent holds and event with an associated context which may carry deadlines or tracing information pertinent to the event.

type EventAddNode 

type EventAddNode[K kad.Key[K], N kad.NodeID[K]] struct {
	NodeID N
}

EventAddNode notifies the routing behaviour of a potential new node.

type EventBootstrapFinished 

type EventBootstrapFinished struct {
	Stats query.QueryStats

	// Err records why the bootstrap ended when it ended for a reason other than visiting
	// every node it could, and is nil otherwise.
	Err error
}

EventBootstrapFinished is emitted by the coordinator when a bootstrap has finished, either through running to completion or by being canceled.

type EventBroadcastFinished 

type EventBroadcastFinished[K kad.Key[K], N kad.NodeID[K]] struct {
	QueryID   coordt.QueryID
	Contacted []N
	Errors    map[string]struct {
		Node N
		Err  error
	}

	// Err records why the broadcast ended when it ended without being attempted, and is
	// nil otherwise. A broadcast that ran records per node outcomes in Errors instead.
	Err error
}

EventBroadcastFinished is emitted by the coordinator when a broadcasting a record to the network has finished, either through running to completion or by being canceled.

type EventGetCloserNodesFailure 

type EventGetCloserNodesFailure[K kad.Key[K], N kad.NodeID[K]] struct {
	QueryID coordt.QueryID
	To      N // To is the node that the GetCloserNodes request was sent to.
	Target  K
	Err     error
}

EventGetCloserNodesFailure notifies a behaviour that a GetCloserNodes request, initiated by an EventOutboundGetCloserNodes event has failed to produce a valid response.

type EventGetCloserNodesSuccess 

type EventGetCloserNodesSuccess[K kad.Key[K], N kad.NodeID[K]] struct {
	QueryID     coordt.QueryID
	To          N // To is the node that the GetCloserNodes request was sent to.
	Target      K
	CloserNodes []N
}

EventGetCloserNodesSuccess notifies a behaviour that a GetCloserNodes request, initiated by an EventOutboundGetCloserNodes event has produced a successful response.

type EventNotifyConnectivity 

type EventNotifyConnectivity[K kad.Key[K], N kad.NodeID[K]] struct {
	NodeID N
}

EventNotifyConnectivity notifies a behaviour that a node's connectivity and support for finding closer nodes has been confirmed such as from a successful query response or an inbound query. This should not be used for general connections to the host but only when it is confirmed that the node responds to requests for closer nodes.

type EventNotifyNonConnectivity 

type EventNotifyNonConnectivity[K kad.Key[K], N kad.NodeID[K]] struct {
	NodeID N
}

EventNotifyNonConnectivity notifies a behaviour that a node does not have connectivity and/or does not support finding closer nodes is known.

type EventOutboundGetCloserNodes 

type EventOutboundGetCloserNodes[K kad.Key[K], N kad.NodeID[K]] struct {
	QueryID coordt.QueryID
	To      N
	Target  K
	Notify  Notify[BehaviourEvent]
}

type EventOutboundSendMessage 

type EventOutboundSendMessage[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	QueryID coordt.QueryID
	To      N
	Message M
	Notify  Notify[BehaviourEvent]
}

type EventQueryFinished 

type EventQueryFinished[K kad.Key[K], N kad.NodeID[K]] struct {
	QueryID      coordt.QueryID
	Stats        query.QueryStats
	ClosestNodes []N

	// Err records why the query ended when it ended for a reason other than visiting
	// every node it could, and is nil otherwise. ClosestNodes is not populated when
	// Err is set.
	Err error
}

EventQueryFinished is emitted by the coordinator when a query has finished, either through running to completion or by being canceled.

type EventQueryProgressed 

type EventQueryProgressed[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	QueryID  coordt.QueryID
	NodeID   N
	Response M
	Stats    query.QueryStats
}

EventQueryProgressed is emitted by the coordinator when a query has received a response from a node.

type EventRoutingPoll 

type EventRoutingPoll struct{}

EventRoutingPoll notifies a routing behaviour that it may proceed with any pending work.

type EventRoutingRemoved 

type EventRoutingRemoved[K kad.Key[K], N kad.NodeID[K]] struct {
	NodeID N
}

EventRoutingRemoved is emitted by the coordinator when new node has been removed from the routing table.

type EventRoutingUpdated 

type EventRoutingUpdated[K kad.Key[K], N kad.NodeID[K]] struct {
	NodeID N
}

EventRoutingUpdated is emitted by the coordinator when a new node has been verified and added to the routing table.

type EventSendMessageFailure 

type EventSendMessageFailure[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	QueryID coordt.QueryID
	Request M
	To      N // To is the node that the SendMessage request was sent to.
	Target  K
	Err     error
}

EventSendMessageFailure notifies a behaviour that a SendMessage request, initiated by an EventOutboundSendMessage event has failed to produce a valid response.

type EventSendMessageSuccess 

type EventSendMessageSuccess[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	QueryID     coordt.QueryID
	Request     M
	To          N // To is the node that the SendMessage request was sent to.
	Response    M
	CloserNodes []N
}

EventSendMessageSuccess notifies a behaviour that a SendMessage request, initiated by an EventOutboundSendMessage event has produced a successful response.

type EventStartBootstrap 

type EventStartBootstrap[K kad.Key[K], N kad.NodeID[K]] struct {
	// SeedNodes are the nodes the bootstrap should start from. When empty the nodes
	// configured as [RoutingConfig.BootstrapPeers] are used.
	SeedNodes []N
}

type EventStartBroadcast 

type EventStartBroadcast[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	QueryID coordt.QueryID
	Target  K
	Message M
	Seed    []N
	Config  brdcst.Config
	Notify  QueryMonitor[K, N, M, *EventBroadcastFinished[K, N]]
}

EventStartBroadcast starts a new

type EventStartFindCloserQuery 

type EventStartFindCloserQuery[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	QueryID           coordt.QueryID
	Target            K
	KnownClosestNodes []N
	Notify            QueryMonitor[K, N, M, *EventQueryFinished[K, N]]
	NumResults        int // the minimum number of nodes to successfully contact before considering iteration complete
}

type EventStartMessageQuery 

type EventStartMessageQuery[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	QueryID           coordt.QueryID
	Target            K
	Message           M
	KnownClosestNodes []N
	Notify            QueryMonitor[K, N, M, *EventQueryFinished[K, N]]
	NumResults        int // the minimum number of nodes to successfully contact before considering iteration complete
}

type EventStopQuery 

type EventStopQuery struct {
	QueryID coordt.QueryID
}

type NetworkBehaviour 

type NetworkBehaviour[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	// contains filtered or unexported fields
}

func NewNetworkBehaviour 

func NewNetworkBehaviour[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]](rtr coordt.Router[K, N, M], cfg *NetworkConfig) (*NetworkBehaviour[K, N, M], error)

func (*NetworkBehaviour[K, N, M]) Close 

func (b *NetworkBehaviour[K, N, M]) Close()

Close stops all the node handlers managed by the behaviour, releasing the goroutines they use to send messages. It is safe to call Close more than once.

func (*NetworkBehaviour[K, N, M]) Notify 

func (b *NetworkBehaviour[K, N, M]) Notify(ctx context.Context, ev BehaviourEvent)

Notify hands a request to the node handler for the node it is addressed to. It does not block: a request that finds no available capacity is dropped and reported back to whoever asked for it as an ordinary failure, since blocking here would stop the event loop that is the only thing able to drain the handler.

func (*NetworkBehaviour[K, N, M]) Perform 

func (b *NetworkBehaviour[K, N, M]) Perform(ctx context.Context) (BehaviourEvent, bool)

func (*NetworkBehaviour[K, N, M]) Ready 

func (b *NetworkBehaviour[K, N, M]) Ready() <-chan struct{}

type NetworkCommand 

type NetworkCommand interface {
	BehaviourEvent
	// contains filtered or unexported methods
}

NetworkCommand is a type of BehaviourEvent that instructs a NetworkBehaviour to perform an action.

type NetworkConfig 

type NetworkConfig struct {
	// Logger is a structured logger that will be used when logging.
	Logger *slog.Logger

	// Tracer is the tracer that should be used to trace execution.
	Tracer trace.Tracer

	// Meter is the meter that should be used to record metrics.
	Meter metric.Meter

	// Capacity is the maximum number of requests that may be queued or in flight across
	// all nodes.
	Capacity int

	// NodeCapacity is the maximum number of requests that may be queued or in flight for
	// any one node.
	NodeCapacity int

	// IdleTimeout is how long an unused node handler is kept before it is evicted and the
	// goroutine it uses to send messages is released.
	IdleTimeout time.Duration
}

func DefaultNetworkConfig 

func DefaultNetworkConfig() *NetworkConfig

func (*NetworkConfig) Validate 

func (cfg *NetworkConfig) Validate() error

Validate checks the configuration options and returns an error if any have invalid values.

type NodeHandler 

type NodeHandler[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	// contains filtered or unexported fields
}

A NodeHandler sends requests to a single node, one at a time, from a goroutine of its own. Requests that arrive with no available capacity are dropped rather than queued.

func NewNodeHandler 

func NewNodeHandler[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]](self N, rtr coordt.Router[K, N, M], slots *slots, cfg *NetworkConfig, onIdle func(*NodeHandler[K, N, M]) bool) *NodeHandler[K, N, M]

func (*NodeHandler[K, N, M]) Close 

func (h *NodeHandler[K, N, M]) Close()

Close stops the handler from sending any further requests and discards the requests it has accepted but not yet sent. It is safe to call Close more than once.

func (*NodeHandler[K, N, M]) ID 

func (h *NodeHandler[K, N, M]) ID() N

func (*NodeHandler[K, N, M]) Notify 

func (h *NodeHandler[K, N, M]) Notify(ctx context.Context, ev NodeHandlerRequest) bool

Notify accepts a request to be sent to the handler's node, reporting whether capacity was available for it. It does not block.

type NodeHandlerRequest 

type NodeHandlerRequest interface {
	BehaviourEvent
	// contains filtered or unexported methods
}

type NodeHandlerResponse 

type NodeHandlerResponse interface {
	BehaviourEvent
	// contains filtered or unexported methods
}

type Notify 

type Notify[E BehaviourEvent] interface {
	Notify(ctx context.Context, ev E)
}

Notify is the interface that a components to implement to be notified of BehaviourEvent's.

type NotifyCloser 

type NotifyCloser[E BehaviourEvent] interface {
	Notify[E]
	Close()
}

type NotifyFunc 

type NotifyFunc[E BehaviourEvent] func(ctx context.Context, ev E)

func (NotifyFunc[E]) Notify 

func (f NotifyFunc[E]) Notify(ctx context.Context, ev E)

type PooledBroadcastBehaviour 

type PooledBroadcastBehaviour[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	// contains filtered or unexported fields
}

func NewPooledBroadcastBehaviour 

func NewPooledBroadcastBehaviour[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]](brdcstPool *brdcst.Pool[K, N, M], cfg *BroadcastConfig[K, N, M]) (*PooledBroadcastBehaviour[K, N, M], error)

func (*PooledBroadcastBehaviour[K, N, M]) Notify 

func (b *PooledBroadcastBehaviour[K, N, M]) Notify(ctx context.Context, ev BehaviourEvent)

func (*PooledBroadcastBehaviour[K, N, M]) Perform 

func (b *PooledBroadcastBehaviour[K, N, M]) Perform(ctx context.Context) (out BehaviourEvent, performed bool)

func (*PooledBroadcastBehaviour[K, N, M]) Ready 

func (b *PooledBroadcastBehaviour[K, N, M]) Ready() <-chan struct{}

type QueryBehaviour 

type QueryBehaviour[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	// contains filtered or unexported fields
}

QueryBehaviour holds the behaviour and state for managing a pool of queries.

func NewQueryBehaviour 

func NewQueryBehaviour[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]](self N, cfg *QueryConfig) (*QueryBehaviour[K, N, M], error)

NewQueryBehaviour initialises a new QueryBehaviour, setting up the query pool and other internal state.

func (*QueryBehaviour[K, N, M]) Notify 

func (p *QueryBehaviour[K, N, M]) Notify(ctx context.Context, ev BehaviourEvent)

Notify receives a behaviour event and takes appropriate actions such as starting, stopping, or updating queries. It also queues events for later processing and triggers the advancement of the query pool if applicable.

func (*QueryBehaviour[K, N, M]) Perform 

func (p *QueryBehaviour[K, N, M]) Perform(ctx context.Context) (out BehaviourEvent, performed bool)

Perform executes the next available task from the queue of pending events or advances the query pool. Returns an event containing the result of the work performed and a true value, or nil and a false value if no event was generated.

func (*QueryBehaviour[K, N, M]) Ready 

func (p *QueryBehaviour[K, N, M]) Ready() <-chan struct{}

Ready returns a channel that signals when the pooled query behaviour is ready to perform work.

type QueryCommand 

type QueryCommand interface {
	BehaviourEvent
	// contains filtered or unexported methods
}

QueryCommand is a type of BehaviourEvent that instructs a QueryBehaviour to perform an action.

type QueryConfig 

type QueryConfig struct {
	// Logger is a structured logger that will be used when logging.
	Logger *slog.Logger

	// Tracer is the tracer that should be used to trace execution.
	Tracer trace.Tracer

	// Meter is the meter that should be used to record metrics.
	Meter metric.Meter

	// QueueCapacity is the maximum number of events that may be waiting to be processed by
	// the behaviour. Events arriving when the queue is full are dropped. It must be larger
	// than [NetworkConfig.Capacity], since a node handler queues a response here before
	// releasing the capacity it held, so that many responses can be waiting at once.
	QueueCapacity int

	// Concurrency is the maximum number of queries that may be waiting for message responses at any one time.
	Concurrency int

	// Timeout the time to wait before terminating a query that is not making progress.
	Timeout time.Duration

	// RequestConcurrency is the maximum number of concurrent requests that each query may have in flight.
	RequestConcurrency int

	// RequestTimeout is the timeout queries should use for contacting a single node
	RequestTimeout time.Duration
}

func DefaultQueryConfig 

func DefaultQueryConfig() *QueryConfig

func (*QueryConfig) Validate 

func (cfg *QueryConfig) Validate() error

Validate checks the configuration options and returns an error if any have invalid values.

type QueryMonitor 

type QueryMonitor[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N], E TerminalQueryEvent] interface {
	// NotifyProgressed returns a channel that can be used to send notification that a
	// query has made progress. If the notification cannot be sent then it will be
	// queued and retried at a later time. If the query completes before the progress
	// notification can be sent the notification will be discarded.
	NotifyProgressed() chan<- CtxEvent[*EventQueryProgressed[K, N, M]]

	// NotifyFinished returns a channel that can be used to send the notification that a
	// query has completed. It is up to the implemention to ensure that the channel has enough
	// capacity to receive the single notification.
	// The sender must close all other QueryNotifier channels before sending on the NotifyFinished channel.
	// The sender may attempt to drain any pending notifications before closing the other channels.
	// The NotifyFinished channel will be closed once the sender has attempted to send the Finished notification.
	NotifyFinished() chan<- CtxEvent[E]
}

A QueryMonitor receives event notifications on the progress of a query

type QueryMonitorHook 

type QueryMonitorHook[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N], E TerminalQueryEvent] struct {
	BeforeProgressed func()
	BeforeFinished   func()
	// contains filtered or unexported fields
}

QueryMonitorHook wraps a QueryMonitor interface and provides hooks that are invoked before calls to the QueryMonitor methods are forwarded.

func NewQueryMonitorHook 

func NewQueryMonitorHook[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N], E TerminalQueryEvent](qm QueryMonitor[K, N, M, E]) *QueryMonitorHook[K, N, M, E]

func (*QueryMonitorHook[K, N, M, E]) NotifyFinished 

func (n *QueryMonitorHook[K, N, M, E]) NotifyFinished() chan<- CtxEvent[E]

func (*QueryMonitorHook[K, N, M, E]) NotifyProgressed 

func (n *QueryMonitorHook[K, N, M, E]) NotifyProgressed() chan<- CtxEvent[*EventQueryProgressed[K, N, M]]

type QueryWaiter 

type QueryWaiter[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	// contains filtered or unexported fields
}

A QueryWaiter implements QueryMonitor for general queries

func NewQueryWaiter 

func NewQueryWaiter[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]](n int) *QueryWaiter[K, N, M]

func (*QueryWaiter[K, N, M]) Finished 

func (w *QueryWaiter[K, N, M]) Finished() <-chan CtxEvent[*EventQueryFinished[K, N]]

func (*QueryWaiter[K, N, M]) NotifyFinished 

func (w *QueryWaiter[K, N, M]) NotifyFinished() chan<- CtxEvent[*EventQueryFinished[K, N]]

func (*QueryWaiter[K, N, M]) NotifyProgressed 

func (w *QueryWaiter[K, N, M]) NotifyProgressed() chan<- CtxEvent[*EventQueryProgressed[K, N, M]]

func (*QueryWaiter[K, N, M]) Progressed 

func (w *QueryWaiter[K, N, M]) Progressed() <-chan CtxEvent[*EventQueryProgressed[K, N, M]]

type RoutingBehaviour 

type RoutingBehaviour[K kad.Key[K], N kad.NodeID[K]] struct {
	// contains filtered or unexported fields
}

A RoutingBehaviour provides the behaviours for bootstrapping and maintaining a DHT's routing table.

func ComposeRoutingBehaviour 

func ComposeRoutingBehaviour[K kad.Key[K], N kad.NodeID[K]](
	self N,
	bootstrap coordt.StateMachine[routing.BootstrapEvent, routing.BootstrapState],
	include coordt.StateMachine[routing.IncludeEvent, routing.IncludeState],
	probe coordt.StateMachine[routing.ProbeEvent, routing.ProbeState],
	explore coordt.StateMachine[routing.ExploreEvent, routing.ExploreState],
	cfg *RoutingConfig[K, N],
) (*RoutingBehaviour[K, N], error)

ComposeRoutingBehaviour creates a RoutingBehaviour composed of the supplied state machines. The state machines are assumed to pre-configured so any RoutingConfig values relating to the state machines will not be applied.

func NewRoutingBehaviour 

func NewRoutingBehaviour[K kad.Key[K], N kad.NodeID[K]](self N, rt routing.RoutingTableCpl[K, N], cplFn routing.NodeIDForCplFunc[K, N], cfg *RoutingConfig[K, N]) (*RoutingBehaviour[K, N], error)

func (*RoutingBehaviour[K, N]) Notify 

func (r *RoutingBehaviour[K, N]) Notify(ctx context.Context, ev BehaviourEvent)

func (*RoutingBehaviour[K, N]) Perform 

func (r *RoutingBehaviour[K, N]) Perform(ctx context.Context) (out BehaviourEvent, performed bool)

func (*RoutingBehaviour[K, N]) Ready 

func (r *RoutingBehaviour[K, N]) Ready() <-chan struct{}

type RoutingCommand 

type RoutingCommand interface {
	BehaviourEvent
	// contains filtered or unexported methods
}

RoutingCommand is a type of BehaviourEvent that instructs a RoutingBehaviour to perform an action.

type RoutingConfig 

type RoutingConfig[K kad.Key[K], N kad.NodeID[K]] struct {
	// Logger is a structured logger that will be used when logging.
	Logger *slog.Logger

	// Tracer is the tracer that should be used to trace execution.
	Tracer trace.Tracer

	// Meter is the meter that should be used to record metrics.
	Meter metric.Meter

	// QueueCapacity is the maximum number of events that may be waiting to be processed by
	// the behaviour. Events arriving when the queue is full are dropped. It must be larger
	// than [NetworkConfig.Capacity], since a node handler queues a response here before
	// releasing the capacity it held, so that many responses can be waiting at once.
	QueueCapacity int

	// BootstrapTimeout is the time the behaviour should wait before terminating a bootstrap if it is not making progress.
	BootstrapTimeout time.Duration

	// BootstrapRequestConcurrency is the maximum number of concurrent requests that the behaviour may have in flight during bootstrap.
	BootstrapRequestConcurrency int

	// BootstrapRequestTimeout is the timeout the behaviour should use when attempting to contact a node during bootstrap.
	BootstrapRequestTimeout time.Duration

	// BootstrapPeers is the list of nodes used to bootstrap the routing table.
	BootstrapPeers []N

	// BootstrapMinimumPopulation is the routing table population below which the behaviour should
	// start a bootstrap automatically. Zero means a bootstrap is only ever started on request.
	BootstrapMinimumPopulation int

	// BootstrapRetryInterval is the minimum time the behaviour should leave between bootstraps
	// started because the routing table population is below BootstrapMinimumPopulation.
	BootstrapRetryInterval time.Duration

	// ConnectivityCheckTimeout is the timeout the behaviour should use when performing a connectivity check.
	ConnectivityCheckTimeout time.Duration

	// ProbeRequestConcurrency is the maximum number of concurrent requests that the behaviour may have in flight while performing
	// connectivity checks for nodes in the routing table.
	ProbeRequestConcurrency int

	// ProbeCheckInterval is the time interval the behaviour should use between connectivity checks for the same node in the routing table.
	ProbeCheckInterval time.Duration

	// IncludeQueueCapacity is the maximum number of nodes the behaviour should keep queued as candidates for inclusion in the routing table.
	IncludeQueueCapacity int

	// IncludeRequestConcurrency is the maximum number of concurrent requests that the behaviour may have in flight while performing
	// connectivity checks for nodes in the inclusion candidate queue.
	IncludeRequestConcurrency int

	// ExploreTimeout is the time the behaviour should wait before terminating an exploration of a routing table bucket if it is not making progress.
	ExploreTimeout time.Duration

	// ExploreRequestConcurrency is the maximum number of concurrent requests that the behaviour may have in flight while exploring the
	// network to increase routing table occupancy.
	ExploreRequestConcurrency int

	// ExploreRequestTimeout is the timeout the behaviour should use when attempting to contact a node while exploring the
	// network to increase routing table occupancy.
	ExploreRequestTimeout time.Duration

	// ExploreMaximumCpl is the maximum CPL (common prefix length) the behaviour should explore to increase routing table occupancy.
	// All CPLs from this value to zero will be explored on a repeating schedule.
	ExploreMaximumCpl int

	// ExploreInterval is the base time interval the behaviour should leave between explorations of the same CPL.
	// See the documentation for [routing.DynamicExploreSchedule] for the precise formula used to calculate explore intervals.
	ExploreInterval time.Duration

	// ExploreIntervalMultiplier is a factor that is applied to the base time interval for CPLs lower than the maximum to increase the delay between
	// explorations for lower CPLs.
	// See the documentation for [routing.DynamicExploreSchedule] for the precise formula used to calculate explore intervals.
	ExploreIntervalMultiplier float64

	// ExploreIntervalJitter is a factor that is used to increase the calculated interval for an exploratiion by a small random amount.
	// It must be between 0 and 0.05. When zero, no jitter is applied.
	// See the documentation for [routing.DynamicExploreSchedule] for the precise formula used to calculate explore intervals.
	ExploreIntervalJitter float64
}

func DefaultRoutingConfig 

func DefaultRoutingConfig[K kad.Key[K], N kad.NodeID[K]]() *RoutingConfig[K, N]

func (*RoutingConfig[K, N]) Validate 

func (cfg *RoutingConfig[K, N]) Validate() error

Validate checks the configuration options and returns an error if any have invalid values.

type RoutingNotification 

type RoutingNotification interface {
	BehaviourEvent
	// contains filtered or unexported methods
}

type RoutingNotifier 

type RoutingNotifier interface {
	Notify(context.Context, RoutingNotification)
}

type Telemetry 

type Telemetry struct {
	Tracer trace.Tracer
	// contains filtered or unexported fields
}

Telemetry is the struct that holds a reference to all metrics and the tracer used by the coordinator and its components. Make sure to also register the [MeterProviderOpts] with your custom or the global metric.MeterProvider.

func NewTelemetry 

func NewTelemetry(meterProvider metric.MeterProvider, tracerProvider trace.TracerProvider) (*Telemetry, error)

NewTelemetry initializes a Telemetry struct with the given meter and tracer providers.

func (*Telemetry) RecordEventLoopPass 

func (t *Telemetry) RecordEventLoopPass(ctx context.Context, busy time.Duration)

RecordEventLoopPass records one pass of the coordinator's event loop and the time it spent working. The rate at which that time accumulates is the loop's occupancy: the fraction of wall clock time its single worker goroutine is unavailable to take on anything else.

type TerminalQueryEvent 

type TerminalQueryEvent interface {
	BehaviourEvent
	// contains filtered or unexported methods
}

TerminalQueryEvent is a type of BehaviourEvent that indicates a query has completed.

type Waiter 

type Waiter[E BehaviourEvent] struct {
	// contains filtered or unexported fields
}

A Waiter is a Notifiee whose Notify method forwards the notified event to a channel which a client can wait on.

func NewWaiter 

func NewWaiter[E BehaviourEvent]() *Waiter[E]

func (*Waiter[E]) Chan 

func (w *Waiter[E]) Chan() <-chan WaiterEvent[E]

func (*Waiter[E]) Close 

func (w *Waiter[E]) Close()

Close signals that the waiter should not forward and further calls to Notify. It closes the waiter channel so a client selecting on it will receive the close operation.

func (*Waiter[E]) Notify 

func (w *Waiter[E]) Notify(ctx context.Context, ev E)

type WaiterEvent 

type WaiterEvent[E BehaviourEvent] struct {
	Ctx   context.Context
	Event E
}

Source Files

View all Source files

Directories

Path Synopsis
Package brdcst contains state machines that implement algorithms for broadcasting records into the DHT network.
 Package brdcst contains state machines that implement algorithms for broadcasting records into the DHT network.
internal
kadtest
nettest
tiny
Package tiny implements Kademlia types suitable for tiny test networks
 Package tiny implements Kademlia types suitable for tiny test networks

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.