refactor: OnDealExpiredOrSlashed get deal by proposal instead of deal ID
This commit is contained in:
Dirk McCormick
parent
f33c02e56b
commit
c7e7f89026
@ -18,7 +18,7 @@ import (
|
||||
"github.com/filecoin-project/go-state-types/big"
|
||||
"github.com/filecoin-project/go-state-types/crypto"
|
||||
"github.com/filecoin-project/go-state-types/exitcode"
|
||||
|
||||
market0 "github.com/filecoin-project/specs-actors/actors/builtin/market"
|
||||
miner2 "github.com/filecoin-project/specs-actors/v2/actors/builtin"
|
||||
market2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/market"
|
||||
|
||||
@ -40,8 +40,8 @@ type ClientNodeAdapter struct {
|
||||
|
||||
fundmgr *market.FundManager
|
||||
ev *events.Events
|
||||
dsMatcher *dealStateMatcher
|
||||
scMgr *SectorCommittedManager
|
||||
deMgr *DealExpiryManager
|
||||
}
|
||||
|
||||
type clientApi struct {
|
||||
@ -60,9 +60,10 @@ func NewClientNodeAdapter(mctx helpers.MetricsCtx, lc fx.Lifecycle, stateapi ful
|
||||
|
||||
fundmgr: fundmgr,
|
||||
ev: ev,
|
||||
dsMatcher: newDealStateMatcher(state.NewStatePredicates(state.WrapFastAPI(capi))),
|
||||
}
|
||||
a.scMgr = NewSectorCommittedManager(ev, a, &apiWrapper{api: capi})
|
||||
dsMatcher := newDealStateMatcher(state.NewStatePredicates(state.WrapFastAPI(capi)))
|
||||
a.deMgr = NewDealExpiryManager(ev, capi, capi, dsMatcher)
|
||||
return a
|
||||
}
|
||||
|
||||
@ -249,94 +250,8 @@ func (c *ClientNodeAdapter) OnDealSectorCommitted(ctx context.Context, provider
|
||||
return c.scMgr.OnDealSectorCommitted(ctx, provider, sectorNumber, marketactor.DealProposal(proposal), *publishCid, cb)
|
||||
}
|
||||
|
||||
// TODO: Replace dealID parameter with DealProposal
|
||||
func (c *ClientNodeAdapter) OnDealExpiredOrSlashed(ctx context.Context, dealID abi.DealID, onDealExpired storagemarket.DealExpiredCallback, onDealSlashed storagemarket.DealSlashedCallback) error {
|
||||
head, err := c.ChainHead(ctx)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("client: failed to get chain head: %w", err)
|
||||
}
|
||||
|
||||
sd, err := c.StateMarketStorageDeal(ctx, dealID, head.Key())
|
||||
if err != nil {
|
||||
return xerrors.Errorf("client: failed to look up deal %d on chain: %w", dealID, err)
|
||||
}
|
||||
|
||||
// Called immediately to check if the deal has already expired or been slashed
|
||||
checkFunc := func(ts *types.TipSet) (done bool, more bool, err error) {
|
||||
if ts == nil {
|
||||
// keep listening for events
|
||||
return false, true, nil
|
||||
}
|
||||
|
||||
// Check if the deal has already expired
|
||||
if sd.Proposal.EndEpoch <= ts.Height() {
|
||||
onDealExpired(nil)
|
||||
return true, false, nil
|
||||
}
|
||||
|
||||
// If there is no deal assume it's already been slashed
|
||||
if sd.State.SectorStartEpoch < 0 {
|
||||
onDealSlashed(ts.Height(), nil)
|
||||
return true, false, nil
|
||||
}
|
||||
|
||||
// No events have occurred yet, so return
|
||||
// done: false, more: true (keep listening for events)
|
||||
return false, true, nil
|
||||
}
|
||||
|
||||
// Called when there was a match against the state change we're looking for
|
||||
// and the chain has advanced to the confidence height
|
||||
stateChanged := func(ts *types.TipSet, ts2 *types.TipSet, states events.StateChange, h abi.ChainEpoch) (more bool, err error) {
|
||||
// Check if the deal has already expired
|
||||
if ts2 == nil || sd.Proposal.EndEpoch <= ts2.Height() {
|
||||
onDealExpired(nil)
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// Timeout waiting for state change
|
||||
if states == nil {
|
||||
log.Error("timed out waiting for deal expiry")
|
||||
return false, nil
|
||||
}
|
||||
|
||||
changedDeals, ok := states.(state.ChangedDeals)
|
||||
if !ok {
|
||||
panic("Expected state.ChangedDeals")
|
||||
}
|
||||
|
||||
deal, ok := changedDeals[dealID]
|
||||
if !ok {
|
||||
// No change to deal
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Deal was slashed
|
||||
if deal.To == nil {
|
||||
onDealSlashed(ts2.Height(), nil)
|
||||
return false, nil
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Called when there was a chain reorg and the state change was reverted
|
||||
revert := func(ctx context.Context, ts *types.TipSet) error {
|
||||
// TODO: Is it ok to just ignore this?
|
||||
log.Warn("deal state reverted; TODO: actually handle this!")
|
||||
return nil
|
||||
}
|
||||
|
||||
// Watch for state changes to the deal
|
||||
match := c.dsMatcher.matcher(ctx, dealID)
|
||||
|
||||
// Wait until after the end epoch for the deal and then timeout
|
||||
timeout := (sd.Proposal.EndEpoch - head.Height()) + 1
|
||||
if err := c.ev.StateChanged(checkFunc, stateChanged, revert, int(build.MessageConfidence)+1, timeout, match); err != nil {
|
||||
return xerrors.Errorf("failed to set up state changed handler: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
func (c *ClientNodeAdapter) OnDealExpiredOrSlashed(ctx context.Context, publishCid cid.Cid, proposal market0.DealProposal, onDealExpired storagemarket.DealExpiredCallback, onDealSlashed storagemarket.DealSlashedCallback) error {
|
||||
return c.deMgr.OnDealExpiredOrSlashed(ctx, publishCid, proposal, onDealExpired, onDealSlashed)
|
||||
}
|
||||
|
||||
func (c *ClientNodeAdapter) SignProposal(ctx context.Context, signer address.Address, proposal market2.DealProposal) (*market2.ClientDealProposal, error) {
|
||||
|
||||
152
markets/storageadapter/ondealexpired.go
Normal file
152
markets/storageadapter/ondealexpired.go
Normal file
@ -0,0 +1,152 @@
|
||||
package storageadapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
market0 "github.com/filecoin-project/specs-actors/actors/builtin/market"
|
||||
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/market"
|
||||
"github.com/filecoin-project/lotus/chain/events"
|
||||
"github.com/filecoin-project/lotus/chain/events/state"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/ipfs/go-cid"
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
"github.com/filecoin-project/go-fil-markets/storagemarket"
|
||||
"github.com/filecoin-project/go-state-types/abi"
|
||||
sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
|
||||
)
|
||||
|
||||
type demEventsAPI interface {
|
||||
Called(check events.CheckFunc, msgHnd events.MsgHandler, rev events.RevertHandler, confidence int, timeout abi.ChainEpoch, mf events.MsgMatchFunc) error
|
||||
StateChanged(check events.CheckFunc, scHnd events.StateChangeHandler, rev events.RevertHandler, confidence int, timeout abi.ChainEpoch, mf events.StateMatchFunc) error
|
||||
}
|
||||
|
||||
type demChainAPI interface {
|
||||
ChainHead(context.Context) (*types.TipSet, error)
|
||||
}
|
||||
|
||||
type DealExpiryManagerAPI interface {
|
||||
demEventsAPI
|
||||
demChainAPI
|
||||
GetCurrentDealInfo(ctx context.Context, tok sealing.TipSetToken, proposal *market.DealProposal, publishCid cid.Cid) (sealing.CurrentDealInfo, error)
|
||||
}
|
||||
|
||||
type dealExpiryManagerAdapter struct {
|
||||
demEventsAPI
|
||||
demChainAPI
|
||||
*sealing.CurrentDealInfoManager
|
||||
}
|
||||
|
||||
type DealExpiryManager struct {
|
||||
demAPI DealExpiryManagerAPI
|
||||
dsMatcher *dealStateMatcher
|
||||
}
|
||||
|
||||
func NewDealExpiryManager(ev demEventsAPI, ch demChainAPI, tskAPI sealing.CurrentDealInfoTskAPI, dsMatcher *dealStateMatcher) *DealExpiryManager {
|
||||
dim := &sealing.CurrentDealInfoManager{
|
||||
CDAPI: &sealing.CurrentDealInfoAPIAdapter{CurrentDealInfoTskAPI: tskAPI},
|
||||
}
|
||||
|
||||
adapter := &dealExpiryManagerAdapter{
|
||||
demEventsAPI: ev,
|
||||
demChainAPI: ch,
|
||||
CurrentDealInfoManager: dim,
|
||||
}
|
||||
return newDealExpiryManager(adapter, dsMatcher)
|
||||
}
|
||||
|
||||
func newDealExpiryManager(demAPI DealExpiryManagerAPI, dsMatcher *dealStateMatcher) *DealExpiryManager {
|
||||
return &DealExpiryManager{demAPI: demAPI, dsMatcher: dsMatcher}
|
||||
}
|
||||
|
||||
func (mgr *DealExpiryManager) OnDealExpiredOrSlashed(ctx context.Context, publishCid cid.Cid, proposal market0.DealProposal, onDealExpired storagemarket.DealExpiredCallback, onDealSlashed storagemarket.DealSlashedCallback) error {
|
||||
head, err := mgr.demAPI.ChainHead(ctx)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("client: failed to get chain head: %w", err)
|
||||
}
|
||||
|
||||
prop := market.DealProposal(proposal)
|
||||
res, err := mgr.demAPI.GetCurrentDealInfo(ctx, head.Key().Bytes(), &prop, publishCid)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("awaiting deal expired: getting deal info: %w", err)
|
||||
}
|
||||
|
||||
// Called immediately to check if the deal has already expired or been slashed
|
||||
checkFunc := func(ts *types.TipSet) (done bool, more bool, err error) {
|
||||
if ts == nil {
|
||||
// keep listening for events
|
||||
return false, true, nil
|
||||
}
|
||||
|
||||
// Check if the deal has already expired
|
||||
if res.MarketDeal.Proposal.EndEpoch <= ts.Height() {
|
||||
onDealExpired(nil)
|
||||
return true, false, nil
|
||||
}
|
||||
|
||||
// If there is no deal assume it's already been slashed
|
||||
if res.MarketDeal.State.SectorStartEpoch < 0 {
|
||||
onDealSlashed(ts.Height(), nil)
|
||||
return true, false, nil
|
||||
}
|
||||
|
||||
// No events have occurred yet, so return
|
||||
// done: false, more: true (keep listening for events)
|
||||
return false, true, nil
|
||||
}
|
||||
|
||||
// Called when there was a match against the state change we're looking for
|
||||
// and the chain has advanced to the confidence height
|
||||
stateChanged := func(ts *types.TipSet, ts2 *types.TipSet, states events.StateChange, h abi.ChainEpoch) (more bool, err error) {
|
||||
// Check if the deal has already expired
|
||||
if ts2 == nil || res.MarketDeal.Proposal.EndEpoch <= ts2.Height() {
|
||||
onDealExpired(nil)
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// Timeout waiting for state change
|
||||
if states == nil {
|
||||
log.Error("timed out waiting for deal expiry")
|
||||
return false, nil
|
||||
}
|
||||
|
||||
changedDeals, ok := states.(state.ChangedDeals)
|
||||
if !ok {
|
||||
panic("Expected state.ChangedDeals")
|
||||
}
|
||||
|
||||
deal, ok := changedDeals[res.DealID]
|
||||
if !ok {
|
||||
// No change to deal
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Deal was slashed
|
||||
if deal.To == nil {
|
||||
onDealSlashed(ts2.Height(), nil)
|
||||
return false, nil
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Called when there was a chain reorg and the state change was reverted
|
||||
revert := func(ctx context.Context, ts *types.TipSet) error {
|
||||
// TODO: Is it ok to just ignore this?
|
||||
log.Warn("deal state reverted; TODO: actually handle this!")
|
||||
return nil
|
||||
}
|
||||
|
||||
// Watch for state changes to the deal
|
||||
match := mgr.dsMatcher.matcher(ctx, res.DealID)
|
||||
|
||||
// Wait until after the end epoch for the deal and then timeout
|
||||
timeout := (res.MarketDeal.Proposal.EndEpoch - head.Height()) + 1
|
||||
if err := mgr.demAPI.StateChanged(checkFunc, stateChanged, revert, int(build.MessageConfidence)+1, timeout, match); err != nil {
|
||||
return xerrors.Errorf("failed to set up state changed handler: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -18,6 +18,8 @@ import (
|
||||
"github.com/filecoin-project/go-state-types/abi"
|
||||
"github.com/filecoin-project/go-state-types/crypto"
|
||||
"github.com/filecoin-project/go-state-types/exitcode"
|
||||
"github.com/filecoin-project/lotus/chain/events/state"
|
||||
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"
|
||||
@ -26,7 +28,6 @@ import (
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/market"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
|
||||
"github.com/filecoin-project/lotus/chain/events"
|
||||
"github.com/filecoin-project/lotus/chain/events/state"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
|
||||
"github.com/filecoin-project/lotus/lib/sigs"
|
||||
@ -51,8 +52,8 @@ type ProviderNodeAdapter struct {
|
||||
|
||||
addBalanceSpec *api.MessageSendSpec
|
||||
maxDealCollateralMultiplier uint64
|
||||
dsMatcher *dealStateMatcher
|
||||
scMgr *SectorCommittedManager
|
||||
deMgr *DealExpiryManager
|
||||
}
|
||||
|
||||
func NewProviderNodeAdapter(fc *config.MinerFeeConfig, dc *config.DealmakingConfig) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, secb *sectorblocks.SectorBlocks, full v1api.FullNode, dealPublisher *DealPublisher) storagemarket.StorageProviderNode {
|
||||
@ -66,7 +67,6 @@ func NewProviderNodeAdapter(fc *config.MinerFeeConfig, dc *config.DealmakingConf
|
||||
secb: secb,
|
||||
ev: ev,
|
||||
dealPublisher: dealPublisher,
|
||||
dsMatcher: newDealStateMatcher(state.NewStatePredicates(state.WrapFastAPI(full))),
|
||||
}
|
||||
if fc != nil {
|
||||
na.addBalanceSpec = &api.MessageSendSpec{MaxFee: abi.TokenAmount(fc.MaxMarketBalanceAddFee)}
|
||||
@ -77,6 +77,9 @@ func NewProviderNodeAdapter(fc *config.MinerFeeConfig, dc *config.DealmakingConf
|
||||
}
|
||||
na.scMgr = NewSectorCommittedManager(ev, na, &apiWrapper{api: full})
|
||||
|
||||
dsMatcher := newDealStateMatcher(state.NewStatePredicates(state.WrapFastAPI(full)))
|
||||
na.deMgr = NewDealExpiryManager(ev, full, full, dsMatcher)
|
||||
|
||||
return na
|
||||
}
|
||||
}
|
||||
@ -328,93 +331,8 @@ func (n *ProviderNodeAdapter) GetDataCap(ctx context.Context, addr address.Addre
|
||||
return sp, err
|
||||
}
|
||||
|
||||
func (n *ProviderNodeAdapter) OnDealExpiredOrSlashed(ctx context.Context, dealID abi.DealID, onDealExpired storagemarket.DealExpiredCallback, onDealSlashed storagemarket.DealSlashedCallback) error {
|
||||
head, err := n.ChainHead(ctx)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("client: failed to get chain head: %w", err)
|
||||
}
|
||||
|
||||
sd, err := n.StateMarketStorageDeal(ctx, dealID, head.Key())
|
||||
if err != nil {
|
||||
return xerrors.Errorf("client: failed to look up deal %d on chain: %w", dealID, err)
|
||||
}
|
||||
|
||||
// Called immediately to check if the deal has already expired or been slashed
|
||||
checkFunc := func(ts *types.TipSet) (done bool, more bool, err error) {
|
||||
if ts == nil {
|
||||
// keep listening for events
|
||||
return false, true, nil
|
||||
}
|
||||
|
||||
// Check if the deal has already expired
|
||||
if sd.Proposal.EndEpoch <= ts.Height() {
|
||||
onDealExpired(nil)
|
||||
return true, false, nil
|
||||
}
|
||||
|
||||
// If there is no deal assume it's already been slashed
|
||||
if sd.State.SectorStartEpoch < 0 {
|
||||
onDealSlashed(ts.Height(), nil)
|
||||
return true, false, nil
|
||||
}
|
||||
|
||||
// No events have occurred yet, so return
|
||||
// done: false, more: true (keep listening for events)
|
||||
return false, true, nil
|
||||
}
|
||||
|
||||
// Called when there was a match against the state change we're looking for
|
||||
// and the chain has advanced to the confidence height
|
||||
stateChanged := func(ts *types.TipSet, ts2 *types.TipSet, states events.StateChange, h abi.ChainEpoch) (more bool, err error) {
|
||||
// Check if the deal has already expired
|
||||
if ts2 == nil || sd.Proposal.EndEpoch <= ts2.Height() {
|
||||
onDealExpired(nil)
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// Timeout waiting for state change
|
||||
if states == nil {
|
||||
log.Error("timed out waiting for deal expiry")
|
||||
return false, nil
|
||||
}
|
||||
|
||||
changedDeals, ok := states.(state.ChangedDeals)
|
||||
if !ok {
|
||||
panic("Expected state.ChangedDeals")
|
||||
}
|
||||
|
||||
deal, ok := changedDeals[dealID]
|
||||
if !ok {
|
||||
// No change to deal
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Deal was slashed
|
||||
if deal.To == nil {
|
||||
onDealSlashed(ts2.Height(), nil)
|
||||
return false, nil
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Called when there was a chain reorg and the state change was reverted
|
||||
revert := func(ctx context.Context, ts *types.TipSet) error {
|
||||
// TODO: Is it ok to just ignore this?
|
||||
log.Warn("deal state reverted; TODO: actually handle this!")
|
||||
return nil
|
||||
}
|
||||
|
||||
// Watch for state changes to the deal
|
||||
match := n.dsMatcher.matcher(ctx, dealID)
|
||||
|
||||
// Wait until after the end epoch for the deal and then timeout
|
||||
timeout := (sd.Proposal.EndEpoch - head.Height()) + 1
|
||||
if err := n.ev.StateChanged(checkFunc, stateChanged, revert, int(build.MessageConfidence)+1, timeout, match); err != nil {
|
||||
return xerrors.Errorf("failed to set up state changed handler: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
func (n *ProviderNodeAdapter) OnDealExpiredOrSlashed(ctx context.Context, publishCid cid.Cid, proposal market0.DealProposal, onDealExpired storagemarket.DealExpiredCallback, onDealSlashed storagemarket.DealSlashedCallback) error {
|
||||
return n.deMgr.OnDealExpiredOrSlashed(ctx, publishCid, proposal, onDealExpired, onDealSlashed)
|
||||
}
|
||||
|
||||
var _ storagemarket.StorageProviderNode = &ProviderNodeAdapter{}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user