Release v0.21.0

.github/workflows/gateway-conformance.yml

@@ -22,7 +22,7 @@ jobs:
     steps:
       # 1. Download the gateway-conformance fixtures
       - name: Download gateway-conformance fixtures
-        uses: ipfs/gateway-conformance/.github/actions/extract-fixtures@v0.5
+        uses: ipfs/gateway-conformance/.github/actions/extract-fixtures@v0.6
         with:
           output: fixtures
           merged: true
@@ -47,15 +47,15 @@ jobs:
 
       # 4. Run the gateway-conformance tests
       - name: Run gateway-conformance tests
-        uses: ipfs/gateway-conformance/.github/actions/test@v0.5
+        uses: ipfs/gateway-conformance/.github/actions/test@v0.6
         with:
           gateway-url: http://127.0.0.1:8040
+          subdomain-url: http://example.net:8040
+          specs: -trustless-ipns-gateway,-path-ipns-gateway,-subdomain-ipns-gateway,-dnslink-gateway
           json: output.json
           xml: output.xml
           html: output.html
           markdown: output.md
-          subdomain-url: http://example.net
-          specs: -trustless-ipns-gateway,-path-ipns-gateway,-subdomain-ipns-gateway,-dnslink-gateway
 
       # 5. Upload the results
       - name: Upload MD summary
@@ -84,7 +84,7 @@ jobs:
     steps:
       # 1. Download the gateway-conformance fixtures
       - name: Download gateway-conformance fixtures
-        uses: ipfs/gateway-conformance/.github/actions/extract-fixtures@v0.5
+        uses: ipfs/gateway-conformance/.github/actions/extract-fixtures@v0.6
         with:
           output: fixtures
           merged: true
@@ -114,16 +114,16 @@ jobs:
 
       # 4. Run the gateway-conformance tests
       - name: Run gateway-conformance tests
-        uses: ipfs/gateway-conformance/.github/actions/test@v0.5
+        uses: ipfs/gateway-conformance/.github/actions/test@v0.6
         with:
           gateway-url: http://127.0.0.1:8040 # we test gateway that is backed by a remote block gateway
+          subdomain-url: http://example.net:8040
+          specs: -trustless-ipns-gateway,-path-ipns-gateway,-subdomain-ipns-gateway,-dnslink-gateway
+          args: -skip 'TestGatewayCache/.*_for_%2Fipfs%2F_with_only-if-cached_succeeds_when_in_local_datastore'
           json: output.json
           xml: output.xml
           html: output.html
           markdown: output.md
-          subdomain-url: http://example.net
-          specs: -trustless-ipns-gateway,-path-ipns-gateway,-subdomain-ipns-gateway,-dnslink-gateway
-          args: -skip 'TestGatewayCache/.*_for_%2Fipfs%2F_with_only-if-cached_succeeds_when_in_local_datastore'
 
       # 5. Upload the results
       - name: Upload MD summary
@@ -152,7 +152,7 @@ jobs:
     steps:
       # 1. Download the gateway-conformance fixtures
       - name: Download gateway-conformance fixtures
-        uses: ipfs/gateway-conformance/.github/actions/extract-fixtures@v0.5
+        uses: ipfs/gateway-conformance/.github/actions/extract-fixtures@v0.6
         with:
           output: fixtures
           merged: true
@@ -182,16 +182,16 @@ jobs:
 
       # 4. Run the gateway-conformance tests
       - name: Run gateway-conformance tests
-        uses: ipfs/gateway-conformance/.github/actions/test@v0.5
+        uses: ipfs/gateway-conformance/.github/actions/test@v0.6
         with:
           gateway-url: http://127.0.0.1:8040 # we test gateway that is backed by a remote car gateway
+          subdomain-url: http://example.net:8040
+          specs: -trustless-ipns-gateway,-path-ipns-gateway,-subdomain-ipns-gateway,-dnslink-gateway
+          args: -skip 'TestGatewayCache/.*_for_%2Fipfs%2F_with_only-if-cached_succeeds_when_in_local_datastore'
           json: output.json
           xml: output.xml
           html: output.html
           markdown: output.md
-          subdomain-url: http://example.net
-          specs: -trustless-ipns-gateway,-path-ipns-gateway,-subdomain-ipns-gateway,-dnslink-gateway
-          args: -skip 'TestGatewayCache/.*_for_%2Fipfs%2F_with_only-if-cached_succeeds_when_in_local_datastore'
 
       # 5. Upload the results
       - name: Upload MD summary

CHANGELOG.md

@@ -24,6 +24,22 @@ The following emojis are used to highlight certain changes:
 
 ### Security
 
