lotus/markets/storageadapter/dealpublisher.go

467 lines
14 KiB
Go
Raw Normal View History

2021-01-08 15:28:38 +00:00
package storageadapter
import (
"context"
"fmt"
2021-01-08 15:28:38 +00:00
"strings"
"sync"
"time"
2021-07-07 17:12:03 +00:00
"github.com/ipfs/go-cid"
2021-01-08 15:28:38 +00:00
"go.uber.org/fx"
2021-07-07 17:12:03 +00:00
"golang.org/x/xerrors"
2021-01-08 15:28:38 +00:00
2021-07-07 17:12:03 +00:00
"github.com/filecoin-project/go-address"
2021-01-08 15:28:38 +00:00
"github.com/filecoin-project/go-state-types/abi"
2021-07-07 17:12:03 +00:00
"github.com/filecoin-project/go-state-types/big"
2022-04-20 21:34:28 +00:00
"github.com/filecoin-project/go-state-types/builtin"
2022-09-26 16:47:48 +00:00
"github.com/filecoin-project/go-state-types/builtin/v9/market"
2022-06-14 15:00:51 +00:00
"github.com/filecoin-project/go-state-types/exitcode"
2021-01-08 15:28:38 +00:00
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
2021-01-08 15:28:38 +00:00
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/chain/types"
2021-07-07 17:12:03 +00:00
"github.com/filecoin-project/lotus/node/config"
"github.com/filecoin-project/lotus/storage/ctladdr"
2021-01-08 15:28:38 +00:00
)
type dealPublisherAPI interface {
ChainHead(context.Context) (*types.TipSet, error)
2021-01-08 15:28:38 +00:00
MpoolPushMessage(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec) (*types.SignedMessage, error)
2022-04-20 21:34:28 +00:00
StateMinerInfo(context.Context, address.Address, types.TipSetKey) (api.MinerInfo, error)
2021-07-07 17:12:03 +00:00
WalletBalance(context.Context, address.Address) (types.BigInt, error)
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)
2021-01-08 15:28:38 +00:00
}
// DealPublisher batches deal publishing so that many deals can be included in
// a single publish message. This saves gas for miners that publish deals
// frequently.
// When a deal is submitted, the DealPublisher waits a configurable amount of
// time for other deals to be submitted before sending the publish message.
// There is a configurable maximum number of deals that can be included in one
// message. When the limit is reached the DealPublisher immediately submits a
// publish message with all deals in the queue.
type DealPublisher struct {
api dealPublisherAPI
as *ctladdr.AddressSelector
2021-01-08 15:28:38 +00:00
ctx context.Context
Shutdown context.CancelFunc
maxDealsPerPublishMsg uint64
publishPeriod time.Duration
publishSpec *api.MessageSendSpec
lk sync.Mutex
2022-04-11 17:32:09 +00:00
pending []*pendingDeal
cancelWaitForMoreDeals context.CancelFunc
publishPeriodStart time.Time
startEpochSealingBuffer abi.ChainEpoch
2021-01-08 15:28:38 +00:00
}
// A deal that is queued to be published
type pendingDeal struct {
ctx context.Context
2022-03-12 18:07:35 +00:00
deal market.ClientDealProposal
Result chan publishResult
}
// The result of publishing a deal
type publishResult struct {
msgCid cid.Cid
err error
}
2022-03-12 18:07:35 +00:00
func newPendingDeal(ctx context.Context, deal market.ClientDealProposal) *pendingDeal {
return &pendingDeal{
ctx: ctx,
deal: deal,
Result: make(chan publishResult),
}
}
type PublishMsgConfig struct {
// The amount of time to wait for more deals to arrive before
// publishing
Period time.Duration
// The maximum number of deals to include in a single PublishStorageDeals
// message
MaxDealsPerMsg uint64
// Minimum start epoch buffer to give time for sealing of sector with deal
StartEpochSealingBuffer uint64
}
2021-01-08 15:28:38 +00:00
func NewDealPublisher(
feeConfig *config.MinerFeeConfig,
publishMsgCfg PublishMsgConfig,
) func(lc fx.Lifecycle, full api.FullNode, as *ctladdr.AddressSelector) *DealPublisher {
return func(lc fx.Lifecycle, full api.FullNode, as *ctladdr.AddressSelector) *DealPublisher {
2021-01-08 15:28:38 +00:00
maxFee := abi.NewTokenAmount(0)
if feeConfig != nil {
maxFee = abi.TokenAmount(feeConfig.MaxPublishDealsFee)
}
publishSpec := &api.MessageSendSpec{MaxFee: maxFee}
2021-07-07 17:12:03 +00:00
dp := newDealPublisher(full, as, publishMsgCfg, publishSpec)
2021-01-08 15:28:38 +00:00
lc.Append(fx.Hook{
OnStop: func(ctx context.Context) error {
dp.Shutdown()
return nil
},
})
return dp
}
}
func newDealPublisher(
dpapi dealPublisherAPI,
as *ctladdr.AddressSelector,
publishMsgCfg PublishMsgConfig,
2021-01-08 15:28:38 +00:00
publishSpec *api.MessageSendSpec,
) *DealPublisher {
ctx, cancel := context.WithCancel(context.Background())
return &DealPublisher{
api: dpapi,
as: as,
ctx: ctx,
Shutdown: cancel,
maxDealsPerPublishMsg: publishMsgCfg.MaxDealsPerMsg,
publishPeriod: publishMsgCfg.Period,
startEpochSealingBuffer: abi.ChainEpoch(publishMsgCfg.StartEpochSealingBuffer),
publishSpec: publishSpec,
2021-01-08 15:28:38 +00:00
}
}
// PendingDeals returns the list of deals that are queued up to be published
2021-02-05 17:58:55 +00:00
func (p *DealPublisher) PendingDeals() api.PendingDealInfo {
p.lk.Lock()
defer p.lk.Unlock()
// Filter out deals whose context has been cancelled
deals := make([]*pendingDeal, 0, len(p.pending))
for _, dl := range p.pending {
if dl.ctx.Err() == nil {
deals = append(deals, dl)
}
}
2022-03-12 18:07:35 +00:00
pending := make([]market.ClientDealProposal, len(deals))
2021-02-05 17:58:55 +00:00
for i, deal := range deals {
2021-02-05 21:26:37 +00:00
pending[i] = deal.deal
2021-02-05 17:58:55 +00:00
}
return api.PendingDealInfo{
Deals: pending,
PublishPeriodStart: p.publishPeriodStart,
PublishPeriod: p.publishPeriod,
}
}
// ForcePublishPendingDeals publishes all pending deals without waiting for
// the publish period to elapse
func (p *DealPublisher) ForcePublishPendingDeals() {
p.lk.Lock()
defer p.lk.Unlock()
log.Infof("force publishing deals")
p.publishAllDeals()
}
2022-03-12 18:07:35 +00:00
func (p *DealPublisher) Publish(ctx context.Context, deal market.ClientDealProposal) (cid.Cid, error) {
2021-01-08 15:28:38 +00:00
pdeal := newPendingDeal(ctx, deal)
// Add the deal to the queue
p.processNewDeal(pdeal)
// Wait for the deal to be submitted
select {
case <-ctx.Done():
return cid.Undef, ctx.Err()
case res := <-pdeal.Result:
return res.msgCid, res.err
}
}
func (p *DealPublisher) processNewDeal(pdeal *pendingDeal) {
p.lk.Lock()
defer p.lk.Unlock()
// Filter out any cancelled deals
p.filterCancelledDeals()
// If all deals have been cancelled, clear the wait-for-deals timer
if len(p.pending) == 0 && p.cancelWaitForMoreDeals != nil {
p.cancelWaitForMoreDeals()
p.cancelWaitForMoreDeals = nil
}
// Make sure the new deal hasn't been cancelled
if pdeal.ctx.Err() != nil {
return
}
pdealPropCid, err := pdeal.deal.Proposal.Cid()
if err != nil {
log.Warn("failed to calculate proposal CID for new pending Deal with piece cid %s", pdeal.deal.Proposal.PieceCID)
return
}
// Sanity check that new deal isn't already in the queue
for _, pd := range p.pending {
pdPropCid, err := pd.deal.Proposal.Cid()
if err != nil {
log.Warn("failed to calculate proposal CID for pending Deal already in publish queue with piece cid %s", pd.deal.Proposal.PieceCID)
return
}
if pdPropCid.Equals(pdealPropCid) {
log.Warn("tried to process new pending deal with piece CID %s that is already in publish queue; returning", pdeal.deal.Proposal.PieceCID)
return
}
}
2021-01-08 15:28:38 +00:00
// Add the new deal to the queue
p.pending = append(p.pending, pdeal)
log.Infof("add deal with piece CID %s to publish deals queue - %d deals in queue (max queue size %d)",
pdeal.deal.Proposal.PieceCID, len(p.pending), p.maxDealsPerPublishMsg)
// If the maximum number of deals per message has been reached or we're not batching, send a
// publish message
if uint64(len(p.pending)) >= p.maxDealsPerPublishMsg || p.publishPeriod == 0 {
2021-01-08 15:28:38 +00:00
log.Infof("publish deals queue has reached max size of %d, publishing deals", p.maxDealsPerPublishMsg)
p.publishAllDeals()
return
}
// Otherwise wait for more deals to arrive or the timeout to be reached
p.waitForMoreDeals()
}
func (p *DealPublisher) waitForMoreDeals() {
// Check if we're already waiting for deals
if !p.publishPeriodStart.IsZero() {
elapsed := build.Clock.Since(p.publishPeriodStart)
2021-01-08 15:28:38 +00:00
log.Infof("%s elapsed of / %s until publish deals queue is published",
elapsed, p.publishPeriod)
return
}
// Set a timeout to wait for more deals to arrive
log.Infof("waiting publish deals queue period of %s before publishing", p.publishPeriod)
ctx, cancel := context.WithCancel(p.ctx)
// Create the timer _before_ taking the current time so publishPeriod+timeout is always >=
// the actual timer timeout.
timer := build.Clock.Timer(p.publishPeriod)
p.publishPeriodStart = build.Clock.Now()
2021-01-08 15:28:38 +00:00
p.cancelWaitForMoreDeals = cancel
go func() {
select {
case <-ctx.Done():
timer.Stop()
case <-timer.C:
p.lk.Lock()
defer p.lk.Unlock()
// The timeout has expired so publish all pending deals
log.Infof("publish deals queue period of %s has expired, publishing deals", p.publishPeriod)
p.publishAllDeals()
}
}()
}
func (p *DealPublisher) publishAllDeals() {
// If the timeout hasn't yet been cancelled, cancel it
if p.cancelWaitForMoreDeals != nil {
p.cancelWaitForMoreDeals()
p.cancelWaitForMoreDeals = nil
p.publishPeriodStart = time.Time{}
}
// Filter out any deals that have been cancelled
p.filterCancelledDeals()
deals := p.pending
2021-01-08 15:28:38 +00:00
p.pending = nil
// Send the publish message
go p.publishReady(deals)
}
func (p *DealPublisher) publishReady(ready []*pendingDeal) {
if len(ready) == 0 {
return
}
// onComplete is called when the publish message has been sent or there
// was an error
onComplete := func(pd *pendingDeal, msgCid cid.Cid, err error) {
// Send the publish result on the pending deal's Result channel
res := publishResult{
msgCid: msgCid,
err: err,
}
select {
case <-p.ctx.Done():
case <-pd.ctx.Done():
case pd.Result <- res:
}
}
// Validate each deal to make sure it can be published
validated := make([]*pendingDeal, 0, len(ready))
2022-03-12 18:07:35 +00:00
deals := make([]market.ClientDealProposal, 0, len(ready))
2021-01-08 15:28:38 +00:00
for _, pd := range ready {
// Validate the deal
if err := p.validateDeal(pd.deal); err != nil {
// Validation failed, complete immediately with an error
go onComplete(pd, cid.Undef, xerrors.Errorf("publish validation failed: %w", err))
continue
}
validated = append(validated, pd)
2021-01-08 15:28:38 +00:00
deals = append(deals, pd.deal)
}
// Send the publish message
msgCid, err := p.publishDealProposals(deals)
// Signal that each deal has been published
for _, pd := range validated {
go onComplete(pd, msgCid, err)
2021-01-08 15:28:38 +00:00
}
}
// validateDeal checks that the deal proposal start epoch hasn't already
// elapsed
2022-03-12 18:07:35 +00:00
func (p *DealPublisher) validateDeal(deal market.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
}
if head.Height()+p.startEpochSealingBuffer > deal.Proposal.StartEpoch {
return xerrors.Errorf(
"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)
}
2022-03-12 18:07:35 +00:00
params, err := actors.SerializeParams(&market.PublishStorageDealsParams{
Deals: []market.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{
2022-04-20 21:34:28 +00:00
To: builtin.StorageMarketActorAddr,
From: addr,
Value: types.NewInt(0),
2022-04-20 21:34:28 +00:00
Method: builtin.MethodsMarket.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
}
2021-01-08 15:28:38 +00:00
// Sends the publish message
2022-03-12 18:07:35 +00:00
func (p *DealPublisher) publishDealProposals(deals []market.ClientDealProposal) (cid.Cid, error) {
if len(deals) == 0 {
return cid.Undef, nil
}
2021-01-08 15:28:38 +00:00
log.Infof("publishing %d deals in publish deals queue with piece CIDs: %s", len(deals), pieceCids(deals))
provider := deals[0].Proposal.Provider
for _, dl := range deals {
if dl.Proposal.Provider != provider {
msg := fmt.Sprintf("publishing %d deals failed: ", len(deals)) +
"not all deals are for same provider: " +
fmt.Sprintf("deal with piece CID %s is for provider %s ", deals[0].Proposal.PieceCID, deals[0].Proposal.Provider) +
fmt.Sprintf("but deal with piece CID %s is for provider %s", dl.Proposal.PieceCID, dl.Proposal.Provider)
return cid.Undef, xerrors.Errorf(msg)
}
}
2021-01-08 15:28:38 +00:00
mi, err := p.api.StateMinerInfo(p.ctx, provider, types.EmptyTSK)
if err != nil {
return cid.Undef, err
}
2022-03-12 18:07:35 +00:00
params, err := actors.SerializeParams(&market.PublishStorageDealsParams{
2021-01-08 15:28:38 +00:00
Deals: deals,
})
if err != nil {
return cid.Undef, xerrors.Errorf("serializing PublishStorageDeals params failed: %w", err)
}
2021-07-07 17:12:03 +00:00
addr, _, err := p.as.AddressFor(p.ctx, p.api, mi, api.DealPublishAddr, big.Zero(), big.Zero())
if err != nil {
return cid.Undef, xerrors.Errorf("selecting address for publishing deals: %w", err)
}
2021-01-08 15:28:38 +00:00
smsg, err := p.api.MpoolPushMessage(p.ctx, &types.Message{
2022-04-20 21:34:28 +00:00
To: builtin.StorageMarketActorAddr,
2021-07-07 17:12:03 +00:00
From: addr,
2021-01-08 15:28:38 +00:00
Value: types.NewInt(0),
2022-04-20 21:34:28 +00:00
Method: builtin.MethodsMarket.PublishStorageDeals,
2021-01-08 15:28:38 +00:00
Params: params,
}, p.publishSpec)
if err != nil {
return cid.Undef, err
}
return smsg.Cid(), nil
}
2022-03-12 18:07:35 +00:00
func pieceCids(deals []market.ClientDealProposal) string {
2021-01-08 15:28:38 +00:00
cids := make([]string, 0, len(deals))
for _, dl := range deals {
cids = append(cids, dl.Proposal.PieceCID.String())
}
return strings.Join(cids, ", ")
}
// filter out deals that have been cancelled
func (p *DealPublisher) filterCancelledDeals() {
filtered := p.pending[:0]
2021-01-08 15:28:38 +00:00
for _, pd := range p.pending {
if pd.ctx.Err() != nil {
continue
2021-01-08 15:28:38 +00:00
}
filtered = append(filtered, pd)
2021-01-08 15:28:38 +00:00
}
p.pending = filtered
2021-01-08 15:28:38 +00:00
}