From 85347fc1a5fb1331830185b5b2f691f00fc62627 Mon Sep 17 00:00:00 2001
From: dirkmc <dirkmdev@gmail.com>
Date: Wed, 31 Aug 2022 16:26:26 +0200
Subject: [PATCH] feat: transfer limiter web UI updates (#740)

---
 gql/resolver.go                   |  51 +++++++++----
 gql/resolver_transfers.go         |  57 ++++++++++++++-
 gql/schema.graphql                |  18 +++++
 react/src/DealTransfers.css       |  23 ++++++
 react/src/DealTransfers.js        | 117 +++++++++++++++++++++++++++---
 react/src/Deals.css               |   5 ++
 react/src/Deals.js                |  38 ++++++++++
 react/src/gql.js                  |  25 +++++++
 storagemarket/provider.go         |   4 -
 storagemarket/transfer_limiter.go |  63 ++++++++++++++++
 storagemarket/transfers.go        |  67 ++++++++++++-----
 11 files changed, 418 insertions(+), 50 deletions(-)
 create mode 100644 react/src/DealTransfers.css

diff --git a/gql/resolver.go b/gql/resolver.go
index 4884fda73..2cabdbbec 100644
--- a/gql/resolver.go
+++ b/gql/resolver.go
@@ -102,7 +102,7 @@ func (r *resolver) Deal(ctx context.Context, args struct{ ID graphql.ID }) (*dea
 		return nil, err
 	}
 
-	return newDealResolver(deal, r.dealsDB, r.logsDB, r.spApi), nil
+	return newDealResolver(deal, r.provider, r.dealsDB, r.logsDB, r.spApi), nil
 }
 
 type dealsArgs struct {
@@ -135,7 +135,7 @@ func (r *resolver) Deals(ctx context.Context, args dealsArgs) (*dealListResolver
 
 	resolvers := make([]*dealResolver, 0, len(deals))
 	for _, deal := range deals {
-		resolvers = append(resolvers, newDealResolver(&deal, r.dealsDB, r.logsDB, r.spApi))
+		resolvers = append(resolvers, newDealResolver(&deal, r.provider, r.dealsDB, r.logsDB, r.spApi))
 	}
 
 	return &dealListResolver{
@@ -168,7 +168,7 @@ func (r *resolver) DealUpdate(ctx context.Context, args struct{ ID graphql.ID })
 	}
 
 	net := make(chan *dealResolver, 1)
-	net <- newDealResolver(deal, r.dealsDB, r.logsDB, r.spApi)
+	net <- newDealResolver(deal, r.provider, r.dealsDB, r.logsDB, r.spApi)
 
 	// Updates to deal state are broadcast on pubsub. Pipe these updates to the
 	// client
@@ -180,7 +180,7 @@ func (r *resolver) DealUpdate(ctx context.Context, args struct{ ID graphql.ID })
 		}
 		return nil, fmt.Errorf("%s: subscribing to deal updates: %w", args.ID, err)
 	}