+## [v0.21.0]
+
+### Changed
+
+- `boxo/gateway` is now tested against [gateway-conformance v6](https://github.com/ipfs/gateway-conformance/releases/tag/v0.6.0)
+- `bitswap/client` supports additional tracing 
+
+### Removed
+
+* 🛠 `routing/none` removed `ConstructNilRouting`, if you need this functionality you can use the Null Router from [go-libp2p-routing-helpers](https://github.com/libp2p/go-libp2p-routing-helpers).
+
+### Fixed
+
+- `routing/http`: the `FindPeer` now returns `routing.ErrNotFound` when no addresses are found
+- `routing/http`: the `FindProvidersAsync` no longer causes a goroutine buildup
+
 ## [v0.20.0]
 
 ### Added

bitswap/client/internal/providerquerymanager/providerquerymanager.go

@@ -2,13 +2,15 @@ package providerquerymanager
 
 import (
 	"context"
-	"fmt"
 	"sync"
 	"time"
 
+	"github.com/ipfs/boxo/bitswap/client/internal"
 	"github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
 	peer "github.com/libp2p/go-libp2p/core/peer"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 var log = logging.Logger("bitswap")
@@ -39,7 +41,7 @@ type ProviderQueryNetwork interface {
 }
 
 type providerQueryMessage interface {
-	debugMessage() string
+	debugMessage()
 	handle(pqm *ProviderQueryManager)
 }
 
@@ -61,6 +63,7 @@ type newProvideQueryMessage struct {
 }
 
 type cancelRequestMessage struct {
+	ctx               context.Context
 	incomingProviders chan peer.ID
 	k                 cid.Cid
 }
@@ -121,6 +124,10 @@ func (pqm *ProviderQueryManager) SetFindProviderTimeout(findProviderTimeout time
 func (pqm *ProviderQueryManager) FindProvidersAsync(sessionCtx context.Context, k cid.Cid) <-chan peer.ID {
 	inProgressRequestChan := make(chan inProgressRequest)
 
+	var span trace.Span
+	sessionCtx, span = internal.StartSpan(sessionCtx, "ProviderQueryManager.FindProvidersAsync", trace.WithAttributes(attribute.Stringer("cid", k)))
+	defer span.End()
+
 	select {
 	case pqm.providerQueryMessages <- &newProvideQueryMessage{
 		ctx:                   sessionCtx,
@@ -182,7 +189,7 @@ func (pqm *ProviderQueryManager) receiveProviders(sessionCtx context.Context, k
 				return
 			case <-sessionCtx.Done():
 				if incomingProviders != nil {
-					pqm.cancelProviderRequest(k, incomingProviders)
+					pqm.cancelProviderRequest(sessionCtx, k, incomingProviders)
 				}
 				return
 			case provider, ok := <-incomingProviders:
@@ -199,11 +206,12 @@ func (pqm *ProviderQueryManager) receiveProviders(sessionCtx context.Context, k
 	return returnedProviders
 }
 
-func (pqm *ProviderQueryManager) cancelProviderRequest(k cid.Cid, incomingProviders chan peer.ID) {
+func (pqm *ProviderQueryManager) cancelProviderRequest(ctx context.Context, k cid.Cid, incomingProviders chan peer.ID) {
 	cancelMessageChannel := pqm.providerQueryMessages
 	for {
 		select {
 		case cancelMessageChannel <- &cancelRequestMessage{
+			ctx:               ctx,
 			incomingProviders: incomingProviders,
 			k:                 k,
 		}:
@@ -235,17 +243,22 @@ func (pqm *ProviderQueryManager) findProviderWorker() {
 			pqm.timeoutMutex.RLock()
 			findProviderCtx, cancel := context.WithTimeout(fpr.ctx, pqm.findProviderTimeout)
 			pqm.timeoutMutex.RUnlock()
+			span := trace.SpanFromContext(findProviderCtx)
+			span.AddEvent("StartFindProvidersAsync")
 			providers := pqm.network.FindProvidersAsync(findProviderCtx, k, maxProviders)
 			wg := &sync.WaitGroup{}
 			for p := range providers {
 				wg.Add(1)
 				go func(p peer.ID) {
 					defer wg.Done()
+					span.AddEvent("FoundProvider", trace.WithAttributes(attribute.Stringer("peer", p)))
 					err := pqm.network.ConnectTo(findProviderCtx, p)
 					if err != nil {
+						span.RecordError(err, trace.WithAttributes(attribute.Stringer("peer", p)))
 						log.Debugf("failed to connect to provider %s: %s", p, err)
 						return
 					}
+					span.AddEvent("ConnectedToProvider", trace.WithAttributes(attribute.Stringer("peer", p)))
 					select {
 					case pqm.providerQueryMessages <- &receivedProviderMessage{
 						ctx: findProviderCtx,
@@ -326,16 +339,17 @@ func (pqm *ProviderQueryManager) run() {
 	for {
 		select {
 		case nextMessage := <-pqm.providerQueryMessages:
-			log.Debug(nextMessage.debugMessage())
+			nextMessage.debugMessage()
 			nextMessage.handle(pqm)
 		case <-pqm.ctx.Done():
 			return
 		}
 	}
 }
 
-func (rpm *receivedProviderMessage) debugMessage() string {
-	return fmt.Sprintf("Received provider (%s) for cid (%s)", rpm.p.String(), rpm.k.String())
+func (rpm *receivedProviderMessage) debugMessage() {
+	log.Debugf("Received provider (%s) (%s)", rpm.p, rpm.k)
+	trace.SpanFromContext(rpm.ctx).AddEvent("ReceivedProvider", trace.WithAttributes(attribute.Stringer("provider", rpm.p), attribute.Stringer("cid", rpm.k)))
 }
 
 func (rpm *receivedProviderMessage) handle(pqm *ProviderQueryManager) {
@@ -354,8 +368,9 @@ func (rpm *receivedProviderMessage) handle(pqm *ProviderQueryManager) {
 	}
 }
 
-func (fpqm *finishedProviderQueryMessage) debugMessage() string {
-	return "Finished Provider Query on cid: " + fpqm.k.String()
+func (fpqm *finishedProviderQueryMessage) debugMessage() {
+	log.Debugf("Finished Provider Query on cid: %s", fpqm.k)
+	trace.SpanFromContext(fpqm.ctx).AddEvent("FinishedProviderQuery", trace.WithAttributes(attribute.Stringer("cid", fpqm.k)))
 }
 
 func (fpqm *finishedProviderQueryMessage) handle(pqm *ProviderQueryManager) {
@@ -371,21 +386,28 @@ func (fpqm *finishedProviderQueryMessage) handle(pqm *ProviderQueryManager) {
 	requestStatus.cancelFn()
 }
 
-func (npqm *newProvideQueryMessage) debugMessage() string {
-	return "New Provider Query on cid: " + npqm.k.String()
+func (npqm *newProvideQueryMessage) debugMessage() {
+	log.Debugf("New Provider Query on cid: %s", npqm.k)
+	trace.SpanFromContext(npqm.ctx).AddEvent("NewProvideQuery", trace.WithAttributes(attribute.Stringer("cid", npqm.k)))
 }
 
 func (npqm *newProvideQueryMessage) handle(pqm *ProviderQueryManager) {
 	requestStatus, ok := pqm.inProgressRequestStatuses[npqm.k]
 	if !ok {
 
 		ctx, cancelFn := context.WithCancel(pqm.ctx)
+		span := trace.SpanFromContext(npqm.ctx)
+		span.AddEvent("NewQuery", trace.WithAttributes(attribute.Stringer("cid", npqm.k)))
+		ctx = trace.ContextWithSpan(ctx, span)
+
 		requestStatus = &inProgressRequestStatus{
 			listeners: make(map[chan peer.ID]struct{}),
 			ctx:       ctx,
 			cancelFn:  cancelFn,
 		}
+
 		pqm.inProgressRequestStatuses[npqm.k] = requestStatus
+
 		select {
 		case pqm.incomingFindProviderRequests <- &findProviderRequest{
 			k:   npqm.k,
@@ -394,6 +416,8 @@ func (npqm *newProvideQueryMessage) handle(pqm *ProviderQueryManager) {
 		case <-pqm.ctx.Done():
 			return
 		}
+	} else {
+		trace.SpanFromContext(npqm.ctx).AddEvent("JoinQuery", trace.WithAttributes(attribute.Stringer("cid", npqm.k)))
 	}
 	inProgressChan := make(chan peer.ID)
 	requestStatus.listeners[inProgressChan] = struct{}{}
@@ -406,8 +430,9 @@ func (npqm *newProvideQueryMessage) handle(pqm *ProviderQueryManager) {
 	}
 }
 
-func (crm *cancelRequestMessage) debugMessage() string {
-	return "Cancel provider query on cid: " + crm.k.String()
+func (crm *cancelRequestMessage) debugMessage() {
+	log.Debugf("Cancel provider query on cid: %s", crm.k)
+	trace.SpanFromContext(crm.ctx).AddEvent("CancelRequest", trace.WithAttributes(attribute.Stringer("cid", crm.k)))
 }
 
 func (crm *cancelRequestMessage) handle(pqm *ProviderQueryManager) {

bitswap/client/internal/session/session.go

@@ -15,6 +15,7 @@ import (
 	delay "github.com/ipfs/go-ipfs-delay"
 	logging "github.com/ipfs/go-log/v2"
 	peer "github.com/libp2p/go-libp2p/core/peer"
+	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 )
 
@@ -301,35 +302,46 @@ func (s *Session) run(ctx context.Context) {
 
 	s.idleTick = time.NewTimer(s.initialSearchDelay)
 	s.periodicSearchTimer = time.NewTimer(s.periodicSearchDelay.NextWaitTime())
+	sessionSpan := trace.SpanFromContext(ctx)
 	for {
 		select {
 		case oper := <-s.incoming:
 			switch oper.op {
 			case opReceive:
 				// Received blocks
+				sessionSpan.AddEvent("Session.ReceiveOp")
 				s.handleReceive(oper.keys)
 			case opWant:
 				// Client wants blocks
+				sessionSpan.AddEvent("Session.WantOp")
 				s.wantBlocks(ctx, oper.keys)
 			case opCancel:
 				// Wants were cancelled
+				sessionSpan.AddEvent("Session.WantCancelOp")
 				s.sw.CancelPending(oper.keys)
 				s.sws.Cancel(oper.keys)
 			case opWantsSent:
 				// Wants were sent to a peer
+				sessionSpan.AddEvent("Session.WantsSentOp")
 				s.sw.WantsSent(oper.keys)
 			case opBroadcast:
 				// Broadcast want-haves to all peers
-				s.broadcast(ctx, oper.keys)
+				opCtx, span := internal.StartSpan(ctx, "Session.BroadcastOp")
+				s.broadcast(opCtx, oper.keys)
+				span.End()
 			default:
 				panic("unhandled operation")
 			}
 		case <-s.idleTick.C:
 			// The session hasn't received blocks for a while, broadcast
-			s.broadcast(ctx, nil)
+			opCtx, span := internal.StartSpan(ctx, "Session.IdleBroadcast")
+			s.broadcast(opCtx, nil)
+			span.End()
 		case <-s.periodicSearchTimer.C:
 			// Periodically search for a random live want
-			s.handlePeriodicSearch(ctx)
+			opCtx, span := internal.StartSpan(ctx, "Session.PeriodicSearch")
+			s.handlePeriodicSearch(opCtx)
+			span.End()
 		case baseTickDelay := <-s.tickDelayReqs:
 			// Set the base tick delay
 			s.baseTickDelay = baseTickDelay
@@ -392,9 +404,12 @@ func (s *Session) handlePeriodicSearch(ctx context.Context) {
 // providers for the given Cid
 func (s *Session) findMorePeers(ctx context.Context, c cid.Cid) {
 	go func(k cid.Cid) {
+		ctx, span := internal.StartSpan(ctx, "Session.FindMorePeers")
+		defer span.End()
 		for p := range s.providerFinder.FindProvidersAsync(ctx, k) {
 			// When a provider indicates that it has a cid, it's equivalent to
 			// the providing peer sending a HAVE
+			span.AddEvent("FoundPeer")
 			s.sws.Update(p, nil, []cid.Cid{c}, nil)
 		}
 	}(c)

examples/car-file-fetcher/.gitignore

@@ -0,0 +1,3 @@
+car-file-fetcher
+fetcher
+hello.txt

examples/car-file-fetcher/README.md

@@ -0,0 +1,27 @@
+# CAR File Fetcher
+
+This example shows how to download a UnixFS file or directory from a gateway that implements
+[application/vnd.ipld.car](https://www.iana.org/assignments/media-types/application/vnd.ipld.car)
+responses of the [Trustles Gateway](https://specs.ipfs.tech/http-gateways/trustless-gateway/)
+specification, in a trustless, verifiable manner.
+
+It relies on [IPIP-402](https://specs.ipfs.tech/ipips/ipip-0402/) to retrieve
+the file entity via a single CAR request with all blocks required for end-to-end
+verification.
+
+## Build
+
+```bash
+> go build -o fetcher
+```
+
+## Usage
+
+First, you need a gateway that complies with the Trustless Gateway specification.
+In our specific case, we need that the gateway supports CAR response type.
+
+As an example, you can verifiably fetch a `hello.txt` file from IPFS gateway at `https://trustless-gateway.link`:
+
+```
+./fetcher -g https://trustless-gateway.link -o hello.txt /ipfs/bafkreifzjut3te2nhyekklss27nh3k72ysco7y32koao5eei66wof36n5e
+```