query

package
v0.0.13 Latest Latest
Warning

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

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

Documentation

Overview

Package query contains state machines that implement strategies for querying the network.

State machines in this package are passive. They do not perform their own I/O or start goroutines. The caller drives them by calling Advance with the current time and an event, and receives back a state that describes what the caller should do next. The caller carries out any network work the state asks for, then reports the outcome as a later event. Time is a parameter of Advance rather than something the machine reads, so a test can supply any clock.

Query

A Query is one iterative Kademlia lookup. It contacts nodes in ascending order of distance from a target key. Each response supplies further nodes, so the query walks towards the target until the closest nodes it knows of stop improving. The order in which nodes are visited is decided by the NodeIter the query is given, not by the query itself; ClosestNodesIter orders by distance from the target.

There are three kinds of query, made by three constructors:

  • NewFindCloserQuery asks each node for the nodes it holds closest to the target and takes those as the nodes to visit next. It emits StateQueryFindCloser. It stops at the QueryConfig.NumResults closest nodes.
  • NewQuery sends a message to each node and takes the closer nodes from the reply, so one reply both answers the message and supplies the next nodes to visit. It emits StateQuerySendMessage. It also stops at the QueryConfig.NumResults closest nodes.
  • NewCoverageQuery is a find-closer query that does not stop at the closest nodes. It enumerates every node inside the region named by a prefix of the target and reports them all.

Each Advance does a bounded amount of work. It walks the whole iterator and returns at the first thing it can do, so one Advance produces at most one request. Along the way it marks any node whose request deadline has passed as unresponsive, which frees a concurrency slot. QueryConfig.Concurrency bounds how many requests may be in flight at once.

QueryConfig.RequestTimeout bounds a single request, and the query enforces it by marking the node unresponsive. QueryConfig.Timeout bounds the whole query and is set when the first node is contacted. Once it passes while the query is still running, the query stops and reports StateQueryTimeout, treating its in-flight requests as failed.

A finished query is sticky. Once it has settled on its result, every later Advance returns the same StateQueryFinished carrying the same nodes. A query that stops on its deadline is sticky in the same way, returning the same StateQueryTimeout.

Pool

A Pool runs many queries at once. The caller adds a query with an event, and the pool advances one query per Advance, reporting that query's state as a pool state. PoolConfig.Concurrency bounds how many queries may be waiting for responses at any one time; PoolConfig.QueryConcurrency is passed through to each query as its own request concurrency. A query that passes its deadline is removed and reported as StatePoolQueryTimeout.

Scheduling

The waiting states carry a NextDue time: the earliest instant at which advancing again could make progress without a response arriving, which is when the next request deadline falls due. A caller with nothing else to do can sleep until NextDue rather than poll.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type ClosestNodesIter 

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

A ClosestNodesIter iterates nodes in order of ascending distance from a key.

func NewClosestNodesIter 

func NewClosestNodesIter[K kad.Key[K], N kad.NodeID[K]](target K) *ClosestNodesIter[K, N]

NewClosestNodesIter creates a new ClosestNodesIter

func (*ClosestNodesIter[K, N]) Add 

func (iter *ClosestNodesIter[K, N]) Add(ni *NodeStatus[K, N])

func (*ClosestNodesIter[K, N]) Each 

func (iter *ClosestNodesIter[K, N]) Each(ctx context.Context, fn func(context.Context, *NodeStatus[K, N]) bool) bool

func (*ClosestNodesIter[K, N]) Find 

func (iter *ClosestNodesIter[K, N]) Find(k K) (*NodeStatus[K, N], bool)

type EventPoolAddFindCloserQuery 

type EventPoolAddFindCloserQuery[K kad.Key[K], N kad.NodeID[K]] struct {
	ActivityID coordt.ActivityID // the id to use for the new query
	Target     K                 // the target key for the query
	Seed       []N               // an initial set of close nodes the query should use
	NumResults int               // the minimum number of nodes to successfully contact before considering iteration complete
}

EventPoolAddQuery is an event that attempts to add a new query that finds closer nodes to a target key.

type EventPoolAddQuery 