-	sub := &subLastUpdate{sub: dealUpdatesSub, dealsDB: r.dealsDB, logsDB: r.logsDB, spApi: r.spApi}
+	sub := &subLastUpdate{sub: dealUpdatesSub, provider: r.provider, dealsDB: r.dealsDB, logsDB: r.logsDB, spApi: r.spApi}
 	go func() {
 		sub.Pipe(ctx, net) // blocks until connection is closed
 		close(net)
@@ -219,7 +219,7 @@ func (r *resolver) DealNew(ctx context.Context) (<-chan *dealNewResolver, error)
 			case evti := <-sub.Out():
 				// Pipe the deal to the new deal channel
 				di := evti.(types.ProviderDealState)
-				rsv := newDealResolver(&di, r.dealsDB, r.logsDB, r.spApi)
+				rsv := newDealResolver(&di, r.provider, r.dealsDB, r.logsDB, r.spApi)
 				totalCount, err := r.dealsDB.Count(ctx, "")
 				if err != nil {
 					log.Errorf("getting total deal count: %w", err)
@@ -330,15 +330,17 @@ func (r *resolver) dealList(ctx context.Context, query string, cursor *graphql.I
 
 type dealResolver struct {
 	types.ProviderDealState
+	provider    *storagemarket.Provider
 	transferred uint64
 	dealsDB     *db.DealsDB
 	logsDB      *db.LogsDB
 	spApi       sealingpipeline.API
 }
 
-func newDealResolver(deal *types.ProviderDealState, dealsDB *db.DealsDB, logsDB *db.LogsDB, spApi sealingpipeline.API) *dealResolver {
+func newDealResolver(deal *types.ProviderDealState, provider *storagemarket.Provider, dealsDB *db.DealsDB, logsDB *db.LogsDB, spApi sealingpipeline.API) *dealResolver {
 	return &dealResolver{
 		ProviderDealState: *deal,
+		provider:          provider,
 		transferred:       uint64(deal.NBytesReceived),
 		dealsDB:           dealsDB,
 		logsDB:            logsDB,
@@ -502,13 +504,17 @@ func (dr *dealResolver) message(ctx context.Context, checkpoint dealcheckpoints.
 		if dr.IsOffline {
 			return "Awaiting Offline Data Import"
 		}
-		switch dr.transferred {
-		case 0:
+		switch {
+		case dr.transferred == 0 && !dr.provider.IsTransferStalled(dr.DealUuid):
 			return "Transfer Queued"
-		case 100:
+		case dr.transferred == 100:
 			return "Transfer Complete"
 		default:
 			pct := (100 * dr.transferred) / dr.ProviderDealState.Transfer.Size
+			isStalled := dr.provider.IsTransferStalled(dr.DealUuid)
+			if isStalled {
+				return fmt.Sprintf("Transfer stalled at %d%% ", pct)
+			}
 			return fmt.Sprintf("Transferring %d%%", pct)
 		}
 	case dealcheckpoints.Transferred:
@@ -533,6 +539,22 @@ func (dr *dealResolver) message(ctx context.Context, checkpoint dealcheckpoints.
 	return checkpoint.String()
 }
 
+func (dr *dealResolver) TransferSamples() []*transferPoint {
+	points := dr.provider.Transfer(dr.DealUuid)
+	pts := make([]*transferPoint, 0, len(points))
+	for _, pt := range points {
+		pts = append(pts, &transferPoint{
+			At:    graphql.Time{Time: pt.At},
+			Bytes: gqltypes.Uint64(pt.Bytes),
+		})
+	}
+	return pts
+}
+
+func (dr *dealResolver) IsTransferStalled() bool {
+	return dr.provider.IsTransferStalled(dr.DealUuid)
+}
+
 func (dr *dealResolver) sealingState(ctx context.Context) string {
 	si, err := dr.spApi.SectorsStatus(ctx, dr.SectorID, false)
 	if err != nil {
@@ -594,10 +616,11 @@ func toUuid(id graphql.ID) (uuid.UUID, error) {
 }
 
 type subLastUpdate struct {
-	sub     event.Subscription
-	dealsDB *db.DealsDB
-	logsDB  *db.LogsDB
-	spApi   sealingpipeline.API
+	sub      event.Subscription
+	provider *storagemarket.Provider
+	dealsDB  *db.DealsDB
+	logsDB   *db.LogsDB
+	spApi    sealingpipeline.API
 }
 
 func (s *subLastUpdate) Pipe(ctx context.Context, net chan *dealResolver) {
@@ -636,7 +659,7 @@ func (s *subLastUpdate) Pipe(ctx context.Context, net chan *dealResolver) {
 	loop:
 		for {
 			di := lastUpdate.(types.ProviderDealState)
-			rsv := newDealResolver(&di, s.dealsDB, s.logsDB, s.spApi)
+			rsv := newDealResolver(&di, s.provider, s.dealsDB, s.logsDB, s.spApi)
 
 			select {
 			case <-ctx.Done():
diff --git a/gql/resolver_transfers.go b/gql/resolver_transfers.go
index a4c7091a3..e7a20f306 100644
--- a/gql/resolver_transfers.go
+++ b/gql/resolver_transfers.go
@@ -6,6 +6,8 @@ import (
 	"time"
 
 	gqltypes "github.com/filecoin-project/boost/gql/types"
+	"github.com/filecoin-project/boost/storagemarket"
+	"github.com/google/uuid"
 	"github.com/graph-gophers/graphql-go"
 )
 
@@ -15,15 +17,62 @@ type transferPoint struct {
 }
 
 // query: transfers: [TransferPoint]
-func (r *resolver) Transfers(_ context.Context) ([]*transferPoint, error) {
-	deals := r.provider.Transfers()
+func (r *resolver) Transfers(_ context.Context) []*transferPoint {
+	return r.getTransferSamples(r.provider.Transfers(), nil)
+}
+
+type transferStats struct {
+	HttpMaxConcurrentDownloads int32
+	Stats                      []*hostTransferStats
+}
+
+type hostTransferStats struct {
+	Host            string
+	Total           int32
+	Started         int32
+	Stalled         int32
+	TransferSamples []*transferPoint
+}
+
+// query: transferStats: TransferStats
+func (r *resolver) TransferStats(_ context.Context) *transferStats {
+	transfersByDeal := r.provider.Transfers()
+	stats := r.provider.TransferStats()
+	gqlStats := make([]*hostTransferStats, 0, len(stats))
+	for _, s := range stats {
+		gqlStats = append(gqlStats, &hostTransferStats{
+			Host:            s.Host,
+			Total:           int32(s.Total),
+			Started:         int32(s.Started),
+			Stalled:         int32(s.Stalled),
+			TransferSamples: r.getTransferSamples(transfersByDeal, s.DealUuids),
+		})
+	}
+	return &transferStats{
+		HttpMaxConcurrentDownloads: int32(r.cfg.Dealmaking.HttpTransferMaxConcurrentDownloads),
+		Stats:                      gqlStats,
+	}
+}
+
+func (r *resolver) getTransferSamples(deals map[uuid.UUID][]storagemarket.TransferPoint, filter []uuid.UUID) []*transferPoint {
+	// If filter is nil, include all deals
+	if filter == nil {
+		for dealUuid := range deals {
+			filter = append(filter, dealUuid)
+		}
+	}
 
 	// We have
 	// dealUUID -> [At: <time>, Transferred: <bytes>, At: <time>, Transferred: <bytes>, ...]
 	// Convert this to
 	// <time> -> <transferred per second>
 	totalAt := make(map[time.Time]uint64)
-	for _, points := range deals {
+	for _, dealUuid := range filter {
+		points, ok := deals[dealUuid]
+		if !ok {
+			continue
+		}
+
 		var prev uint64
 		first := true
 		for _, pt := range points {
@@ -53,5 +102,5 @@ func (r *resolver) Transfers(_ context.Context) ([]*transferPoint, error) {
 		return pts[i].At.Before(pts[j].At.Time)
 	})
 
-	return pts, nil
+	return pts
 }
diff --git a/gql/schema.graphql b/gql/schema.graphql
index 2be280db6..dec85697e 100644
--- a/gql/schema.graphql
+++ b/gql/schema.graphql
@@ -59,6 +59,8 @@ type Deal {
   PublishCid: String!
   IsOffline: Boolean!
   Transfer: TransferParams!
+  TransferSamples: [TransferPoint]!
+  IsTransferStalled: Boolean!
   Checkpoint: String!
   CheckpointAt: Time!
   Err: String!
@@ -287,6 +289,19 @@ type TransferPoint {
   Bytes: Uint64!
 }
 
+type HostStats {
+  Host: String!
+  Total: Int!
+  Started: Int!
+  Stalled: Int!
+  TransferSamples: [TransferPoint]!
+}
+
+type TransferStats {
+  HttpMaxConcurrentDownloads: Int!
+  Stats: [HostStats]!
+}
+
 type MpoolMessage {
   From: String!
   To: String!
@@ -377,6 +392,9 @@ type RootQuery {
   """Get ongoing transfers"""
   transfers: [TransferPoint]!
 
+  """Get stats about queued / active transfers"""
+  transferStats: TransferStats!
+
   """Get local messages in the mpool"""
   mpool(local: Boolean!): [MpoolMessage]!
 
diff --git a/react/src/DealTransfers.css b/react/src/DealTransfers.css
new file mode 100644
index 000000000..e5c7ece5f
--- /dev/null
+++ b/react/src/DealTransfers.css
@@ -0,0 +1,23 @@
+.transfer-stats {
+    margin: 1em;
+}
+
+.transfer-stats .config {
+    margin: 2em 0 1em 0;
+}
+
+.transfer-stats table th, .transfer-stats table td {
+    padding: 0.5em 1em;
+}
+
+.transfer-stats table th {
+    text-align: left;
+}
+
+.transfer-stats table td {
+    text-align: right;
+}
+
+.transfer-stats table td.transfer-rate {
+    color: #999999;
+}
\ No newline at end of file
diff --git a/react/src/DealTransfers.js b/react/src/DealTransfers.js
index 49622fbaf..178aaf53e 100644
--- a/react/src/DealTransfers.js
+++ b/react/src/DealTransfers.js
@@ -1,12 +1,13 @@
 import React from "react";
 import {Chart} from "react-google-charts";
 import {useQuery} from "@apollo/react-hooks";
-import {TransfersQuery} from "./gql";
+import {TransfersQuery, TransferStatsQuery} from "./gql";
 import moment from "moment"
 import {PageContainer} from "./Components";
 import {Link} from "react-router-dom";
 import {toFixed} from "./util";
 import arrowLeftRightImg from './bootstrap-icons/icons/arrow-left-right.svg'
+import './DealTransfers.css'
 
 var maxMegabits = 0
 
@@ -17,6 +18,13 @@ export function DealTransfersPage(props) {
 }
 
 function DealTransfersContent(props) {
+    return <div>
+        <DealTransfersChart />
+        <TransferStats />
+    </div>
+}
+
+function DealTransfersChart(props) {
     const {loading, error, data} = useQuery(TransfersQuery, {
         pollInterval: 1000,
         fetchPolicy: 'network-only',
@@ -70,33 +78,120 @@ function DealTransfersContent(props) {
     </div>
 }
 
-export function DealTransfersMenuItem(props) {
-    const {data} = useQuery(TransfersQuery, {
-        pollInterval: 5000,
+function TransferStats(props) {
+    const {loading, error, data} = useQuery(TransferStatsQuery, {
+        pollInterval: 1000,
         fetchPolicy: 'network-only',
     })
 
+    if (loading) {
+        return <div>Loading...</div>
+    }
+    if (error) {
+        return <div>Error: {error.message}</div>
+    }
+
+    const stats = data.transferStats
+    return <div className="transfer-stats">
+        <div>
+            <h3>Transfers by Host</h3>
+            <table>
+                <tbody>
+                <tr>
+                    <th>Host</th>
+                    <th>Total</th>
+                    <th>Queued</th>
+                    <th>Active</th>
+                    <th>Stalled</th>
+                    <th>Rate</th>
+                </tr>
+                { stats.Stats.map((hostStats) => {
+                    return <tr key={hostStats.Host}>
+                        <td>{hostStats.Host}</td>
+                        <td>{hostStats.Total}</td>
+                        <td>{hostStats.Total-hostStats.Started}</td>
+                        <td>{hostStats.Started-hostStats.Stalled}</td>
+                        <td>{hostStats.Stalled}</td>
+                        <td className="transfer-rate">{getTransferRate(hostStats.TransferSamples)}</td>
+                    </tr>
+                }) }
+                </tbody>
+            </table>
+        </div>
+
+        <div className="config">
+            <h3>Config</h3>
+            <table>
+                <tbody>
+                <tr>
+                    <th>HTTP concurrent download limit:</th>
+                    <td>{stats.HttpMaxConcurrentDownloads}</td>
+                </tr>
+                </tbody>
+            </table>
+        </div>
+    </div>
+}
+
+function getTransferRate(samples) {
     var dataRate = 0
-    if (data && data.transfers.length) {
+    if (samples && samples.length) {
         // Clone from read-only to writable array
-        var points = data.transfers.map(p => ({At: p.At, Bytes: p.Bytes}))
+        var points = samples.map(p => ({At: p.At, Bytes: p.Bytes}))
         points.sort((a, b) => a.At.getTime() - b.At.getTime())
 
         // Get average transfer rate over last 10 seconds
+        const subset = points.slice(-10)
+        // Allow for some clock skew, but ignore samples older than 2 minutes
+        const cutOff = new Date(new Date().getTime() - 2*60*1000)
         var total = 0
-        const subset = points.slice(points.length-10)
+        var count = 0
         for (const point of subset) {
-            const megabits = 8 * Number(point.Bytes) / 1e6
-            total += megabits
+            if (point.At > cutOff) {
+                total += Number(point.Bytes)
+                count++
+            }
         }
-        dataRate = total / subset.length
+        if (count > 0) {
+            dataRate = total / count
+        }
+    }
+    return humanTransferRate(dataRate)
+}
+
+export function humanTransferRate(bytesPerSecond) {
+    // Convert from bytes to kbps or Mbps
+    var transferRate = ""
+    const kbps = (bytesPerSecond*8) / 1024
+    if (kbps < 1024) {
+        if (kbps < 10) {
+            return kbps.toFixed(2) + " kbps"
+        }
+        return kbps.toFixed(1) + " kbps"
+    }
+    const mbps = kbps / 1024
+    if (mbps < 10) {
+        return mbps.toFixed(2) + " Mbps"
+    }
+    return mbps.toFixed(1) + " Mbps"
+}
+
+export function DealTransfersMenuItem(props) {
+    const {data} = useQuery(TransfersQuery, {
+        pollInterval: 5000,
+        fetchPolicy: 'network-only',
+    })
+
+    var dataRate = ""
+    if (data && data.transfers) {
+        dataRate = getTransferRate(data.transfers)
     }
 
     return <Link key="deal-transfers" className="menu-item" to="/deal-transfers">
         <img className="icon" alt="" src={arrowLeftRightImg} />
         <h3>Deal Transfers</h3>
         <div className="menu-desc">
-            <b>{toFixed(dataRate, 1)}</b> Mbps (10s avg)
+            <b>{dataRate}</b> (10s avg)
         </div>
     </Link>
 }
diff --git a/react/src/Deals.css b/react/src/Deals.css
index 0fa422829..e36d35498 100644
--- a/react/src/Deals.css
+++ b/react/src/Deals.css
@@ -51,6 +51,11 @@
     flex-grow: 1;
 }
 
+.deals tr.show-actions td.message .message-content .message-text .transfer-rate {
+    padding-left: 1em;
+    color: #999999;
+}
+
 .deals tr.show-actions td.message .buttons {
     display: inline-block;
     min-width: 6em;
diff --git a/react/src/Deals.js b/react/src/Deals.js
index 59d66e02c..7dd1649cd 100644
--- a/react/src/Deals.js
+++ b/react/src/Deals.js
@@ -1,3 +1,4 @@
+/* global BigInt */
 import {useQuery} from "@apollo/react-hooks";
 import {
     DealsCountQuery,
@@ -18,6 +19,7 @@ import xImg from './bootstrap-icons/icons/x-lg.svg'
 import './Deals.css'
 import {Pagination} from "./Pagination";
 import {DealActions, IsPaused, IsTransferring} from "./DealDetail";
+import {humanTransferRate} from "./DealTransfers";
 
 const dealsBasePath = '/storage-deals'
 
@@ -174,6 +176,7 @@ function DealRow(props) {
                 <div className="message-content">
                     <span className="message-text">
                         {deal.Message}
+                        <TransferRate deal={deal} />
                     </span>
                     {showActions ? <DealActions deal={props.deal} refetchQueries={[DealsListQuery]} compact={true} /> : null}
                 </div>
@@ -182,6 +185,41 @@ function DealRow(props) {
     )
 }
 
+function TransferRate({deal}) {
+    if (!IsTransferring(deal) || IsPaused(deal) || deal.Transferred === 0 || deal.IsTransferStalled) {
+        return null
+    }
+
+    if(deal.TransferSamples.length < 2) {
+        return null
+    }
+
+    // Clone from read-only to writable array and sort points
+    var points = deal.TransferSamples.map(p => ({ At: p.At, Bytes: p.Bytes }))
+    points.sort((a, b) => a.At.getTime() - b.At.getTime())
+
+    // Get the average rate from the last 10 seconds of samples.
+    points = points.slice(-10)
+    // Allow for some clock skew, but ignore samples older than 2 minutes
+    const cutOff = new Date(new Date().getTime() - 2*60*1000)
+    var samples = []
+    for (const pt of points) {
+        if (pt.At > cutOff) {
+            samples.push(pt)
+        }
+    }
+    if (!samples.length) {
+        return null
+    }
+
+    // Get the delta between the first sample and last sample.
+    const delta = samples[samples.length-1].Bytes - samples[0].Bytes
+
+    return <span className="transfer-rate">
+        {humanTransferRate(Number(delta) / samples.length)}
+    </span>
+}
+
 export function StorageDealsMenuItem(props) {
     const {data} = useQuery(DealsCountQuery, {
         pollInterval: 5000,
diff --git a/react/src/gql.js b/react/src/gql.js
index 999ee87b6..c980942d9 100644
--- a/react/src/gql.js
+++ b/react/src/gql.js
@@ -87,6 +87,12 @@ const DealsListQuery = gql`
                     Size
                     Params
                 }
+                Transferred
+                TransferSamples {
+                    At
+                    Bytes
+                }
+                IsTransferStalled
                 Sector {
                     ID
                     Offset
@@ -454,6 +460,24 @@ const TransfersQuery = gql`
     }
 `;
 
+const TransferStatsQuery = gql`
+    query AppTransferStatsQuery {
+        transferStats {
+            HttpMaxConcurrentDownloads
+            Stats {
+                Host
+                Total
+                Started
+                Stalled
+                TransferSamples {
+                    At
+                    Bytes
+                }
+            }
+        }
+    }
+`;
+
 const FundsLogsQuery = gql`
     query AppFundsLogsQuery($cursor: BigInt, $offset: Int, $limit: Int) {
         fundsLogs(cursor: $cursor, offset: $offset, limit: $limit) {
@@ -573,6 +597,7 @@ export {
     FundsMoveToEscrow,
     StorageAskUpdate,
     TransfersQuery,
+    TransferStatsQuery,
     MpoolQuery,
     SealingPipelineQuery,
     Libp2pAddrInfoQuery,
diff --git a/storagemarket/provider.go b/storagemarket/provider.go
index c8530c144..ef53ed5a6 100644
--- a/storagemarket/provider.go
+++ b/storagemarket/provider.go
@@ -195,10 +195,6 @@ func (p *Provider) DealBySignedProposalCid(ctx context.Context, propCid cid.Cid)
 	return deal, nil
 }
 
-func (p *Provider) NBytesReceived(dealUuid uuid.UUID) uint64 {
-	return p.transfers.getBytes(dealUuid)
-}
-
 func (p *Provider) GetAsk() *storagemarket.SignedStorageAsk {
 	return p.askGetter.GetAsk()
 }
diff --git a/storagemarket/transfer_limiter.go b/storagemarket/transfer_limiter.go
index 29098e8ce..e1709470f 100644
--- a/storagemarket/transfer_limiter.go
+++ b/storagemarket/transfer_limiter.go
@@ -279,3 +279,66 @@ func (tl *transferLimiter) setBytes(dealUuid uuid.UUID, bytes uint64) {
 		xfer.bytes = bytes
 	}
 }
+
+func (tl *transferLimiter) isStalled(dealUuid uuid.UUID) bool {
+	now := time.Now()
+
+	tl.lk.RLock()
+	defer tl.lk.RUnlock()
+
+	xfer, ok := tl.xfers[dealUuid]
+	if !ok {
+		return false
+	}
+
+	if !xfer.isStarted() {
+		return false
+	}
+
+	return now.Sub(xfer.updatedAt) >= tl.cfg.StallTimeout
+}
+
+type HostTransferStats struct {
+	Host      string
+	Total     int
+	Started   int
+	Stalled   int
+	DealUuids []uuid.UUID
+}
+
+func (tl *transferLimiter) stats() []*HostTransferStats {
+	now := time.Now()
+
+	tl.lk.RLock()
+	defer tl.lk.RUnlock()
+
+	// Count the number of transfers, started and stalled per host
+	stats := make(map[string]*HostTransferStats)
+	for _, xfer := range tl.xfers {
+		hostStats, ok := stats[xfer.host]
+		if !ok {
+			hostStats = &HostTransferStats{Host: xfer.host}
+			stats[xfer.host] = hostStats
+		}
+
+		hostStats.Total++
+		hostStats.DealUuids = append(hostStats.DealUuids, xfer.deal.DealUuid)
+		if !xfer.isStarted() {
+			continue
+		}
+		hostStats.Started++
+		if now.Sub(xfer.updatedAt) >= tl.cfg.StallTimeout {
+			hostStats.Stalled++
+		}
+	}
+
+	// Sort the stats by host
+	statsArr := make([]*HostTransferStats, 0, len(stats))
+	for _, s := range stats {
+		statsArr = append(statsArr, s)
+	}
+	sort.Slice(statsArr, func(i, j int) bool {
+		return statsArr[i].Host < statsArr[j].Host
+	})
+	return statsArr
+}
diff --git a/storagemarket/transfers.go b/storagemarket/transfers.go
index f23715cd1..e1d3fe0c9 100644
--- a/storagemarket/transfers.go
+++ b/storagemarket/transfers.go
@@ -8,16 +8,11 @@ import (
 	"github.com/google/uuid"
 )
 
-// Keep up to 60s of samples
+// Keep up to 20s of samples
 const maxSamples = 20
 
-// Transfers returns a map of active transfers, sampled for up to 60s
-func (p *Provider) Transfers() map[uuid.UUID][]transferPoint {
-	return p.transfers.transfers()
-}
-
 // A sample of the number of bytes transferred at the given time
-type transferPoint struct {
+type TransferPoint struct {
 	// The time at which the sample was taken, truncated to the nearest second
 	At time.Time
 	// The number of bytes transferred
@@ -29,7 +24,7 @@ type dealTransfers struct {
 	// maps from deal UUID -> samples of the number of bytes transferred in
 	// the last 60 seconds
 	samplesLk sync.RWMutex
-	samples   map[uuid.UUID][]transferPoint
+	samples   map[uuid.UUID][]TransferPoint
 
 	// Maps from active transfer deal UUID to number of bytes transferred
 	activeLk sync.RWMutex
@@ -38,7 +33,7 @@ type dealTransfers struct {
 
 func newDealTransfers() *dealTransfers {
 	return &dealTransfers{
-		samples: make(map[uuid.UUID][]transferPoint),
+		samples: make(map[uuid.UUID][]TransferPoint),
 		active:  make(map[uuid.UUID]uint64),
 	}
 }
@@ -77,11 +72,11 @@ func (dt *dealTransfers) sample(now time.Time) {
 		// Get the array of samples
 		points, ok := dt.samples[dealUUID]
 		if !ok {
-			points = []transferPoint{}
+			points = []TransferPoint{}
 		}
 
 		// Add a point for the current second
-		point := transferPoint{
+		point := TransferPoint{
 			At:    now,
 			Bytes: bytes,
 		}
@@ -91,7 +86,7 @@ func (dt *dealTransfers) sample(now time.Time) {
 		if len(points) > maxSamples {
 			// We need to make a copy of the array to avoid leaking memory.
 			// It may be worth using a ring-buffer if this is too slow.
-			newPoints := make([]transferPoint, maxSamples)
+			newPoints := make([]TransferPoint, maxSamples)
 			copy(newPoints, points[len(points)-maxSamples:])
 			points = newPoints
 		}
@@ -109,14 +104,14 @@ func (dt *dealTransfers) sample(now time.Time) {
 
 		// Add a point for non-active samples, with the same transferred bytes
 		// as the previous point
-		points = append(points, transferPoint{
+		points = append(points, TransferPoint{
 			At:    now,
 			Bytes: points[len(points)-1].Bytes,
 		})
 
 		// If the maximum number of samples has been reached
 		if len(points) > maxSamples {
-			// If the transfer has been inactive for 60s, delete the samples
+			// If the transfer has been inactive for 20s, delete the samples
 			if points[0].Bytes == points[len(points)-1].Bytes {
 				delete(dt.samples, dealUUID)
 				continue
@@ -130,19 +125,32 @@ func (dt *dealTransfers) sample(now time.Time) {
 	}
 }
 
-func (dt *dealTransfers) transfers() map[uuid.UUID][]transferPoint {
+func (dt *dealTransfers) transfers() map[uuid.UUID][]TransferPoint {
 	dt.samplesLk.RLock()
 	defer dt.samplesLk.RUnlock()
 
-	deals := make(map[uuid.UUID][]transferPoint, len(dt.samples))
+	deals := make(map[uuid.UUID][]TransferPoint, len(dt.samples))
 	for dealUUID, points := range dt.samples {
-		pts := make([]transferPoint, len(points))
+		pts := make([]TransferPoint, len(points))
 		copy(pts, points)
 		deals[dealUUID] = pts
 	}
 	return deals
 }
 
+func (dt *dealTransfers) transfer(dealUUID uuid.UUID) []TransferPoint {
+	dt.samplesLk.RLock()
+	defer dt.samplesLk.RUnlock()
+
+	points, ok := dt.samples[dealUUID]
+	if !ok {
+		return nil
+	}
+	pts := make([]TransferPoint, len(points))
+	copy(pts, points)
+	return pts
+}
+
 func (dt *dealTransfers) setBytes(dealUUID uuid.UUID, bytes uint64) {
 	dt.activeLk.Lock()
 	defer dt.activeLk.Unlock()
@@ -163,3 +171,28 @@ func (dt *dealTransfers) complete(dealUUID uuid.UUID) {
 
 	delete(dt.active, dealUUID)
 }
+
+// Transfers returns a map of active transfers, sampled for up to 20s
+func (p *Provider) Transfers() map[uuid.UUID][]TransferPoint {
+	return p.transfers.transfers()
+}
+
+// Transfer returns samples of an active transfer, sampled for up to 20s
+func (p *Provider) Transfer(dealUuid uuid.UUID) []TransferPoint {
+	return p.transfers.transfer(dealUuid)
+}
+
+// Get the number of bytes downloaded in total for the given deal
+func (p *Provider) NBytesReceived(dealUuid uuid.UUID) uint64 {
+	return p.transfers.getBytes(dealUuid)
+}
+
+// Indicates if a transfer has been marked as "stalled", ie the transfer is
+// not making any progress
+func (p *Provider) IsTransferStalled(dealUuid uuid.UUID) bool {
+	return p.xferLimiter.isStalled(dealUuid)
+}
+
+func (p *Provider) TransferStats() []*HostTransferStats {
+	return p.xferLimiter.stats()
+}