dealpublisher: Fully validate deals before publishing

markets/storageadapter/dealpublisher.go

@@ -14,6 +14,8 @@ import (
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/big"
+	"github.com/filecoin-project/go-state-types/exitcode"
+	market0 "github.com/filecoin-project/specs-actors/actors/builtin/market"
 	market2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/market"
 
 	"github.com/filecoin-project/lotus/api"
@@ -35,6 +37,7 @@ type dealPublisherAPI interface {
 	WalletHas(context.Context, address.Address) (bool, error)
 	StateAccountKey(context.Context, address.Address, types.TipSetKey) (address.Address, error)
 	StateLookupID(context.Context, address.Address, types.TipSetKey) (address.Address, error)
+	StateCall(context.Context, *types.Message, types.TipSetKey) (*api.InvocResult, error)
 }
 
 // DealPublisher batches deal publishing so that many deals can be included in
@@ -295,7 +298,7 @@ func (p *DealPublisher) publishReady(ready []*pendingDeal) {
 		// Validate the deal
 		if err := p.validateDeal(pd.deal); err != nil {
 			// Validation failed, complete immediately with an error
-			go onComplete(pd, cid.Undef, err)
+			go onComplete(pd, cid.Undef, xerrors.Errorf("publish validation failed: %w", err))
 			continue
 		}
 
@@ -315,6 +318,13 @@ func (p *DealPublisher) publishReady(ready []*pendingDeal) {
 // validateDeal checks that the deal proposal start epoch hasn't already
 // elapsed
 func (p *DealPublisher) validateDeal(deal market2.ClientDealProposal) error {
+	start := time.Now()
+
+	pcid, err := deal.Proposal.Cid()
+	if err != nil {
+		return xerrors.Errorf("computing proposal cid: %w", err)
+	}
+
 	head, err := p.api.ChainHead(p.ctx)
 	if err != nil {
 		return err
@@ -324,6 +334,41 @@ func (p *DealPublisher) validateDeal(deal market2.ClientDealProposal) error {
 			"cannot publish deal with piece CID %s: current epoch %d has passed deal proposal start epoch %d",
 			deal.Proposal.PieceCID, head.Height(), deal.Proposal.StartEpoch)
 	}
+
+	mi, err := p.api.StateMinerInfo(p.ctx, deal.Proposal.Provider, types.EmptyTSK)
+	if err != nil {
+		return xerrors.Errorf("getting provider info: %w", err)
+	}
+
+	params, err := actors.SerializeParams(&market2.PublishStorageDealsParams{
+		Deals: []market0.ClientDealProposal{deal},
+	})
+	if err != nil {
+		return xerrors.Errorf("serializing PublishStorageDeals params failed: %w", err)
+	}
+
+	addr, _, err := p.as.AddressFor(p.ctx, p.api, mi, api.DealPublishAddr, big.Zero(), big.Zero())
+	if err != nil {
+		return xerrors.Errorf("selecting address for publishing deals: %w", err)
+	}
+
+	res, err := p.api.StateCall(p.ctx, &types.Message{
+		To:     market.Address,
+		From:   addr,
+		Value:  types.NewInt(0),
+		Method: market.Methods.PublishStorageDeals,
+		Params: params,
+	}, head.Key())
+	if err != nil {
+		return xerrors.Errorf("simulating deal publish message: %w", err)
+	}
+	if res.MsgRct.ExitCode != exitcode.Ok {
+		return xerrors.Errorf("simulating deal publish message: non-zero exitcode %s; message: %s", res.MsgRct.ExitCode, res.Error)
+	}
+
+	took := time.Now().Sub(start)
+	log.Infow("validating deal", "took", took, "proposal", pcid)
+
 	return nil
 }
 

markets/storageadapter/dealpublisher_test.go

@@ -381,6 +381,10 @@ func (d *dpAPI) StateLookupID(ctx context.Context, a address.Address, key types.
 	panic("don't call me")
 }
 
+func (d *dpAPI) StateCall(ctx context.Context, message *types.Message, key types.TipSetKey) (*api.InvocResult, error) {
+	return &api.InvocResult{MsgRct: &types.MessageReceipt{ExitCode: 0}}, nil
+}
+
 func getClientActor(t *testing.T) address.Address {
 	return tutils.NewActorAddr(t, "client")
 }