routing/http/client: add support for GetClosestPeers

As the server part, this is heavily inspired in FindPeers.

Author: hsanjuan

routing/http/client/client.go

@@ -9,8 +9,10 @@ import (
 	"io"
 	"mime"
 	"net/http"
+	"net/url"
 	gourl "net/url"
 	"slices"
+	"strconv"
 	"strings"
 	"time"
 
@@ -583,3 +585,91 @@ func (c *Client) PutIPNS(ctx context.Context, name ipns.Name, record *ipns.Recor
 
 	return nil
 }
+
+// GetClosestPeers obtains the closest peers to the given peer ID.
+func (c *Client) GetClosestPeers(ctx context.Context, peerID, closerThan peer.ID, count int) (peers iter.ResultIter[*types.PeerRecord], err error) {
+	m := newMeasurement("GetClosestPeers")
+
+	// Build the base URL path
+	u, err := gourl.JoinPath(c.baseURL, "routing/v1/dht/closest/peers", peer.ToCid(peerID).String())
+	if err != nil {
+		return nil, err
+	}
+
+	// Add query parameters
+	queryParams := make(url.Values)
+	if closerThan != "" {
+		queryParams.Set("closerThan", closerThan.String())
+	}
+	if count > 0 {
+		queryParams.Set("count", strconv.Itoa(count))
+	}
+	u += "?" + queryParams.Encode()
+
+	// Create the HTTP request
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Accept", c.accepts)
+
+	m.host = req.Host
+	start := c.clock.Now()
+	resp, err := c.httpClient.Do(req)
+	m.latency = c.clock.Since(start)
+	m.err = err
+
+	if err != nil {
+		m.record(ctx)
+		return nil, err
+	}
+
+	m.statusCode = resp.StatusCode
+	if resp.StatusCode == http.StatusNotFound {
+		resp.Body.Close()
+		m.record(ctx)
+		return iter.FromSlice[iter.Result[*types.PeerRecord]](nil), nil
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		err := httpError(resp.StatusCode, resp.Body)
+		resp.Body.Close()
+		m.record(ctx)
+		return nil, err
+	}
+
+	respContentType := resp.Header.Get("Content-Type")
+	mediaType, _, err := mime.ParseMediaType(respContentType)
+	if err != nil {
+		resp.Body.Close()
+		m.err = err
+		m.record(ctx)
+		return nil, fmt.Errorf("parsing Content-Type: %w", err)
+	}
+
+	m.mediaType = mediaType
+
+	var skipBodyClose bool
+	defer func() {
+		if !skipBodyClose {
+			resp.Body.Close()
+		}
+	}()
+
+	var it iter.ResultIter[*types.PeerRecord]
+	switch mediaType {
+	case mediaTypeJSON:
+		parsedResp := &jsontypes.PeersResponse{}
+		err = json.NewDecoder(resp.Body).Decode(parsedResp)
+		var sliceIt iter.Iter[*types.PeerRecord] = iter.FromSlice(parsedResp.Peers)
+		it = iter.ToResultIter(sliceIt)
+	case mediaTypeNDJSON:
+		skipBodyClose = true
+		it = ndjson.NewPeerRecordsIter(resp.Body)
+	default:
+		logger.Errorw("unknown media type", "MediaType", mediaType, "ContentType", respContentType)
+		return nil, errors.New("unknown content type")
+	}
+
+	return &measuringIter[iter.Result[*types.PeerRecord]]{Iter: it, ctx: ctx, m: m}, nil
+}

routing/http/client/client_test.go

@@ -47,6 +47,11 @@ func (m *mockContentRouter) FindPeers(ctx context.Context, pid peer.ID, limit in
 	return args.Get(0).(iter.ResultIter[*types.PeerRecord]), args.Error(1)
 }
 
+func (m *mockContentRouter) GetClosestPeers(ctx context.Context, peerID, closerThan peer.ID, count int) (iter.ResultIter[*types.PeerRecord], error) {
+	args := m.Called(ctx, peerID, closerThan, count)
+	return args.Get(0).(iter.ResultIter[*types.PeerRecord]), args.Error(1)
+}
+
 func (m *mockContentRouter) GetIPNS(ctx context.Context, name ipns.Name) (*ipns.Record, error) {
 	args := m.Called(ctx, name)
 	rec, _ := args.Get(0).(*ipns.Record)
@@ -676,6 +681,146 @@ func TestClient_FindPeers(t *testing.T) {
 	}
 }
 
+func TestClient_GetClosestPeers(t *testing.T) {
+	bitswapPeerRecord := makePeerRecord([]string{"transport-bitswap"})
+	httpPeerRecord := makePeerRecord([]string{"transport-ipfs-gateway-http"})
+
+	peerRecords := []iter.Result[*types.PeerRecord]{
+		{Val: &bitswapPeerRecord},
+		{Val: &httpPeerRecord},
+	}
+
+	pid := *bitswapPeerRecord.ID
+
+	cases := []struct {
+		name                    string
+		httpStatusCode          int
+		stopServer              bool
+		routerResult            []iter.Result[*types.PeerRecord]
+		routerErr               error
+		clientRequiresStreaming bool
+		serverStreamingDisabled bool
+		closerThan              peer.ID
+		count                   int
+
+		expErrContains       osErrContains
+		expResult            []iter.Result[*types.PeerRecord]
+		expStreamingResponse bool
+		expJSONResponse      bool
+	}{
+		{
+			name:                 "happy case",
+			routerResult:         peerRecords,
+			expResult:            peerRecords,
+			expStreamingResponse: true,
+		},
+		{
+			name:                    "server doesn't support streaming",
+			routerResult:            peerRecords,
+			expResult:               peerRecords,
+			serverStreamingDisabled: true,
+			expJSONResponse:         true,
+		},
+		{
+			name:                    "client requires streaming but server doesn't support it",
+			serverStreamingDisabled: true,
+			clientRequiresStreaming: true,
+			expErrContains:          osErrContains{expContains: "HTTP error with StatusCode=400: no supported content types"},
+		},
+		{
+			name:           "returns an error if there's a non-200 response",
+			httpStatusCode: 500,
+			expErrContains: osErrContains{expContains: "HTTP error with StatusCode=500"},
+		},
+		{
+			name:       "returns an error if the HTTP client returns a non-HTTP error",
+			stopServer: true,
+			expErrContains: osErrContains{
+				expContains:    "connect: connection refused",
+				expContainsWin: "connectex: No connection could be made because the target machine actively refused it.",
+			},
+		},
+		{
+			name:           "returns no providers if the HTTP server returns a 404 response",
+			httpStatusCode: 404,
+			expResult:      nil,
+		},
+		{
+			name:                 "passes count and closerThan along",
+			expStreamingResponse: true,
+			routerResult:         peerRecords,
+			expResult:            peerRecords,
+			count:                10,
+			closerThan:           pid,
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			var clientOpts []Option
+			var serverOpts []server.Option
+			var onRespReceived []func(*http.Response)
+			var onReqReceived []func(*http.Request)
+
+			if c.serverStreamingDisabled {
+				serverOpts = append(serverOpts, server.WithStreamingResultsDisabled())
+			}
+
+			if c.clientRequiresStreaming {
+				clientOpts = append(clientOpts, WithStreamResultsRequired())
+				onReqReceived = append(onReqReceived, func(r *http.Request) {
+					assert.Equal(t, mediaTypeNDJSON, r.Header.Get("Accept"))
+				})
+			}
+
+			if c.expStreamingResponse {
+				onRespReceived = append(onRespReceived, func(r *http.Response) {
+					assert.Equal(t, mediaTypeNDJSON, r.Header.Get("Content-Type"))
+				})
+			}
+
+			if c.expJSONResponse {
+				onRespReceived = append(onRespReceived, func(r *http.Response) {
+					assert.Equal(t, mediaTypeJSON, r.Header.Get("Content-Type"))
+				})
+			}
+
+			deps := makeTestDeps(t, clientOpts, serverOpts)
+
+			deps.recordingHTTPClient.f = append(deps.recordingHTTPClient.f, onRespReceived...)
+			deps.recordingHandler.f = append(deps.recordingHandler.f, onReqReceived...)
+
+			client := deps.client
+			router := deps.router
+
+			ctx, cancel := context.WithCancel(context.Background())
+			t.Cleanup(cancel)
+
+			if c.httpStatusCode != 0 {
+				deps.server.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					w.WriteHeader(c.httpStatusCode)
+				})
+			}
+
+			if c.stopServer {
+				deps.server.Close()
+			}
+
+			count := c.count
+			if count == 0 {
+				count = 20
+			}
+			routerResultIter := iter.FromSlice(c.routerResult)
+			router.On("GetClosestPeers", mock.Anything, pid, c.closerThan, count).Return(routerResultIter, c.routerErr)
+
+			resultIter, err := client.GetClosestPeers(ctx, pid, c.closerThan, c.count)
+			c.expErrContains.errContains(t, err)
+
+			results := iter.ReadAll(resultIter)
+			assert.Equal(t, c.expResult, results)
+		})
+	}
+}
+
 func TestNormalizeBaseURL(t *testing.T) {
 	cases := []struct {
 		name        string

routing/http/contentrouter/contentrouter.go

@@ -23,6 +23,7 @@ var logger = logging.Logger("routing/http/contentrouter")
 
 const ttl = 24 * time.Hour
 
+// A Client provides HTTP Content Routing methods. See also [server.ContentRouter].
 type Client interface {
 	FindProviders(ctx context.Context, key cid.Cid) (iter.ResultIter[types.Record], error)
 	ProvideBitswap(ctx context.Context, keys []cid.Cid, ttl time.Duration) (time.Duration, error)
@@ -59,6 +60,8 @@ func WithMaxProvideBatchSize(max int) option {
 	}
 }
 
+// NewContentRoutingClient returns a client that conforms to the
+// ContentRouting interfaces.
 func NewContentRoutingClient(c Client, opts ...option) *contentRouter {
 	cr := &contentRouter{
 		client:                c,