type EventPoolAddQuery[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	ActivityID coordt.ActivityID // the id to use for the new query
	Target     K                 // the target key for the query
	Message    M                 // message to be sent to each node
	Seed       []N               // an initial set of close nodes the query should use
	NumResults int               // the minimum number of nodes to successfully contact before considering iteration complete
}

EventPoolAddQuery is an event that attempts to add a new query that sends a message.

type EventPoolNodeFailure 

type EventPoolNodeFailure[K kad.Key[K], N kad.NodeID[K]] struct {
	ActivityID coordt.ActivityID // the id of the query that sent the message
	NodeID     N                 // the node the message was sent to
	Error      error             // the error that caused the failure, if any
}

EventPoolNodeFailure notifies a Pool that an attempt to contact a node has failed.

type EventPoolNodeResponse 

type EventPoolNodeResponse[K kad.Key[K], N kad.NodeID[K]] struct {
	ActivityID  coordt.ActivityID // the id of the query that sent the message
	NodeID      N                 // the node the message was sent to
	CloserNodes []N               // the closer nodes sent by the node
}

EventPoolNodeResponse notifies a Pool that an attempt to contact a node has received a successful response.

type EventPoolPoll 

type EventPoolPoll struct{}

EventPoolPoll is an event that signals the pool that it can perform housekeeping work such as time out queries.

type EventPoolStopQuery 

type EventPoolStopQuery struct {
	ActivityID coordt.ActivityID // the id of the query that should be stopped
}

EventPoolStopQuery notifies a Pool to stop a query.

type EventQueryCancel 

type EventQueryCancel struct{}

EventQueryMessageResponse notifies a query to stop all work and enter the finished state.

type EventQueryNodeFailure 

type EventQueryNodeFailure[K kad.Key[K], N kad.NodeID[K]] struct {
	NodeID N     // the node the message was sent to
	Error  error // the error that caused the failure, if any
}

EventQueryNodeFailure notifies a Query that an attempt to to contact a node has failed.

type EventQueryNodeResponse 

type EventQueryNodeResponse[K kad.Key[K], N kad.NodeID[K]] struct {
	NodeID      N   // the node the message was sent to
	CloserNodes []N // the closer nodes sent by the node
}

EventQueryNodeResponse notifies a Query that an attempt to contact a node has received a successful response.

type EventQueryPoll 

type EventQueryPoll struct{}

EventQueryPoll is an event that signals a Query that it can perform housekeeping work.

type NodeIter 

type NodeIter[K kad.Key[K], N kad.NodeID[K]] interface {
	// Add adds node information to the iterator
	Add(*NodeStatus[K, N])

	// Find returns the node information corresponding to the given Kademlia key
	Find(K) (*NodeStatus[K, N], bool)

	// Each applies fn to each entry in the iterator in order. Each stops and returns true if fn returns true.
	// Otherwise, Each returns false when there are no further entries.
	Each(ctx context.Context, fn func(context.Context, *NodeStatus[K, N]) bool) bool
}

A NodeIter iterates nodes according to some strategy.

type NodeState 

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

type NodeStatus 

type NodeStatus[K kad.Key[K], N kad.NodeID[K]] struct {
	NodeID N
	State  NodeState
}

type Pool 

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

func NewPool 

func NewPool[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]](self N, cfg *PoolConfig) (*Pool[K, N, M], error)

func (*Pool[K, N, M]) Advance 

func (p *Pool[K, N, M]) Advance(ctx context.Context, now time.Time, ev PoolEvent) PoolState

Advance advances the state of the pool by attempting to advance one of its queries

func (*Pool[K, N, M]) Stats added in v0.0.10 

func (p *Pool[K, N, M]) Stats(activityID coordt.ActivityID) (QueryStats, bool)

Stats returns the stats accumulated so far by the query with the given id. The bool result is false if no query with that id is running in the pool.

type PoolConfig 

