feat: add job to cancel stalled retrievals (#1233) (#1239)

* feat: add job to clean up stalled retrievals

* chore: run configs doc gen

* chore: refactor stalled retrieval name

add warning if logs duration is less than stalled retrieval timeout

Author: jacobheun

node/builder.go

@@ -531,7 +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(HandleRetrievalEventsKey, modules.HandleRetrievalGraphsyncUpdates(time.Duration(cfg.Dealmaking.RetrievalLogDuration), time.Duration(cfg.Dealmaking.StalledRetrievalTimeout))),
 		Override(HandleRetrievalKey, lotus_modules.HandleRetrieval),
 		Override(new(*lp2pimpl.TransportsListener), modules.NewTransportsListener(cfg)),
 		Override(new(*protocolproxy.ProtocolProxy), modules.NewProtocolProxy(cfg)),

node/config/def.go

@@ -89,6 +89,7 @@ func DefaultBoost() *Boost {
 
 			DealProposalLogDuration: Duration(time.Hour * 24),
 			RetrievalLogDuration:    Duration(time.Hour * 24),
+			StalledRetrievalTimeout: Duration(time.Minute * 30),
 
 			RetrievalPricing: &lotus_config.RetrievalPricing{
 				Strategy: RetrievalPricingDefaultMode,

node/config/doc_gen.go

@@ -175,7 +175,11 @@ type DealmakingConfig struct {
 	// 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.
+	// Note RetrievalLogDuration should exceed the StalledRetrievalTimeout as the
+	// logs db is leveraged for pruning stalled retrievals.
 	RetrievalLogDuration Duration
+	// The amount of time stalled retrieval deals will remain open before being canceled.
+	StalledRetrievalTimeout Duration
 
 	// A command used for fine-grained evaluation of storage deals
 	// see https://boost.filecoin.io/configuration/deal-filters for more details

node/config/types.go

@@ -142,9 +142,9 @@ func NewRetrievalLogDB(db *RetrievalSqlDB) *rtvllog.RetrievalLogDB {
 }
 
 // Write graphsync retrieval updates to the database
-func HandleRetrievalGraphsyncUpdates(duration time.Duration) func(lc fx.Lifecycle, db *rtvllog.RetrievalLogDB, m lotus_retrievalmarket.RetrievalProvider, dt lotus_dtypes.ProviderDataTransfer) {
+func HandleRetrievalGraphsyncUpdates(duration time.Duration, stalledDuration time.Duration) func(lc fx.Lifecycle, db *rtvllog.RetrievalLogDB, m lotus_retrievalmarket.RetrievalProvider, dt lotus_dtypes.ProviderDataTransfer) {
 	return func(lc fx.Lifecycle, db *rtvllog.RetrievalLogDB, m lotus_retrievalmarket.RetrievalProvider, dt lotus_dtypes.ProviderDataTransfer) {
-		rel := rtvllog.NewRetrievalLog(db, duration)
+		rel := rtvllog.NewRetrievalLog(db, duration, dt, stalledDuration)
 
 		relctx, cancel := context.WithCancel(context.Background())
 		type unsubFn func()

node/modules/retrieval.go

@@ -123,6 +123,10 @@ func (d *RetrievalLogDB) List(ctx context.Context, cursor *time.Time, offset int
 	return d.list(ctx, offset, limit, where, whereArgs...)
 }
 
+func (d *RetrievalLogDB) ListLastUpdatedAndOpen(ctx context.Context, lastUpdated time.Time) ([]RetrievalDealState, error) {
+	return d.list(ctx, 0, 0, "UpdatedAt <= ? AND Status != 'DealStatusCompleted' AND Status != 'DealStatusCancelled'", lastUpdated)
+}
+
 func (d *RetrievalLogDB) list(ctx context.Context, offset int, limit int, where string, whereArgs ...interface{}) ([]RetrievalDealState, error) {
 	qry := "SELECT " +
 		"CreatedAt, " +

retrievalmarket/rtvllog/db.go

@@ -7,32 +7,42 @@ import (
 
 	datatransfer "github.com/filecoin-project/go-data-transfer"
 	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
+	lotus_dtypes "github.com/filecoin-project/lotus/node/modules/dtypes"
 	logging "github.com/ipfs/go-log/v2"
 )
 
 var log = logging.Logger("rtrvlog")
 
 type RetrievalLog struct {
-	db       *RetrievalLogDB
-	duration time.Duration
-	ctx      context.Context
+	db             *RetrievalLogDB
+	duration       time.Duration
+	dataTransfer   lotus_dtypes.ProviderDataTransfer
+	stalledTimeout time.Duration
+	ctx            context.Context
 
 	lastUpdateLk sync.Mutex
 	lastUpdate   map[string]time.Time
 }
 
-func NewRetrievalLog(db *RetrievalLogDB, duration time.Duration) *RetrievalLog {
+func NewRetrievalLog(db *RetrievalLogDB, duration time.Duration, dt lotus_dtypes.ProviderDataTransfer, stalledTimeout time.Duration) *RetrievalLog {
+	if duration < stalledTimeout {
+		log.Warnf("the RetrievalLogDuration (%s) should exceed the StalledRetrievalTimeout (%s)", duration.String(), stalledTimeout.String())
+	}
+
 	return &RetrievalLog{
-		db:         db,
-		duration:   duration,
-		lastUpdate: make(map[string]time.Time),
+		db:             db,
+		duration:       duration,
+		dataTransfer:   dt,
+		stalledTimeout: stalledTimeout,
+		lastUpdate:     make(map[string]time.Time),
 	}
 }
 
 func (r *RetrievalLog) Start(ctx context.Context) {
 	r.ctx = ctx
 	go r.gcUpdateMap(ctx)
 	go r.gcDatabase(ctx)
+	go r.gcRetrievals(ctx)
 }
 
 // Called when there is a retrieval ask query
@@ -225,3 +235,34 @@ func (r *RetrievalLog) gcDatabase(ctx context.Context) {
 		}
 	}
 }
+
+// Periodically cancels stalled retrievals older than 30mins
+func (r *RetrievalLog) gcRetrievals(ctx context.Context) {
+	ticker := time.NewTicker(5 * time.Minute)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case now := <-ticker.C:
+			// Get retrievals last updated
+			rows, err := r.db.ListLastUpdatedAndOpen(ctx, now.Add(-r.stalledTimeout))
+
+			if err != nil {
+				log.Errorw("error fetching open, stalled retrievals", "err", err)
+				continue
+			}
+
+			for _, row := range rows {
+				chid := datatransfer.ChannelID{Initiator: row.PeerID, Responder: row.LocalPeerID, ID: row.TransferID}
+				err := r.dataTransfer.CloseDataTransferChannel(ctx, chid)
+				if err != nil {
+					log.Errorw("error canceling retrieval", "dealID", row.DealID, "err", err)
+				} else {
+					log.Infof("Canceled retrieval %s, older than %s", row.DealID, r.stalledTimeout)
+				}
+			}
+		}
+	}
+}