diff --git a/chain/actors/policy/policy.go b/chain/actors/policy/policy.go index bb35025ec..22ec4c742 100644 --- a/chain/actors/policy/policy.go +++ b/chain/actors/policy/policy.go @@ -3,6 +3,8 @@ package policy import ( "sort" + "github.com/filecoin-project/go-state-types/big" + "github.com/filecoin-project/go-state-types/abi" "github.com/filecoin-project/go-state-types/network" "github.com/filecoin-project/lotus/chain/actors" @@ -367,3 +369,31 @@ func GetDeclarationsMax(nwVer network.Version) int { panic("unsupported network version") } } + +func AggregateNetworkFee(nwVer network.Version, aggregateSize int, baseFee abi.TokenAmount) abi.TokenAmount { + switch actors.VersionForNetwork(nwVer) { + + case actors.Version0: + + return big.Zero() + + case actors.Version2: + + return big.Zero() + + case actors.Version3: + + return big.Zero() + + case actors.Version4: + + return big.Zero() + + case actors.Version5: + + return miner5.AggregateNetworkFee(aggregateSize, baseFee) + + default: + panic("unsupported network version") + } +} diff --git a/chain/actors/policy/policy.go.template b/chain/actors/policy/policy.go.template index 6b8f80bd0..feca2370a 100644 --- a/chain/actors/policy/policy.go.template +++ b/chain/actors/policy/policy.go.template @@ -3,6 +3,8 @@ package policy import ( "sort" + "github.com/filecoin-project/go-state-types/big" + "github.com/filecoin-project/go-state-types/abi" "github.com/filecoin-project/go-state-types/network" "github.com/filecoin-project/lotus/chain/actors" @@ -246,3 +248,18 @@ func GetDeclarationsMax(nwVer network.Version) int { panic("unsupported network version") } } + +func AggregateNetworkFee(nwVer network.Version, aggregateSize int, baseFee abi.TokenAmount) abi.TokenAmount { + switch actors.VersionForNetwork(nwVer) { + {{range .versions}} + case actors.Version{{.}}: + {{if (le . 4)}} + return big.Zero() + {{else}} + return miner{{.}}.AggregateNetworkFee(aggregateSize, baseFee) + {{end}} + {{end}} + default: + panic("unsupported network version") + } +} diff --git a/extern/storage-sealing/commit_batch.go b/extern/storage-sealing/commit_batch.go index b991eaa8d..819cb7fc7 100644 --- a/extern/storage-sealing/commit_batch.go +++ b/extern/storage-sealing/commit_batch.go @@ -7,6 +7,10 @@ import ( "sync" "time" + "github.com/filecoin-project/go-state-types/network" + + "github.com/filecoin-project/lotus/chain/actors" + "github.com/ipfs/go-cid" "golang.org/x/xerrors" @@ -32,9 +36,11 @@ type CommitBatcherApi interface { SendMsg(ctx context.Context, from, to address.Address, method abi.MethodNum, value, maxFee abi.TokenAmount, params []byte) (cid.Cid, error) StateMinerInfo(context.Context, address.Address, TipSetToken) (miner.MinerInfo, error) ChainHead(ctx context.Context) (TipSetToken, abi.ChainEpoch, error) + ChainBaseFee(context.Context, TipSetToken) (abi.TokenAmount, error) StateSectorPreCommitInfo(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tok TipSetToken) (*miner.SectorPreCommitOnChainInfo, error) StateMinerInitialPledgeCollateral(context.Context, address.Address, miner.SectorPreCommitInfo, TipSetToken) (big.Int, error) + StateNetworkVersion(ctx context.Context, tok TipSetToken) (network.Version, error) } type AggregateInput struct { @@ -52,9 +58,9 @@ type CommitBatcher struct { getConfig GetSealingConfigFunc prover ffiwrapper.Prover - deadlines map[abi.SectorNumber]time.Time - todo map[abi.SectorNumber]AggregateInput - waiting map[abi.SectorNumber][]chan sealiface.CommitBatchRes + cutoffs map[abi.SectorNumber]time.Time + todo map[abi.SectorNumber]AggregateInput + waiting map[abi.SectorNumber][]chan sealiface.CommitBatchRes notify, stop, stopped chan struct{} force chan chan []sealiface.CommitBatchRes @@ -71,9 +77,9 @@ func NewCommitBatcher(mctx context.Context, maddr address.Address, api CommitBat getConfig: getConfig, prover: prov, - deadlines: map[abi.SectorNumber]time.Time{}, - todo: map[abi.SectorNumber]AggregateInput{}, - waiting: map[abi.SectorNumber][]chan sealiface.CommitBatchRes{}, + cutoffs: map[abi.SectorNumber]time.Time{}, + todo: map[abi.SectorNumber]AggregateInput{}, + waiting: map[abi.SectorNumber][]chan sealiface.CommitBatchRes{}, notify: make(chan struct{}, 1), force: make(chan chan []sealiface.CommitBatchRes), @@ -133,30 +139,30 @@ func (b *CommitBatcher) batchWait(maxWait, slack time.Duration) <-chan time.Time return nil } - var deadline time.Time + var cutoff time.Time for sn := range b.todo { - sectorDeadline := b.deadlines[sn] - if deadline.IsZero() || (!sectorDeadline.IsZero() && sectorDeadline.Before(deadline)) { - deadline = sectorDeadline + sectorCutoff := b.cutoffs[sn] + if cutoff.IsZero() || (!sectorCutoff.IsZero() && sectorCutoff.Before(cutoff)) { + cutoff = sectorCutoff } } for sn := range b.waiting { - sectorDeadline := b.deadlines[sn] - if deadline.IsZero() || (!sectorDeadline.IsZero() && sectorDeadline.Before(deadline)) { - deadline = sectorDeadline + sectorCutoff := b.cutoffs[sn] + if cutoff.IsZero() || (!sectorCutoff.IsZero() && sectorCutoff.Before(cutoff)) { + cutoff = sectorCutoff } } - if deadline.IsZero() { + if cutoff.IsZero() { return time.After(maxWait) } - deadline = deadline.Add(-slack) - if deadline.Before(now) { + cutoff = cutoff.Add(-slack) + if cutoff.Before(now) { return time.After(time.Nanosecond) // can't return 0 } - wait := deadline.Sub(now) + wait := cutoff.Sub(now) if wait > maxWait { wait = maxWait } @@ -209,7 +215,7 @@ func (b *CommitBatcher) maybeStartBatch(notif, after bool) ([]sealiface.CommitBa delete(b.waiting, sn) delete(b.todo, sn) - delete(b.deadlines, sn) + delete(b.cutoffs, sn) } } @@ -287,7 +293,21 @@ func (b *CommitBatcher) processBatch(cfg sealiface.Config) ([]sealiface.CommitBa } maxFee := b.feeCfg.MaxCommitBatchGasFee.FeeForSectors(len(infos)) - goodFunds := big.Add(maxFee, collateral) + + bf, err := b.api.ChainBaseFee(b.mctx, tok) + if err != nil { + return []sealiface.CommitBatchRes{res}, xerrors.Errorf("couldn't get base fee: %w", err) + } + + nv, err := b.api.StateNetworkVersion(b.mctx, tok) + if err != nil { + log.Errorf("getting network version: %s", err) + return []sealiface.CommitBatchRes{res}, xerrors.Errorf("getting network version: %s", err) + } + + aggFee := policy.AggregateNetworkFee(nv, len(infos), bf) + + goodFunds := big.Add(maxFee, big.Add(collateral, aggFee)) from, _, err := b.addrSel(b.mctx, mi, api.CommitAddr, goodFunds, collateral) if err != nil { @@ -371,16 +391,15 @@ func (b *CommitBatcher) processSingle(mi miner.MinerInfo, sn abi.SectorNumber, i // register commit, wait for batch message, return message CID func (b *CommitBatcher) AddCommit(ctx context.Context, s SectorInfo, in AggregateInput) (res sealiface.CommitBatchRes, err error) { - _, curEpoch, err := b.api.ChainHead(b.mctx) - if err != nil { - log.Errorf("getting chain head: %s", err) - return sealiface.CommitBatchRes{}, nil - } - sn := s.SectorNumber + cu, err := b.getCommitCutoff(s) + if err != nil { + return sealiface.CommitBatchRes{}, err + } + b.lk.Lock() - b.deadlines[sn] = getSectorDeadline(curEpoch, s) + b.cutoffs[sn] = cu b.todo[sn] = in sent := make(chan sealiface.CommitBatchRes, 1) @@ -454,24 +473,43 @@ func (b *CommitBatcher) Stop(ctx context.Context) error { } } -func getSectorDeadline(curEpoch abi.ChainEpoch, si SectorInfo) time.Time { - deadlineEpoch := si.TicketEpoch + policy.MaxPreCommitRandomnessLookback +// TODO: If this returned epochs, it would make testing much easier +func (b *CommitBatcher) getCommitCutoff(si SectorInfo) (time.Time, error) { + tok, curEpoch, err := b.api.ChainHead(b.mctx) + if err != nil { + return time.Now(), xerrors.Errorf("getting chain head: %s", err) + } + + nv, err := b.api.StateNetworkVersion(b.mctx, tok) + if err != nil { + log.Errorf("getting network version: %s", err) + return time.Now(), xerrors.Errorf("getting network version: %s", err) + } + + pci, err := b.api.StateSectorPreCommitInfo(b.mctx, b.maddr, si.SectorNumber, tok) + if err != nil { + log.Errorf("getting precommit info: %s", err) + return time.Now(), err + } + + cutoffEpoch := pci.PreCommitEpoch + policy.GetMaxProveCommitDuration(actors.VersionForNetwork(nv), si.SectorType) + for _, p := range si.Pieces { if p.DealInfo == nil { continue } startEpoch := p.DealInfo.DealSchedule.StartEpoch - if startEpoch < deadlineEpoch { - deadlineEpoch = startEpoch + if startEpoch < cutoffEpoch { + cutoffEpoch = startEpoch } } - if deadlineEpoch <= curEpoch { - return time.Now() + if cutoffEpoch <= curEpoch { + return time.Now(), nil } - return time.Now().Add(time.Duration(deadlineEpoch-curEpoch) * time.Duration(build.BlockDelaySecs) * time.Second) + return time.Now().Add(time.Duration(cutoffEpoch-curEpoch) * time.Duration(build.BlockDelaySecs) * time.Second), nil } func (b *CommitBatcher) getSectorCollateral(sn abi.SectorNumber, tok TipSetToken) (abi.TokenAmount, error) { diff --git a/extern/storage-sealing/precommit_batch.go b/extern/storage-sealing/precommit_batch.go index 7dffd848b..9439ae14c 100644 --- a/extern/storage-sealing/precommit_batch.go +++ b/extern/storage-sealing/precommit_batch.go @@ -7,6 +7,9 @@ import ( "sync" "time" + "github.com/filecoin-project/lotus/build" + "github.com/filecoin-project/lotus/chain/actors/policy" + "github.com/ipfs/go-cid" "golang.org/x/xerrors" @@ -41,9 +44,9 @@ type PreCommitBatcher struct { feeCfg config.MinerFeeConfig getConfig GetSealingConfigFunc - deadlines map[abi.SectorNumber]time.Time - todo map[abi.SectorNumber]*preCommitEntry - waiting map[abi.SectorNumber][]chan sealiface.PreCommitBatchRes + cutoffs map[abi.SectorNumber]time.Time + todo map[abi.SectorNumber]*preCommitEntry + waiting map[abi.SectorNumber][]chan sealiface.PreCommitBatchRes notify, stop, stopped chan struct{} force chan chan []sealiface.PreCommitBatchRes @@ -59,9 +62,9 @@ func NewPreCommitBatcher(mctx context.Context, maddr address.Address, api PreCom feeCfg: feeCfg, getConfig: getConfig, - deadlines: map[abi.SectorNumber]time.Time{}, - todo: map[abi.SectorNumber]*preCommitEntry{}, - waiting: map[abi.SectorNumber][]chan sealiface.PreCommitBatchRes{}, + cutoffs: map[abi.SectorNumber]time.Time{}, + todo: map[abi.SectorNumber]*preCommitEntry{}, + waiting: map[abi.SectorNumber][]chan sealiface.PreCommitBatchRes{}, notify: make(chan struct{}, 1), force: make(chan chan []sealiface.PreCommitBatchRes), @@ -121,30 +124,30 @@ func (b *PreCommitBatcher) batchWait(maxWait, slack time.Duration) <-chan time.T return nil } - var deadline time.Time + var cutoff time.Time for sn := range b.todo { - sectorDeadline := b.deadlines[sn] - if deadline.IsZero() || (!sectorDeadline.IsZero() && sectorDeadline.Before(deadline)) { - deadline = sectorDeadline + sectorCutoff := b.cutoffs[sn] + if cutoff.IsZero() || (!sectorCutoff.IsZero() && sectorCutoff.Before(cutoff)) { + cutoff = sectorCutoff } } for sn := range b.waiting { - sectorDeadline := b.deadlines[sn] - if deadline.IsZero() || (!sectorDeadline.IsZero() && sectorDeadline.Before(deadline)) { - deadline = sectorDeadline + sectorCutoff := b.cutoffs[sn] + if cutoff.IsZero() || (!sectorCutoff.IsZero() && sectorCutoff.Before(cutoff)) { + cutoff = sectorCutoff } } - if deadline.IsZero() { + if cutoff.IsZero() { return time.After(maxWait) } - deadline = deadline.Add(-slack) - if deadline.Before(now) { + cutoff = cutoff.Add(-slack) + if cutoff.Before(now) { return time.After(time.Nanosecond) // can't return 0 } - wait := deadline.Sub(now) + wait := cutoff.Sub(now) if wait > maxWait { wait = maxWait } @@ -192,7 +195,7 @@ func (b *PreCommitBatcher) maybeStartBatch(notif, after bool) ([]sealiface.PreCo delete(b.waiting, sn) delete(b.todo, sn) - delete(b.deadlines, sn) + delete(b.cutoffs, sn) } } @@ -256,7 +259,7 @@ func (b *PreCommitBatcher) AddPreCommit(ctx context.Context, s SectorInfo, depos sn := s.SectorNumber b.lk.Lock() - b.deadlines[sn] = getSectorDeadline(curEpoch, s) + b.cutoffs[sn] = getPreCommitCutoff(curEpoch, s) b.todo[sn] = &preCommitEntry{ deposit: deposit, pci: in, @@ -332,3 +335,24 @@ func (b *PreCommitBatcher) Stop(ctx context.Context) error { return ctx.Err() } } + +// TODO: If this returned epochs, it would make testing much easier +func getPreCommitCutoff(curEpoch abi.ChainEpoch, si SectorInfo) time.Time { + cutoffEpoch := si.TicketEpoch + policy.MaxPreCommitRandomnessLookback + for _, p := range si.Pieces { + if p.DealInfo == nil { + continue + } + + startEpoch := p.DealInfo.DealSchedule.StartEpoch + if startEpoch < cutoffEpoch { + cutoffEpoch = startEpoch + } + } + + if cutoffEpoch <= curEpoch { + return time.Now() + } + + return time.Now().Add(time.Duration(cutoffEpoch-curEpoch) * time.Duration(build.BlockDelaySecs) * time.Second) +} diff --git a/extern/storage-sealing/sealing.go b/extern/storage-sealing/sealing.go index dae90d91b..2d53223db 100644 --- a/extern/storage-sealing/sealing.go +++ b/extern/storage-sealing/sealing.go @@ -67,6 +67,7 @@ type SealingAPI interface { StateMinerPartitions(ctx context.Context, m address.Address, dlIdx uint64, tok TipSetToken) ([]api.Partition, error) SendMsg(ctx context.Context, from, to address.Address, method abi.MethodNum, value, maxFee abi.TokenAmount, params []byte) (cid.Cid, error) ChainHead(ctx context.Context) (TipSetToken, abi.ChainEpoch, error) + ChainBaseFee(context.Context, TipSetToken) (abi.TokenAmount, error) ChainGetMessage(ctx context.Context, mc cid.Cid) (*types.Message, error) ChainGetRandomnessFromBeacon(ctx context.Context, tok TipSetToken, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte) (abi.Randomness, error) ChainGetRandomnessFromTickets(ctx context.Context, tok TipSetToken, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte) (abi.Randomness, error) diff --git a/storage/adapter_storage_miner.go b/storage/adapter_storage_miner.go index 1376ae5fb..895e7846d 100644 --- a/storage/adapter_storage_miner.go +++ b/storage/adapter_storage_miner.go @@ -360,6 +360,20 @@ func (s SealingAPIAdapter) ChainHead(ctx context.Context) (sealing.TipSetToken, return head.Key().Bytes(), head.Height(), nil } +func (s SealingAPIAdapter) ChainBaseFee(ctx context.Context, tok sealing.TipSetToken) (abi.TokenAmount, error) { + tsk, err := types.TipSetKeyFromBytes(tok) + if err != nil { + return big.Zero(), err + } + + ts, err := s.delegate.ChainGetTipSet(ctx, tsk) + if err != nil { + return big.Zero(), err + } + + return ts.Blocks()[0].ParentBaseFee, nil +} + func (s SealingAPIAdapter) ChainGetMessage(ctx context.Context, mc cid.Cid) (*types.Message, error) { return s.delegate.ChainGetMessage(ctx, mc) }