docs: code comments

Author: dennis-tra

pkg/core/core.go

@@ -1,6 +1,7 @@
 package core
 
 import (
+	"context"
 	"time"
 
 	"github.com/libp2p/go-libp2p/core/peer"
@@ -54,6 +55,14 @@ type Driver[I PeerInfo[I], R WorkResult[I]] interface {
 	Close()
 }
 
+// A Worker processes tasks of type T and returns results of type R or an error.
+// Workers are used to process a single peer or store a crawl result to the
+// database. It is the unit of concurrency in this system.
+type Worker[T any, R any] interface {
+	// Work instructs the Worker to process the task and produce a result.
+	Work(ctx context.Context, task T) (R, error)
+}
+
 // Handler defines the interface that the engine will call every time
 // it has received a result from any of its workers.
 type Handler[I PeerInfo[I], R WorkResult[I]] interface {
@@ -94,6 +103,13 @@ type WorkResult[I PeerInfo[I]] interface {
 	LogEntry() *log.Entry
 }
 
+// Result is a generic result object. It captures a generic value or any
+// error that might have occurred when producing this result.
+type Result[R any] struct {
+	Value R
+	Error error
+}
+
 // WriteResult must be returned by write workers.
 type WriteResult struct {
 	*db.InsertVisitResult

pkg/core/engine.go

@@ -368,7 +368,10 @@ func (e *Engine[I, R]) handlePeerResult(result Result[R]) {
 		}
 
 		// If we don't know any multi addresses for the peer yet, we push it
-		// to the end of our priority queue by giving it a low priority.
+		// to the end of our priority queue by giving it a low priority. If we
+		// find that peer again in another routing table, we might find another
+		// multi address. In that case, we update the set of addresses and
+		// increase the priority.
 		priority := 1
 		if len(e.maddrFilter(task.Addrs())) == 0 {
 			priority = 0

pkg/core/queue.go

@@ -4,19 +4,20 @@ import (
 	"container/heap"
 )
 
-// An item is something we manage in a priority queue.
-type item[T any] struct {
-	value    T // The value of the item; arbitrary.
-	key      string
-	priority int // The priority of the item in the queue.
-	index    int // The index of the item in the heap.
-}
-
+// PriorityQueue is can take unique items and pops them according to their priority.
 type PriorityQueue[T any] struct {
 	queue  *priorityQueue[T]
 	lookup map[string]*item[T]
 }
 
+// An item is something we manage in a priority queue.
+type item[T any] struct {
+	value    T      // The value of the item; arbitrary.
+	key      string // The key belonging to the value
+	priority int    // The priority of the item in the queue.
+	index    int    // The index of the item in the heap.
+}
+
 func NewPriorityQueue[T any]() *PriorityQueue[T] {
 	queue := make(priorityQueue[T], 0)
 	return &PriorityQueue[T]{

pkg/core/worker.go

@@ -5,10 +5,7 @@ import (
 	"sync"
 )
 
-type Worker[T any, R any] interface {
-	Work(ctx context.Context, task T) (R, error)
-}
-
+// A Pool manages a set of Workers
 type Pool[T any, R any] struct {
 	start   sync.Once
 	results chan Result[R]
@@ -22,6 +19,10 @@ func NewPool[T any, R any](workers ...Worker[T, R]) *Pool[T, R] {
 	}
 }
 
+// Start takes a channel on which tasks will be scheduled. It is guaranteed that
+// the Pool reads from this channel as fast as possible. To stop the worker pool
+// you need to close the channel. To wait until all Workers have finished, wait
+// until the results channel returned from this method was closed as well.
 func (w *Pool[T, R]) Start(ctx context.Context, tasks <-chan T) <-chan Result[R] {
 	w.start.Do(func() {
 		var wg sync.WaitGroup
@@ -52,8 +53,3 @@ func (w *Pool[T, R]) Start(ctx context.Context, tasks <-chan T) <-chan Result[R]
 func (w *Pool[T, R]) Size() int {
 	return len(w.workers)
 }
-
-type Result[R any] struct {
-	Value R
-	Error error
-}

pkg/libp2p/crawler_p2p.go

@@ -74,7 +74,7 @@ func (c *Crawler) crawlP2P(ctx context.Context, pi PeerInfo) <-chan P2PResult {
 				result.CrawlErrorStr = db.NetError(result.CrawlError)
 			}
 
-			// wait for the Identify exchange to complete
+			// wait for the Identify exchange to complete (no-op if already done)
 			c.identifyWait(ctx, pi.AddrInfo)
 
 			// Extract information from peer store
@@ -189,7 +189,8 @@ func (c *Crawler) fetchNeighbors(ctx context.Context, pi peer.AddrInfo) (*core.R
 				// This error happens in: https://github.com/libp2p/go-libp2p/blob/4e2a16dd3f4f980bf9429572b3d2aed885594ec4/p2p/host/basic/basic_host.go#L645
 				if err.Error() == "connection failed" {
 					// This means we were connected to the peer, tried to open
-					// a stream but then failed to do so.
+					// a stream but then failed to do so. Try to reconnect as
+					// the peer appears to be online
 
 					select {
 					case <-ctx.Done():

pkg/utils/utils.go

@@ -92,7 +92,8 @@ func FilterPublicMaddrs(maddrs []ma.Multiaddr) []ma.Multiaddr {
 	return filtered
 }
 
-// MergeMaddrs strips private multiaddrs from the given peer address information.
+// MergeMaddrs takes two slices of multi addresses and merges them into a single
+// one.
 func MergeMaddrs(maddrSet1 []ma.Multiaddr, maddrSet2 []ma.Multiaddr) []ma.Multiaddr {
 	maddrSetOut := make(map[string]ma.Multiaddr, len(maddrSet1))
 	for _, maddr := range maddrSet1 {