Surface retrieval logs in Web UI

go.mod

@@ -27,7 +27,7 @@ require (
 	github.com/filecoin-project/go-data-transfer v1.15.2
 	github.com/filecoin-project/go-fil-commcid v0.1.0
 	github.com/filecoin-project/go-fil-commp-hashhash v0.1.0
-	github.com/filecoin-project/go-fil-markets v1.24.3
+	github.com/filecoin-project/go-fil-markets v1.25.1
 	github.com/filecoin-project/go-jsonrpc v0.1.8
 	github.com/filecoin-project/go-legs v0.4.9
 	github.com/filecoin-project/go-padreader v0.0.1

go.sum

@@ -344,8 +344,8 @@ github.com/filecoin-project/go-fil-commcid v0.1.0 h1:3R4ds1A9r6cr8mvZBfMYxTS88Oq
 github.com/filecoin-project/go-fil-commcid v0.1.0/go.mod h1:Eaox7Hvus1JgPrL5+M3+h7aSPHc0cVqpSxA+TxIEpZQ=
 github.com/filecoin-project/go-fil-commp-hashhash v0.1.0 h1:imrrpZWEHRnNqqv0tN7LXep5bFEVOVmQWHJvl2mgsGo=
 github.com/filecoin-project/go-fil-commp-hashhash v0.1.0/go.mod h1:73S8WSEWh9vr0fDJVnKADhfIv/d6dCbAGaAGWbdJEI8=
-github.com/filecoin-project/go-fil-markets v1.24.3 h1:B+8RGW9e7UCJM6rr8rPsmMNKWWPfWYW8Wo0I/gnQqEU=
-github.com/filecoin-project/go-fil-markets v1.24.3/go.mod h1:3lzXZt5mRHTHAmZ10sUviiutaLVL57B99FgBU1MYqWY=
+github.com/filecoin-project/go-fil-markets v1.25.1 h1:LZ+yabDaTJGA8Y2zsSimWM4WriW/V0H8vM+YMT8IF8I=
+github.com/filecoin-project/go-fil-markets v1.25.1/go.mod h1:3lzXZt5mRHTHAmZ10sUviiutaLVL57B99FgBU1MYqWY=
 github.com/filecoin-project/go-hamt-ipld v0.1.5 h1:uoXrKbCQZ49OHpsTCkrThPNelC4W3LPEk0OrS/ytIBM=
 github.com/filecoin-project/go-hamt-ipld v0.1.5/go.mod h1:6Is+ONR5Cd5R6XZoCse1CWaXZc0Hdb/JeX+EQCQzX24=
 github.com/filecoin-project/go-hamt-ipld/v2 v2.0.0 h1:b3UDemBYN2HNfk3KOXNuxgTTxlWi3xVvbQP0IT38fvM=

gql/resolver.go

@@ -10,6 +10,7 @@ import (
 	"github.com/filecoin-project/boost/fundmanager"
 	gqltypes "github.com/filecoin-project/boost/gql/types"
 	"github.com/filecoin-project/boost/node/config"
+	"github.com/filecoin-project/boost/retrievalmarket/rtvllog"
 	"github.com/filecoin-project/boost/sealingpipeline"
 	"github.com/filecoin-project/boost/storagemanager"
 	"github.com/filecoin-project/boost/storagemarket"
@@ -45,6 +46,7 @@ type resolver struct {
 	h          host.Host
 	dealsDB    *db.DealsDB
 	logsDB     *db.LogsDB
+	retDB      *rtvllog.RetrievalLogDB
 	plDB       *db.ProposalLogsDB
 	fundsDB    *db.FundsDB
 	fundMgr    *fundmanager.FundManager
@@ -60,13 +62,14 @@ type resolver struct {
 	fullNode   v1api.FullNode
 }
 
-func NewResolver(cfg *config.Boost, r lotus_repo.LockedRepo, h host.Host, dealsDB *db.DealsDB, logsDB *db.LogsDB, plDB *db.ProposalLogsDB, fundsDB *db.FundsDB, fundMgr *fundmanager.FundManager, storageMgr *storagemanager.StorageManager, spApi sealingpipeline.API, provider *storagemarket.Provider, legacyProv lotus_storagemarket.StorageProvider, legacyDT lotus_dtypes.ProviderDataTransfer, ps piecestore.PieceStore, sa retrievalmarket.SectorAccessor, dagst dagstore.Interface, publisher *storageadapter.DealPublisher, fullNode v1api.FullNode) *resolver {
+func NewResolver(cfg *config.Boost, r lotus_repo.LockedRepo, h host.Host, dealsDB *db.DealsDB, logsDB *db.LogsDB, retDB *rtvllog.RetrievalLogDB, plDB *db.ProposalLogsDB, fundsDB *db.FundsDB, fundMgr *fundmanager.FundManager, storageMgr *storagemanager.StorageManager, spApi sealingpipeline.API, provider *storagemarket.Provider, legacyProv lotus_storagemarket.StorageProvider, legacyDT lotus_dtypes.ProviderDataTransfer, ps piecestore.PieceStore, sa retrievalmarket.SectorAccessor, dagst dagstore.Interface, publisher *storageadapter.DealPublisher, fullNode v1api.FullNode) *resolver {
 	return &resolver{
 		cfg:        cfg,
 		repo:       r,
 		h:          h,
 		dealsDB:    dealsDB,
 		logsDB:     logsDB,
+		retDB:      retDB,
 		plDB:       plDB,
 		fundsDB:    fundsDB,
 		fundMgr:    fundMgr,

gql/resolver_rtvllog.go

@@ -0,0 +1,206 @@
+package gql
+
+import (
+	"context"
+	"time"
+
+	gqltypes "github.com/filecoin-project/boost/gql/types"
+	"github.com/filecoin-project/boost/retrievalmarket/rtvllog"
+	"github.com/graph-gophers/graphql-go"
+)
+
+type retrievalStateResolver struct {
+	rtvllog.RetrievalDealState
+	db *rtvllog.RetrievalLogDB
+}
+
+func (r *retrievalStateResolver) CreatedAt() graphql.Time {
+	return graphql.Time{Time: r.RetrievalDealState.CreatedAt}
+}
+
+func (r *retrievalStateResolver) UpdatedAt() graphql.Time {
+	return graphql.Time{Time: r.RetrievalDealState.UpdatedAt}
+}
+
+func (r *retrievalStateResolver) DealID() gqltypes.Uint64 {
+	return gqltypes.Uint64(r.RetrievalDealState.DealID)
+}
+
+func (r *retrievalStateResolver) TransferID() gqltypes.Uint64 {
+	return gqltypes.Uint64(r.RetrievalDealState.TransferID)
+}
+
+func (r *retrievalStateResolver) PeerID() string {
+	return r.RetrievalDealState.PeerID.String()
+}
+
+func (r *retrievalStateResolver) PayloadCID() string {
+	return r.RetrievalDealState.PayloadCID.String()
+}
+
+func (r *retrievalStateResolver) PieceCid() string {
+	if r.RetrievalDealState.PieceCID == nil {
+		return ""
+	}
+	return r.RetrievalDealState.PieceCID.String()
+}
+
+func (r *retrievalStateResolver) PaymentInterval() gqltypes.Uint64 {
+	return gqltypes.Uint64(r.RetrievalDealState.PaymentInterval)
+}
+
+func (r *retrievalStateResolver) PaymentIntervalIncrease() gqltypes.Uint64 {
+	return gqltypes.Uint64(r.RetrievalDealState.PaymentIntervalIncrease)
+}
+
+func (r *retrievalStateResolver) PricePerByte() gqltypes.BigInt {
+	return gqltypes.BigInt{Int: r.RetrievalDealState.PricePerByte}
+}
+
+func (r *retrievalStateResolver) UnsealPrice() gqltypes.BigInt {
+	return gqltypes.BigInt{Int: r.RetrievalDealState.UnsealPrice}
+}
+
+func (r *retrievalStateResolver) TotalSent() gqltypes.Uint64 {
+	return gqltypes.Uint64(r.RetrievalDealState.TotalSent)
+}
+
+func (r *retrievalStateResolver) DTEvents(ctx context.Context) ([]*retrievalDTEventResolver, error) {
+	if r.RetrievalDealState.TransferID == 0 || r.RetrievalDealState.LocalPeerID == "" {
+		return nil, nil
+	}
+
+	pid := r.RetrievalDealState.PeerID.String()
+	evts, err := r.db.ListDTEvents(ctx, pid, r.RetrievalDealState.TransferID)
+	if err != nil {
+		log.Warnw("getting data-transfer events for retrieval %s/%d: %s", pid, r.RetrievalDealState.TransferID, err)
+		return nil, nil
+	}
+
+	evtResolvers := make([]*retrievalDTEventResolver, 0, len(evts))
+	for _, evt := range evts {
+		evtResolvers = append(evtResolvers, &retrievalDTEventResolver{DTEvent: evt})
+	}
+	return evtResolvers, nil
+}
+
+type retrievalDTEventResolver struct {
+	rtvllog.DTEvent
+}
+
+func (r *retrievalDTEventResolver) CreatedAt() graphql.Time {
+	return graphql.Time{Time: r.DTEvent.CreatedAt}
+}
+
+func (r *retrievalStateResolver) MarketEvents(ctx context.Context) ([]*retrievalMarketEventResolver, error) {
+	if r.RetrievalDealState.DealID == 0 || r.RetrievalDealState.LocalPeerID == "" {
+		return nil, nil
+	}
+
+	pid := r.RetrievalDealState.PeerID.String()
+	evts, err := r.db.ListMarketEvents(ctx, pid, r.RetrievalDealState.DealID)
+	if err != nil {
+		log.Warnw("getting market events for retrieval %s/%d: %s", pid, r.RetrievalDealState.DealID, err)
+		return nil, nil
+	}
+
+	evtResolvers := make([]*retrievalMarketEventResolver, 0, len(evts))
+	for _, evt := range evts {
+		evtResolvers = append(evtResolvers, &retrievalMarketEventResolver{MarketEvent: evt})
+	}
+	return evtResolvers, nil
+}
+
+type retrievalMarketEventResolver struct {
+	rtvllog.MarketEvent
+}
+
+func (r *retrievalMarketEventResolver) CreatedAt() graphql.Time {
+	return graphql.Time{Time: r.MarketEvent.CreatedAt}
+}
+
+type retrievalStateListResolver struct {
+	TotalCount int32
+	Logs       []*retrievalStateResolver
+	More       bool
+}
+
+type retLogArgs struct {
+	PeerID string
+	DealID gqltypes.Uint64
+}
+
+func (r *resolver) RetrievalLog(ctx context.Context, args retLogArgs) (*retrievalStateResolver, error) {
+	st, err := r.retDB.Get(ctx, args.PeerID, uint64(args.DealID))
+	if err != nil {
+		return nil, err
+	}
+
+	return &retrievalStateResolver{RetrievalDealState: *st, db: r.retDB}, nil
+}
+
+type retrievalStatesArgs struct {
+	Cursor *gqltypes.BigInt // CreatedAt in milli-seconds
+	Offset graphql.NullInt
+	Limit  graphql.NullInt
+}
+
+func (r *resolver) RetrievalLogs(ctx context.Context, args retrievalStatesArgs) (*retrievalStateListResolver, error) {
+	offset := 0
+	if args.Offset.Set && args.Offset.Value != nil && *args.Offset.Value > 0 {
+		offset = int(*args.Offset.Value)
+	}
+
+	limit := 10
+	if args.Limit.Set && args.Limit.Value != nil && *args.Limit.Value > 0 {
+		limit = int(*args.Limit.Value)
+	}
+
+	// Fetch one extra row so that we can check if there are more rows
+	// beyond the limit
+	var cursor *time.Time
+	if args.Cursor != nil {
+		val := (*args.Cursor).Int64()
+		asTime := time.Unix(val/1000, (val%1000)*1e6)
+		cursor = &asTime
+	}
+	rows, err := r.retDB.List(ctx, cursor, offset, limit+1)
+	if err != nil {
+		return nil, err
+	}
+	more := len(rows) > limit
+	if more {
+		// Truncate list to limit
+		rows = rows[:limit]
+	}
+
+	// Get the total row count
+	count, err := r.retDB.Count(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	resolvers := make([]*retrievalStateResolver, 0, len(rows))
+	for _, row := range rows {
+		resolvers = append(resolvers, &retrievalStateResolver{RetrievalDealState: row, db: r.retDB})
+	}
+
+	return &retrievalStateListResolver{
+		TotalCount: int32(count),
+		Logs:       resolvers,
+		More:       more,
+	}, nil
+}
+
+type retStateCount struct {
+	Count  int32
+	Period gqltypes.Uint64
+}
+
+func (r *resolver) RetrievalLogsCount(ctx context.Context) (*retStateCount, error) {
+	count, err := r.retDB.Count(ctx)
+	return &retStateCount{
+		Count:  int32(count),
+		Period: gqltypes.Uint64(r.cfg.Dealmaking.RetrievalLogDuration),
+	}, err
+}

gql/schema.graphql

@@ -192,6 +192,51 @@ type ProposalLogsCount {
   Period: Uint64!
 }
 
+type DTEvent {
+  CreatedAt: Time!
+  Name: String!
+  Message: String!
+}
+
+type MarketEvent {
+  CreatedAt: Time!
+  Name: String!
+  Status: String!
+  Message: String!
+}
+
+type RetrievalState {
+  CreatedAt: Time!
+  UpdatedAt: Time!
+  PeerID: String!
+  TransferID: Uint64!
+  DealID: Uint64!
+  PayloadCID: String!
+  PieceCID: String!
+  PaymentInterval: Uint64!
+  PaymentIntervalIncrease: Uint64!
+  PricePerByte: BigInt!
+  UnsealPrice: BigInt!
+  Status: String!
+  Message: String!
+  TotalSent: Uint64!
+  DTStatus: String!
+  DTMessage: String!
+  DTEvents: [DTEvent]!
+  MarketEvents: [MarketEvent]!
+}
+
+type RetrievalStateList {
+  totalCount: Int!
+  logs: [RetrievalState]!
+  more: Boolean!
+}
+
+type RetrievalStatesCount {
+  Count: Int!
+  Period: Uint64!
+}
+
 type Storage {
   Staged: Uint64!
   Transferred: Uint64!
@@ -365,6 +410,15 @@ type RootQuery {
   """Get the number of accepted and rejected deal proposal logs"""
   proposalLogsCount: ProposalLogsCount!
 
+  """Get individual retrieval log"""
+  retrievalLog(peerID: String!, dealID: Uint64!): RetrievalState
+
+  """Get retrieval logs"""
+  retrievalLogs(cursor: BigInt, offset: Int, limit: Int): RetrievalStateList!
+
+  """Get the number of retrieval logs"""
+  retrievalLogsCount: RetrievalStatesCount!
+
   """Get information about a piece from the piece store, DAG store and database"""
   pieceStatus(pieceCid: String!): PieceStatus!
 

node/builder.go

@@ -20,6 +20,7 @@ import (
 	"github.com/filecoin-project/boost/node/modules/dtypes"
 	"github.com/filecoin-project/boost/protocolproxy"
 	"github.com/filecoin-project/boost/retrievalmarket/lp2pimpl"
+	"github.com/filecoin-project/boost/retrievalmarket/rtvllog"
 	"github.com/filecoin-project/boost/sealingpipeline"
 	"github.com/filecoin-project/boost/storagemanager"
 	"github.com/filecoin-project/boost/storagemarket"
@@ -141,6 +142,8 @@ const (
 	GetParamsKey
 	HandleMigrateProviderFundsKey
 	HandleDealsKey
+	HandleCreateRetrievalTablesKey
+	HandleRetrievalEventsKey
 	HandleRetrievalKey
 	HandleRetrievalTransportsKey
 	HandleProtocolProxyKey
@@ -404,10 +407,13 @@ var BoostNode = Options(
 	Override(new(lotus_dtypes.NetworkName), lotus_modules.StorageNetworkName),
 	Override(new(*sql.DB), modules.NewBoostDB),
 	Override(new(*modules.LogSqlDB), modules.NewLogsSqlDB),
+	Override(new(*modules.RetrievalSqlDB), modules.NewRetrievalSqlDB),
+	Override(HandleCreateRetrievalTablesKey, modules.CreateRetrievalTables),
 	Override(new(*db.DealsDB), modules.NewDealsDB),
 	Override(new(*db.LogsDB), modules.NewLogsDB),
 	Override(new(*db.ProposalLogsDB), modules.NewProposalLogsDB),
 	Override(new(*db.FundsDB), modules.NewFundsDB),
+	Override(new(*rtvllog.RetrievalLogDB), modules.NewRetrievalLogDB),
 )
 
 func ConfigBoost(cfg *config.Boost) Option {
@@ -525,6 +531,7 @@ func ConfigBoost(cfg *config.Boost) Option {
 		Override(new(retrievalmarket.RetrievalProviderNode), retrievaladapter.NewRetrievalProviderNode),
 		Override(new(rmnet.RetrievalMarketNetwork), lotus_modules.RetrievalNetwork),
 		Override(new(retrievalmarket.RetrievalProvider), lotus_modules.RetrievalProvider),
+		Override(HandleRetrievalEventsKey, modules.HandleRetrievalGraphsyncUpdates(time.Duration(cfg.Dealmaking.RetrievalLogDuration))),
 		Override(HandleRetrievalKey, lotus_modules.HandleRetrieval),
 		Override(new(*lp2pimpl.TransportsListener), modules.NewTransportsListener(cfg)),
 		Override(new(*protocolproxy.ProtocolProxy), modules.NewProtocolProxy(cfg)),

node/config/def.go

@@ -88,6 +88,7 @@ func DefaultBoost() *Boost {
 			StartEpochSealingBuffer: 480, // 480 epochs buffer == 4 hours from adding deal to sector to sector being sealed
 
 			DealProposalLogDuration: Duration(time.Hour * 24),
+			RetrievalLogDuration:    Duration(time.Hour * 24),
 
 			RetrievalPricing: &lotus_config.RetrievalPricing{
 				Strategy: RetrievalPricingDefaultMode,

node/config/types.go

@@ -174,6 +174,8 @@ type DealmakingConfig struct {
 	StartEpochSealingBuffer uint64
 	// The amount of time to keep deal proposal logs for before cleaning them up.
 	DealProposalLogDuration Duration
+	// The amount of time to keep retrieval deal logs for before cleaning them up.
+	RetrievalLogDuration Duration
 
 	// A command used for fine-grained evaluation of storage deals
 	// see https://docs.filecoin.io/mine/lotus/miner-configuration/#using-filters-for-fine-grained-storage-and-retrieval-deal-acceptance for more details