type PoolConfig struct {
	Concurrency      int           // the maximum number of queries that may be waiting for message responses at any one time
	Timeout          time.Duration // the time to wait before terminating a query that is not making progress
	Replication      int           // the 'k' parameter defined by Kademlia
	QueryConcurrency int           // the maximum number of concurrent requests that each query may have in flight
	RequestTimeout   time.Duration // the timeout queries should use for contacting a single node

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

PoolConfig specifies optional configuration for a Pool

func DefaultPoolConfig 

func DefaultPoolConfig() *PoolConfig

DefaultPoolConfig returns the default configuration options for a Pool. Options may be overridden before passing to NewPool

func (*PoolConfig) Validate 

func (cfg *PoolConfig) Validate() error

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

type PoolEvent 

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

PoolEvent is an event intended to advance the state of a pool.

type PoolState 

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

type Query 

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

A Query is one iterative Kademlia lookup.

It contacts nodes in ascending order of distance from a target key. Each response supplies further nodes, so the query walks towards the target until the closest nodes it knows of stop improving. The ordering is decided by the NodeIter the query is given rather than by the query itself.

A query makes two independent choices. It either asks each node for the nodes it holds closest to the target, emitting StateQueryFindCloser, or sends each node a message and takes the closer nodes from the reply, emitting StateQuerySendMessage. Separately, it either stops at the QueryConfig.NumResults closest nodes or, for a find-closer query, enumerates every node inside the region named by a prefix of the target. NewFindCloserQuery and NewQuery create the two request kinds, both stopping at the closest nodes; NewCoverageQuery creates a find-closer query that enumerates a region instead.

Every advance walks the whole iterator and returns at the first thing it can do, so one advance produces at most one request. Along the way it marks any node whose request deadline has passed as unresponsive, which frees a concurrency slot. The first node that has not been contacted is then sent a request if a slot is free, and that instruction is what the advance returns. The walk stops early if every slot is already in use.

A closest-nodes query is finished when, walking outward, it reaches a node that responded successfully having already counted QueryConfig.NumResults successes with no request outstanding nearer the target. A coverage query is instead finished when the walk reaches a node outside the region with nothing nearer still in flight, having either found a region member or contacted QueryConfig.NumResults nodes without one. Either kind also finishes when the walk reaches the end with nothing in flight and nothing left to contact, or when it is cancelled by EventQueryCancel. Finishing is sticky, since the finished flag is tested before the event, so every later advance returns the same StateQueryFinished carrying the same nodes.

Two deadlines apply and the query enforces both. QueryConfig.RequestTimeout bounds a single request, which the query enforces by marking the node unresponsive. QueryConfig.Timeout bounds the whole query and is set when the first node is contacted; once it passes while the query is still running, the query stops and reports StateQueryTimeout, treating its in-flight requests as failed. Timing out is sticky in the same way finishing is.

The node the query is running on is excluded throughout, both from the seed set and from the closer nodes carried by any response.

func NewCoverageQuery added in v0.0.11 

func NewCoverageQuery[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]](self N, id coordt.ActivityID, target K, prefixLen int, iter NodeIter[K, N], knownClosestNodes []N, cfg *QueryConfig) (*Query[K, N, M], error)

NewCoverageQuery creates a query that finds every node inside the region named by the first prefixLen bits of target. Like a find-closer query it asks each node it contacts for the nodes closest to target, but instead of stopping at the closest NumResults nodes it walks in to the region and continues until every node it has heard of inside the region has been contacted, then reports them all.

target is any key inside the region. QueryConfig.NumResults is the walk-in bound rather than a result cap: if the query contacts that many nodes without entering the region it concludes the region is empty. The result is never truncated to it.

func NewFindCloserQuery 

func NewFindCloserQuery[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]](self N, id coordt.ActivityID, target K, iter NodeIter[K, N], knownClosestNodes []N, cfg *QueryConfig) (*Query[K, N, M], error)

NewFindCloserQuery creates a query that asks each node it contacts for the nodes that node holds closest to target, and takes those as the nodes to contact next. It sends no message of its own, so the message type parameter goes unused, and it reports the node it wants contacted by emitting StateQueryFindCloser.

The query is seeded with knownClosestNodes, from which self is excluded, and orders every node it learns of using iter. It reports its progress under the query id id. A nil cfg uses DefaultQueryConfig, and a non-nil one is validated.

func NewQuery 

