fix: Circular search for deals

config/util.go

@@ -52,6 +52,10 @@ func (addr Address) MarshalText() ([]byte, error) {
 	return []byte(address.Address(addr).String()), nil
 }
 
+func (addr Address) Unwrap() address.Address {
+	return address.Address(addr)
+}
+
 func CfgAddrArrToNative(addrs []Address) []address.Address {
 	nativeAddrs := make([]address.Address, len(addrs))
 	for _, addr := range addrs {

retrievalprovider/requestvalidation.go

@@ -182,7 +182,7 @@ func (rv *ProviderRequestValidator) runDealDecisionLogic(ctx context.Context, de
 }
 
 func (rv *ProviderRequestValidator) acceptDeal(ctx context.Context, deal *types.ProviderDealState) (retrievalmarket.DealStatus, error) {
-	minerdeals, err := rv.pieceInfo.GetPieceInfoFromCid(ctx, deal.PayloadCID, deal.PieceCID)
+	minerDeals, err := rv.pieceInfo.GetPieceInfoFromCid(ctx, deal.PayloadCID, deal.PieceCID)
 	if err != nil {
 		if errors.Is(err, repo.ErrNotFound) {
 			return retrievalmarket.DealStatusDealNotFound, err
@@ -194,10 +194,24 @@ func (rv *ProviderRequestValidator) acceptDeal(ctx context.Context, deal *types.
 	defer cancel()
 
 	//todo this deal may not match with query ask, no way to get miner id in current protocol
-	selectDeal := minerdeals[0]
-	deal.SelStorageProposalCid = selectDeal.ProposalCid
-	ask, err := rv.retrievalAsk.GetAsk(ctx, selectDeal.Proposal.Provider)
-	if err != nil {
+	var ask *types.RetrievalAsk
+	for _, minerDeal := range minerDeals {
+		minerCfg, err := rv.cfg.MinerProviderConfig(minerDeal.Proposal.Provider, true)
+		if err != nil {
+			return retrievalmarket.DealStatusErrored, err
+		}
+		if minerCfg.RetrievalPaymentAddress.Unwrap().Empty() {
+			continue
+		}
+		deal.SelStorageProposalCid = minerDeal.ProposalCid
+		ask, err = rv.retrievalAsk.GetAsk(ctx, minerDeal.Proposal.Provider)
+		if err != nil {
+			log.Warnf("got %s ask failed: %v", minerDeal.Proposal.Provider, err)
+		} else {
+			break
+		}
+	}
+	if ask == nil {
 		return retrievalmarket.DealStatusErrored, err
 	}
 

retrievalprovider/stream_handler.go

@@ -82,7 +82,7 @@ func (p *RetrievalStreamHandler) HandleQueryStream(stream rmnet.RetrievalQuerySt
 	if err != nil {
 		answer.Status = retrievalmarket.QueryResponseError
 		if errors.Is(err, repo.ErrNotFound) {
-			answer.Message = fmt.Sprintf("retrieve piece(%s) or payload(%s) failed, not found",
+			answer.Message = fmt.Sprintf("retrieve piece(%v) or payload(%s) failed, not found",
 				query.PieceCID, query.PayloadCID)
 		} else {
 			answer.Message = fmt.Sprintf("failed to fetch piece to retrieve from: %s", err)
@@ -91,38 +91,57 @@ func (p *RetrievalStreamHandler) HandleQueryStream(stream rmnet.RetrievalQuerySt
 		return
 	}
 
-	selectDeal := minerDeals[0]
-
-	answer.Status = retrievalmarket.QueryResponseAvailable
-	// todo payload size maybe different with real piece size.
-	answer.Size = uint64(selectDeal.Proposal.PieceSize.Unpadded()) // TODO: verify on intermediate
-	answer.PieceCIDFound = retrievalmarket.QueryItemAvailable
-	minerCfg, err := p.cfg.MinerProviderConfig(selectDeal.Proposal.Provider, true)
-	if err != nil {
-		answer.Status = retrievalmarket.QueryResponseError
-		answer.Message = err.Error()
-		sendResp(answer)
+	var validDealCount int
+	answers := make([]retrievalmarket.QueryResponse, len(minerDeals))
+	for i, deal := range minerDeals {
+		answers[i] = answer
+		answers[i].Status = retrievalmarket.QueryResponseAvailable
+		// todo payload size maybe different with real piece size.
+		answers[i].Size = uint64(deal.Proposal.PieceSize.Unpadded()) // TODO: verify on intermediate
+		answers[i].PieceCIDFound = retrievalmarket.QueryItemAvailable
+
+		minerCfg, err := p.cfg.MinerProviderConfig(deal.Proposal.Provider, true)
+		if err != nil {
+			answer.Status = retrievalmarket.QueryResponseError
+			answer.Message = err.Error()
+			sendResp(answer)
+		}
+		paymentAddr := minerCfg.RetrievalPaymentAddress.Unwrap()
+		if paymentAddr == address.Undef {
+			answers[i].Status = retrievalmarket.QueryResponseError
+			answers[i].Message = "must specify payment address"
+			continue
+		}
+		answers[i].PaymentAddress = paymentAddr
+		validDealCount++
 	}
-	paymentAddr := address.Address(minerCfg.RetrievalPaymentAddress)
-	if paymentAddr == address.Undef {
-		answer.Status = retrievalmarket.QueryResponseError
-		answer.Message = "must specific payment address in venus-market"
-		sendResp(answer)
+	if validDealCount == 0 {
+		sendResp(answers[0])
 		return
 	}
-	answer.PaymentAddress = paymentAddr
 
-	ask, err := p.askRepo.GetAsk(ctx, selectDeal.Proposal.Provider)
-	if err != nil {
-		answer.Status = retrievalmarket.QueryResponseError
-		answer.Message = fmt.Sprintf("failed to price deal: %s", err)
-		sendResp(answer)
-		return
+	validAnswers := make([]*retrievalmarket.QueryResponse, 0, validDealCount)
+	for i, deal := range minerDeals {
+		if len(answers[i].Message) != 0 {
+			continue
+		}
+		ask, err := p.askRepo.GetAsk(ctx, deal.Proposal.Provider)
+		if err != nil {
+			answer.Status = retrievalmarket.QueryResponseError
+			answer.Message = fmt.Sprintf("failed to got deal price: %s, %s", deal.Proposal.Provider, err)
+			continue
+		}
+
+		answers[i].MinPricePerByte = ask.PricePerByte
+		answers[i].MaxPaymentInterval = ask.PaymentInterval
+		answers[i].MaxPaymentIntervalIncrease = ask.PaymentIntervalIncrease
+		answers[i].UnsealPrice = ask.UnsealPrice
+		validAnswers = append(validAnswers, &answers[i])
 	}
 
-	answer.MinPricePerByte = ask.PricePerByte
-	answer.MaxPaymentInterval = ask.PaymentInterval
-	answer.MaxPaymentIntervalIncrease = ask.PaymentIntervalIncrease
-	answer.UnsealPrice = ask.UnsealPrice
-	sendResp(answer)
+	if len(validAnswers) == 0 {
+		sendResp(answers[0])
+	} else {
+		sendResp(*validAnswers[0])
+	}
 }