func NewQuery[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]](self N, id coordt.ActivityID, target K, msg M, iter NodeIter[K, N], knownClosestNodes []N, cfg *QueryConfig) (*Query[K, N, M], error)

NewQuery creates a query that sends msg to each node it contacts and takes the nodes closer to target from each reply, so the reply serves both as the answer to the message and as the source of the next nodes to walk to. It reports the node it wants contacted by emitting StateQuerySendMessage.

The query is seeded with knownClosestNodes, from which self is excluded, and orders every node it learns of using iter. It reports its progress under the query id id. A nil cfg uses DefaultQueryConfig, and a non-nil one is validated.

func (*Query[K, N, M]) Advance 

func (q *Query[K, N, M]) Advance(ctx context.Context, now time.Time, ev QueryEvent) (out QueryState)

type QueryConfig 

type QueryConfig struct {
	Concurrency    int           // the maximum number of concurrent requests that may be in flight
	NumResults     int           // for a closest-nodes query, the number of nodes to contact successfully before completing and the most it returns; for a coverage query, the number of nodes to contact without finding a region member before concluding the region is empty
	RequestTimeout time.Duration // the timeout for contacting a single node
	Timeout        time.Duration // the time to wait before the query is considered to have stopped making progress

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

QueryConfig specifies optional configuration for a Query

func DefaultQueryConfig 

func DefaultQueryConfig() *QueryConfig

DefaultQueryConfig returns the default configuration options for a Query. Options may be overridden before passing to NewQuery

func (*QueryConfig) Validate 

func (cfg *QueryConfig) Validate() error

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

type QueryEvent 

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

type QueryState 

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

type QueryStats 

type QueryStats struct {
	Start    time.Time // the time the first request was dispatched, zero until then
	End      time.Time // the time the query finished, zero until then
	Requests int       // the number of requests dispatched
	Success  int       // the number of requests answered within their deadline
	Failure  int       // the number of requests that errored or passed their deadline
}

QueryStats holds the counts and timings a Query accumulates as it runs. A query reports them with every state it emits.

The counters track requests rather than nodes, and each request that has completed counts once, so Success and Failure together never exceed Requests.

type SequentialIter 

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

A SequentialIter iterates nodes in the order they were added to the iterator.

func NewSequentialIter 

func NewSequentialIter[K kad.Key[K], N kad.NodeID[K]]() *SequentialIter[K, N]

NewSequentialIter creates a new SequentialIter

func (*SequentialIter[K, N]) Add 

func (iter *SequentialIter[K, N]) Add(ni *NodeStatus[K, N])

func (*SequentialIter[K, N]) Each 

func (iter *SequentialIter[K, N]) Each(ctx context.Context, fn func(context.Context, *NodeStatus[K, N]) bool) bool

func (*SequentialIter[K, N]) Find 

func (iter *SequentialIter[K, N]) Find(k K) (*NodeStatus[K, N], bool)

Find returns the node information corresponding to the given Kademlia key. It uses a linear search which makes it unsuitable for large numbers of entries.

type StateNodeFailed 

type StateNodeFailed struct{}

StateNodeFailed indicates that the attempt to contact the node failed.

type StateNodeNotContacted 

type StateNodeNotContacted struct{}

StateNodeNotContacted indicates that the node has not been contacted yet.

type StateNodeSucceeded 

type StateNodeSucceeded struct{}

StateNodeSucceeded indicates that the attempt to contact the node succeeded.

type StateNodeUnresponsive 

type StateNodeUnresponsive struct{}

StateNodeUnresponsive indicates that the node did not respond within the configured timeout.

type StateNodeWaiting 

type StateNodeWaiting struct {
	Deadline time.Time
}

StateNodeWaiting indicates that a query is waiting for a response from the node.

type StatePoolFindCloser 

type StatePoolFindCloser[K kad.Key[K], N kad.NodeID[K]] struct {
	ActivityID coordt.ActivityID
	Target     K         // the key that the query wants to find closer nodes for
	NodeID     N         // the node to send the message to
	Deadline   time.Time // the time by which the request should complete
	Stats      QueryStats
}

StatePoolFindCloser indicates that a pool query wants to send a find closer nodes message to a node.

type StatePoolIdle 

type StatePoolIdle struct {
	NextDue time.Time // the earliest time advancing the pool could make progress, zero if there is none
}

StatePoolIdle indicates that the pool is idle, i.e. there are no queries to process.

type StatePoolQueryFinished 

type StatePoolQueryFinished[K kad.Key[K], N kad.NodeID[K]] struct {
	ActivityID   coordt.ActivityID
	Stats        QueryStats
	Target       K // the key the query was looking for the closest nodes to
	ClosestNodes []N
}

StatePoolQueryFinished indicates that a query has finished.

type StatePoolQueryTimeout 

type StatePoolQueryTimeout struct {
	ActivityID coordt.ActivityID
	Stats      QueryStats
}

StatePoolQueryTimeout indicates that a query has timed out.

type StatePoolSendMessage 

type StatePoolSendMessage[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	ActivityID coordt.ActivityID
	NodeID     N // the node to send the message to
	Message    M
	Deadline   time.Time // the time by which the request should complete
	Stats      QueryStats
}

StatePoolSendMessage indicates that a pool query wants to send a message to a node.

type StatePoolWaitingAtCapacity 

type StatePoolWaitingAtCapacity struct {
	NextDue time.Time // the earliest time advancing the pool could make progress, zero if there is none
}

StatePoolWaitingAtCapacity indicates that at least one query is waiting for results and the pool has reached its maximum number of concurrent queries.

type StatePoolWaitingWithCapacity 

type StatePoolWaitingWithCapacity struct {
	NextDue time.Time // the earliest time advancing the pool could make progress, zero if there is none
}

StatePoolWaitingWithCapacity indicates that at least one query is waiting for results but capacity to start more is available.

type StateQueryFindCloser 

type StateQueryFindCloser[K kad.Key[K], N kad.NodeID[K]] struct {
	ActivityID coordt.ActivityID
	Target     K         // the key that the query wants to find closer nodes for
	NodeID     N         // the node to send the message to
	Deadline   time.Time // the time by which the request should complete
	Stats      QueryStats
}

StateQueryFindCloser indicates that the Query wants to send a find closer nodes message to a node.

type StateQueryFinished 

type StateQueryFinished[K kad.Key[K], N kad.NodeID[K]] struct {
	ActivityID   coordt.ActivityID
	Stats        QueryStats
	Target       K   // the key the query was looking for the closest nodes to
	ClosestNodes []N // the nodes the query settled on: the closest to the target, or a coverage query's region members
}

StateQueryFinished indicates that the Query has finished.

type StateQuerySendMessage 

type StateQuerySendMessage[K kad.Key[K], N kad.NodeID[K], M coordt.Message[K, N]] struct {
	ActivityID coordt.ActivityID
	NodeID     N // the node to send the message to
	Message    M
	Deadline   time.Time // the time by which the request should complete
	Stats      QueryStats
}

StateQuerySendMessage indicates that the Query wants to send a message to a node.

type StateQueryTimeout added in v0.0.13 

type StateQueryTimeout struct {
	ActivityID coordt.ActivityID
	Stats      QueryStats
}

StateQueryTimeout indicates that the Query stopped because it passed its own deadline while still running, so its in-flight requests are treated as failed and their errors go unreported.

type StateQueryWaitingAtCapacity 

type StateQueryWaitingAtCapacity struct {
	ActivityID coordt.ActivityID
	Stats      QueryStats
	Deadline   time.Time // the time by which the query must have completed
	NextDue    time.Time // the earliest time advancing the query could make progress, zero if there is none
}

StateQueryWaitingAtCapacity indicates that the Query is waiting for results and is at capacity.

type StateQueryWaitingWithCapacity 

type StateQueryWaitingWithCapacity struct {
	ActivityID coordt.ActivityID
	Stats      QueryStats
	Deadline   time.Time // the time by which the query must have completed
	NextDue    time.Time // the earliest time advancing the query could make progress, zero if there is none
}

StateQueryWaitingWithCapacity indicates that the Query is waiting for results but has no further nodes to contact.

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.