cerc-io/lotus
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge pull request #10647 from filecoin-project/sbansal/split-pcb

Browse Source 
feat: sealing: Split PCA/PCB batches if gas used exceeds block limit
This commit is contained in:
Łukasz Magiera 2023-04-21 18:30:27 +02:00 committed by GitHub
commit b44ae9a7d8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 228 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
node/impl/full/gas.go
View File

@ -388,6 +388,7 @@ func (m *GasModule) GasEstimateMessageGas(ctx context.Context, msg *types.Messag
return nil, err
}
msg.GasLimit = int64(float64(gasLimit) * m.Mpool.GetConfig().GasLimitOverestimation)
// Gas overestimation can cause us to exceed the block gas limit, cap it.
if msg.GasLimit > build.BlockGasLimit {
msg.GasLimit = build.BlockGasLimit

55
storage/pipeline/commit_batch.go
View File

@ -37,6 +37,7 @@ var aggFeeDen = big.NewInt(100)
type CommitBatcherApi interface {
MpoolPushMessage(context.Context, *types.Message, *api.MessageSendSpec) (*types.SignedMessage, error)
GasEstimateMessageGas(context.Context, *types.Message, *api.MessageSendSpec, types.TipSetKey) (*types.Message, error)
StateMinerInfo(context.Context, address.Address, types.TipSetKey) (api.MinerInfo, error)
ChainHead(ctx context.Context) (*types.TipSet, error)
@ -234,7 +235,11 @@ func (b *CommitBatcher) maybeStartBatch(notif bool) ([]sealiface.CommitBatchRes,
if individual {
res, err = b.processIndividually(cfg)
} else {
res, err = b.processBatch(cfg)
var sectors []abi.SectorNumber
for sn := range b.todo {
sectors = append(sectors, sn)
}
res, err = b.processBatch(cfg, sectors)
}
if err != nil {
@ -264,13 +269,13 @@ func (b *CommitBatcher) maybeStartBatch(notif bool) ([]sealiface.CommitBatchRes,
return res, nil
}
func (b *CommitBatcher) processBatch(cfg sealiface.Config) ([]sealiface.CommitBatchRes, error) {
func (b *CommitBatcher) processBatch(cfg sealiface.Config, sectors []abi.SectorNumber) ([]sealiface.CommitBatchRes, error) {
ts, err := b.api.ChainHead(b.mctx)
if err != nil {
return nil, err
}
total := len(b.todo)
total := len(sectors)
res := sealiface.CommitBatchRes{
FailedSectors: map[abi.SectorNumber]string{},
@ -284,24 +289,19 @@ func (b *CommitBatcher) processBatch(cfg sealiface.Config) ([]sealiface.CommitBa
infos := make([]proof.AggregateSealVerifyInfo, 0, total)
collateral := big.Zero()
for id, p := range b.todo {
if len(infos) >= cfg.MaxCommitBatch {
log.Infow("commit batch full")
break
}
for _, sector := range sectors {
res.Sectors = append(res.Sectors, sector)
res.Sectors = append(res.Sectors, id)
sc, err := b.getSectorCollateral(id, ts.Key())
sc, err := b.getSectorCollateral(sector, ts.Key())
if err != nil {
res.FailedSectors[id] = err.Error()
res.FailedSectors[sector] = err.Error()
continue
}
collateral = big.Add(collateral, sc)
params.SectorNumbers.Set(uint64(id))
infos = append(infos, p.Info)
params.SectorNumbers.Set(uint64(sector))
infos = append(infos, b.todo[sector].Info)
}
if len(infos) == 0 {
@ -318,17 +318,20 @@ func (b *CommitBatcher) processBatch(cfg sealiface.Config) ([]sealiface.CommitBa
mid, err := address.IDFromAddress(b.maddr)
if err != nil {
res.Error = err.Error()
return []sealiface.CommitBatchRes{res}, xerrors.Errorf("getting miner id: %w", err)
}
nv, err := b.api.StateNetworkVersion(b.mctx, ts.Key())
if err != nil {
res.Error = err.Error()
log.Errorf("getting network version: %s", err)
return []sealiface.CommitBatchRes{res}, xerrors.Errorf("getting network version: %s", err)
}
arp, err := b.aggregateProofType(nv)
if err != nil {
res.Error = err.Error()
return []sealiface.CommitBatchRes{res}, xerrors.Errorf("getting aggregate proof type: %w", err)
}
@ -339,16 +342,19 @@ func (b *CommitBatcher) processBatch(cfg sealiface.Config) ([]sealiface.CommitBa
Infos: infos,
}, proofs)
if err != nil {
res.Error = err.Error()
return []sealiface.CommitBatchRes{res}, xerrors.Errorf("aggregating proofs: %w", err)
}
enc := new(bytes.Buffer)
if err := params.MarshalCBOR(enc); err != nil {
res.Error = err.Error()
return []sealiface.CommitBatchRes{res}, xerrors.Errorf("couldn't serialize ProveCommitAggregateParams: %w", err)
}
mi, err := b.api.StateMinerInfo(b.mctx, b.maddr, types.EmptyTSK)
if err != nil {
res.Error = err.Error()
return []sealiface.CommitBatchRes{res}, xerrors.Errorf("couldn't get miner info: %w", err)
}
@ -356,6 +362,7 @@ func (b *CommitBatcher) processBatch(cfg sealiface.Config) ([]sealiface.CommitBa
aggFeeRaw, err := policy.AggregateProveCommitNetworkFee(nv, len(infos), ts.MinTicketBlock().ParentBaseFee)
if err != nil {
res.Error = err.Error()
log.Errorf("getting aggregate commit network fee: %s", err)
return []sealiface.CommitBatchRes{res}, xerrors.Errorf("getting aggregate commit network fee: %s", err)
}
@ -365,6 +372,7 @@ func (b *CommitBatcher) processBatch(cfg sealiface.Config) ([]sealiface.CommitBa
needFunds := big.Add(collateral, aggFee)
needFunds, err = collateralSendAmount(b.mctx, b.api, b.maddr, cfg, needFunds)
if err != nil {
res.Error = err.Error()
return []sealiface.CommitBatchRes{res}, err
}
@ -372,9 +380,28 @@ func (b *CommitBatcher) processBatch(cfg sealiface.Config) ([]sealiface.CommitBa
from, _, err := b.addrSel.AddressFor(b.mctx, b.api, mi, api.CommitAddr, goodFunds, needFunds)
if err != nil {
res.Error = err.Error()
return []sealiface.CommitBatchRes{res}, xerrors.Errorf("no good address found: %w", err)
}
_, err = simulateMsgGas(b.mctx, b.api, from, b.maddr, builtin.MethodsMiner.ProveCommitAggregate, needFunds, maxFee, enc.Bytes())
if err != nil && (!api.ErrorIsIn(err, []error{&api.ErrOutOfGas{}}) || len(sectors) < miner.MinAggregatedSectors*2) {
log.Errorf("simulating CommitBatch message failed: %s", err)
res.Error = err.Error()
return []sealiface.CommitBatchRes{res}, xerrors.Errorf("simulating CommitBatch message failed: %w", err)
}
// If we're out of gas, split the batch in half and evaluate again
if api.ErrorIsIn(err, []error{&api.ErrOutOfGas{}}) {
log.Warnf("CommitAggregate message ran out of gas, splitting batch in half and trying again (sectors: %d)", len(sectors))
mid := len(sectors) / 2
ret0, _ := b.processBatch(cfg, sectors[:mid])
ret1, _ := b.processBatch(cfg, sectors[mid:])
return append(ret0, ret1...), nil
}
mcid, err := sendMsg(b.mctx, b.api, from, b.maddr, builtin.MethodsMiner.ProveCommitAggregate, needFunds, maxFee, enc.Bytes())
if err != nil {
return []sealiface.CommitBatchRes{res}, xerrors.Errorf("sending message failed: %w", err)

52
storage/pipeline/commit_batch_test.go
View File

@ -201,7 +201,7 @@ func TestCommitBatcher(t *testing.T) {
if batch {
s.EXPECT().StateNetworkVersion(gomock.Any(), gomock.Any()).Return(network.Version13, nil)
//s.EXPECT().ChainBaseFee(gomock.Any(), gomock.Any()).Return(basefee, nil)
s.EXPECT().GasEstimateMessageGas(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&types.Message{GasLimit: 100000}, nil)
}
s.EXPECT().MpoolPushMessage(gomock.Any(), funMatcher(func(i interface{}) bool {
@ -224,6 +224,48 @@ func TestCommitBatcher(t *testing.T) {
}
}
expectProcessBatch := func(expect []abi.SectorNumber, aboveBalancer, failOnePCI bool, gasOverLimit bool) action {
return func(t *testing.T, s *mocks.MockCommitBatcherApi, pcb *pipeline.CommitBatcher) promise {
s.EXPECT().StateMinerInfo(gomock.Any(), gomock.Any(), gomock.Any()).Return(api.MinerInfo{Owner: t0123, Worker: t0123}, nil)
ti := len(expect)
batch := false
if ti >= minBatch {
batch = true
ti = 1
}
if !aboveBalancer {
batch = false
ti = len(expect)
}
pciC := len(expect)
if failOnePCI {
s.EXPECT().StateSectorPreCommitInfo(gomock.Any(), gomock.Any(), abi.SectorNumber(1), gomock.Any()).Return(nil, nil).Times(1) // not found
pciC = len(expect) - 1
if !batch {
ti--
}
}
s.EXPECT().StateSectorPreCommitInfo(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&minertypes.SectorPreCommitOnChainInfo{
PreCommitDeposit: big.Zero(),
}, nil).Times(pciC)
s.EXPECT().StateMinerInitialPledgeCollateral(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(big.Zero(), nil).Times(pciC)
if batch {
s.EXPECT().StateNetworkVersion(gomock.Any(), gomock.Any()).Return(network.Version18, nil)
if gasOverLimit {
s.EXPECT().GasEstimateMessageGas(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, &api.ErrOutOfGas{})
} else {
s.EXPECT().GasEstimateMessageGas(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&types.Message{GasLimit: 100000}, nil)
}
}
return nil
}
}
flush := func(expect []abi.SectorNumber, aboveBalancer, failOnePCI bool) action {
return func(t *testing.T, s *mocks.MockCommitBatcherApi, pcb *pipeline.CommitBatcher) promise {
_ = expectSend(expect, aboveBalancer, failOnePCI)(t, s, pcb)
@ -309,6 +351,14 @@ func TestCommitBatcher(t *testing.T) {
addSectors(getSectors(maxBatch), true),
},
},
"addMax-aboveBalancer-gasAboveLimit": {
actions: []action{
expectProcessBatch(getSectors(maxBatch), true, false, true),
expectSend(getSectors(maxBatch)[:maxBatch/2], true, false),
expectSend(getSectors(maxBatch)[maxBatch/2:], true, false),
addSectors(getSectors(maxBatch), true),
},
},
"addSingle-belowBalancer": {
actions: []action{
addSector(0, false),

15
storage/pipeline/mocks/api.go
View File

@ -94,6 +94,21 @@ func (mr *MockSealingAPIMockRecorder) ChainReadObj(arg0, arg1 interface{}) *gomo
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChainReadObj", reflect.TypeOf((*MockSealingAPI)(nil).ChainReadObj), arg0, arg1)
}
// GasEstimateMessageGas mocks base method.
func (m *MockSealingAPI) GasEstimateMessageGas(arg0 context.Context, arg1 *types.Message, arg2 *api.MessageSendSpec, arg3 types.TipSetKey) (*types.Message, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GasEstimateMessageGas", arg0, arg1, arg2, arg3)
ret0, _ := ret[0].(*types.Message)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GasEstimateMessageGas indicates an expected call of GasEstimateMessageGas.
func (mr *MockSealingAPIMockRecorder) GasEstimateMessageGas(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GasEstimateMessageGas", reflect.TypeOf((*MockSealingAPI)(nil).GasEstimateMessageGas), arg0, arg1, arg2, arg3)
}
// MpoolPushMessage mocks base method.
func (m *MockSealingAPI) MpoolPushMessage(arg0 context.Context, arg1 *types.Message, arg2 *api.MessageSendSpec) (*types.SignedMessage, error) {
m.ctrl.T.Helper()

15
storage/pipeline/mocks/mock_commit_batcher.go
View File

@ -58,6 +58,21 @@ func (mr *MockCommitBatcherApiMockRecorder) ChainHead(arg0 interface{}) *gomock.
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChainHead", reflect.TypeOf((*MockCommitBatcherApi)(nil).ChainHead), arg0)
}
// GasEstimateMessageGas mocks base method.
func (m *MockCommitBatcherApi) GasEstimateMessageGas(arg0 context.Context, arg1 *types.Message, arg2 *api.MessageSendSpec, arg3 types.TipSetKey) (*types.Message, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GasEstimateMessageGas", arg0, arg1, arg2, arg3)
ret0, _ := ret[0].(*types.Message)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GasEstimateMessageGas indicates an expected call of GasEstimateMessageGas.
func (mr *MockCommitBatcherApiMockRecorder) GasEstimateMessageGas(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GasEstimateMessageGas", reflect.TypeOf((*MockCommitBatcherApi)(nil).GasEstimateMessageGas), arg0, arg1, arg2, arg3)
}
// MpoolPushMessage mocks base method.
func (m *MockCommitBatcherApi) MpoolPushMessage(arg0 context.Context, arg1 *types.Message, arg2 *api.MessageSendSpec) (*types.SignedMessage, error) {
m.ctrl.T.Helper()

15
storage/pipeline/mocks/mock_precommit_batcher.go
View File

@ -58,6 +58,21 @@ func (mr *MockPreCommitBatcherApiMockRecorder) ChainHead(arg0 interface{}) *gomo
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChainHead", reflect.TypeOf((*MockPreCommitBatcherApi)(nil).ChainHead), arg0)
}
// GasEstimateMessageGas mocks base method.
func (m *MockPreCommitBatcherApi) GasEstimateMessageGas(arg0 context.Context, arg1 *types.Message, arg2 *api.MessageSendSpec, arg3 types.TipSetKey) (*types.Message, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GasEstimateMessageGas", arg0, arg1, arg2, arg3)
ret0, _ := ret[0].(*types.Message)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GasEstimateMessageGas indicates an expected call of GasEstimateMessageGas.
func (mr *MockPreCommitBatcherApiMockRecorder) GasEstimateMessageGas(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GasEstimateMessageGas", reflect.TypeOf((*MockPreCommitBatcherApi)(nil).GasEstimateMessageGas), arg0, arg1, arg2, arg3)
}
// MpoolPushMessage mocks base method.
func (m *MockPreCommitBatcherApi) MpoolPushMessage(arg0 context.Context, arg1 *types.Message, arg2 *api.MessageSendSpec) (*types.SignedMessage, error) {
m.ctrl.T.Helper()

49
storage/pipeline/precommit_batch.go
View File

@ -31,6 +31,7 @@ import (
type PreCommitBatcherApi interface {
MpoolPushMessage(context.Context, *types.Message, *api.MessageSendSpec) (*types.SignedMessage, error)
GasEstimateMessageGas(context.Context, *types.Message, *api.MessageSendSpec, types.TipSetKey) (*types.Message, error)
StateMinerInfo(context.Context, address.Address, types.TipSetKey) (api.MinerInfo, error)
StateMinerAvailableBalance(context.Context, address.Address, types.TipSetKey) (big.Int, error)
ChainHead(ctx context.Context) (*types.TipSet, error)
@ -319,17 +320,12 @@ func (b *PreCommitBatcher) processSingle(cfg sealiface.Config, mi api.MinerInfo,
return mcid, nil
}
func (b *PreCommitBatcher) processBatch(cfg sealiface.Config, tsk types.TipSetKey, bf abi.TokenAmount, nv network.Version) ([]sealiface.PreCommitBatchRes, error) {
func (b *PreCommitBatcher) processPreCommitBatch(cfg sealiface.Config, bf abi.TokenAmount, entries []*preCommitEntry, nv network.Version) ([]sealiface.PreCommitBatchRes, error) {
params := miner.PreCommitSectorBatchParams{}
deposit := big.Zero()
var res sealiface.PreCommitBatchRes
for _, p := range b.todo {
if len(params.Sectors) >= cfg.MaxPreCommitBatch {
log.Infow("precommit batch full")
break
}
for _, p := range entries {
res.Sectors = append(res.Sectors, p.pci.SectorNumber)
params.Sectors = append(params.Sectors, *infoToPreCommitSectorParams(p.pci))
deposit = big.Add(deposit, p.deposit)
@ -337,11 +333,13 @@ func (b *PreCommitBatcher) processBatch(cfg sealiface.Config, tsk types.TipSetKe
enc := new(bytes.Buffer)
if err := params.MarshalCBOR(enc); err != nil {
res.Error = err.Error()
return []sealiface.PreCommitBatchRes{res}, xerrors.Errorf("couldn't serialize PreCommitSectorBatchParams: %w", err)
}
mi, err := b.api.StateMinerInfo(b.mctx, b.maddr, types.EmptyTSK)
if err != nil {
res.Error = err.Error()
return []sealiface.PreCommitBatchRes{res}, xerrors.Errorf("couldn't get miner info: %w", err)
}
@ -350,6 +348,7 @@ func (b *PreCommitBatcher) processBatch(cfg sealiface.Config, tsk types.TipSetKe
aggFeeRaw, err := policy.AggregatePreCommitNetworkFee(nv, len(params.Sectors), bf)
if err != nil {
log.Errorf("getting aggregate precommit network fee: %s", err)
res.Error = err.Error()
return []sealiface.PreCommitBatchRes{res}, xerrors.Errorf("getting aggregate precommit network fee: %s", err)
}
@ -368,18 +367,42 @@ func (b *PreCommitBatcher) processBatch(cfg sealiface.Config, tsk types.TipSetKe
return []sealiface.PreCommitBatchRes{res}, xerrors.Errorf("no good address found: %w", err)
}
mcid, err := sendMsg(b.mctx, b.api, from, b.maddr, builtin.MethodsMiner.PreCommitSectorBatch, needFunds, maxFee, enc.Bytes())
if err != nil {
return []sealiface.PreCommitBatchRes{res}, xerrors.Errorf("sending message failed: %w", err)
_, err = simulateMsgGas(b.mctx, b.api, from, b.maddr, builtin.MethodsMiner.PreCommitSectorBatch, needFunds, maxFee, enc.Bytes())
if err != nil && (!api.ErrorIsIn(err, []error{&api.ErrOutOfGas{}}) || len(entries) == 1) {
res.Error = err.Error()
return []sealiface.PreCommitBatchRes{res}, xerrors.Errorf("simulating PreCommitBatch message failed: %w", err)
}
// If we're out of gas, split the batch in half and evaluate again
if api.ErrorIsIn(err, []error{&api.ErrOutOfGas{}}) {
log.Warnf("PreCommitBatch out of gas, splitting batch in half and trying again")
mid := len(entries) / 2
ret0, _ := b.processPreCommitBatch(cfg, bf, entries[:mid], nv)
ret1, _ := b.processPreCommitBatch(cfg, bf, entries[mid:], nv)
return append(ret0, ret1...), nil
}
// If state call succeeds, we can send the message for real
mcid, err := sendMsg(b.mctx, b.api, from, b.maddr, builtin.MethodsMiner.PreCommitSectorBatch, needFunds, maxFee, enc.Bytes())
if err != nil {
res.Error = err.Error()
return []sealiface.PreCommitBatchRes{res}, xerrors.Errorf("pushing message to mpool: %w", err)
}
res.Msg = &mcid
log.Infow("Sent PreCommitSectorBatch message", "cid", mcid, "from", from, "sectors", len(b.todo))
return []sealiface.PreCommitBatchRes{res}, nil
}
func (b *PreCommitBatcher) processBatch(cfg sealiface.Config, tsk types.TipSetKey, bf abi.TokenAmount, nv network.Version) ([]sealiface.PreCommitBatchRes, error) {
var pcEntries []*preCommitEntry
for _, p := range b.todo {
pcEntries = append(pcEntries, p)
}
return b.processPreCommitBatch(cfg, bf, pcEntries, nv)
}
// register PreCommit, wait for batch message, return message CID
func (b *PreCommitBatcher) AddPreCommit(ctx context.Context, s SectorInfo, deposit abi.TokenAmount, in *miner.SectorPreCommitInfo) (res sealiface.PreCommitBatchRes, err error) {
ts, err := b.api.ChainHead(b.mctx)

52
storage/pipeline/precommit_batch_test.go
View File

@ -156,21 +156,34 @@ func TestPrecommitBatcher(t *testing.T) {
}
//stm: @CHAIN_STATE_MINER_INFO_001, @CHAIN_STATE_NETWORK_VERSION_001
expectSend := func(expect []abi.SectorNumber) action {
expectSend := func(expect []abi.SectorNumber, gasOverLimit bool) action {
return func(t *testing.T, s *mocks.MockPreCommitBatcherApi, pcb *pipeline.PreCommitBatcher) promise {
s.EXPECT().StateMinerInfo(gomock.Any(), gomock.Any(), gomock.Any()).Return(api.MinerInfo{Owner: t0123, Worker: t0123}, nil)
if gasOverLimit {
s.EXPECT().GasEstimateMessageGas(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, &api.ErrOutOfGas{})
} else {
s.EXPECT().GasEstimateMessageGas(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&types.Message{GasLimit: 100000}, nil)
}
if !gasOverLimit {
s.EXPECT().MpoolPushMessage(gomock.Any(), funMatcher(func(i interface{}) bool {
b := i.(*types.Message)
var params miner6.PreCommitSectorBatchParams
require.NoError(t, params.UnmarshalCBOR(bytes.NewReader(b.Params)))
for s, number := range expect {
require.Equal(t, number, params.Sectors[s].SectorNumber)
}
return true
}), gomock.Any()).Return(dummySmsg, nil)
}
return nil
}
}
expectInitialCalls := func() action {
return func(t *testing.T, s *mocks.MockPreCommitBatcherApi, pcb *pipeline.PreCommitBatcher) promise {
s.EXPECT().ChainHead(gomock.Any()).Return(makeBFTs(t, big.NewInt(10001), 1), nil)
s.EXPECT().StateNetworkVersion(gomock.Any(), gomock.Any()).Return(network.Version14, nil)
s.EXPECT().StateMinerInfo(gomock.Any(), gomock.Any(), gomock.Any()).Return(api.MinerInfo{Owner: t0123, Worker: t0123}, nil)
s.EXPECT().MpoolPushMessage(gomock.Any(), funMatcher(func(i interface{}) bool {
b := i.(*types.Message)
var params miner6.PreCommitSectorBatchParams
require.NoError(t, params.UnmarshalCBOR(bytes.NewReader(b.Params)))
for s, number := range expect {
require.Equal(t, number, params.Sectors[s].SectorNumber)
}
return true
}), gomock.Any()).Return(dummySmsg, nil)
return nil
}
}
@ -198,7 +211,8 @@ func TestPrecommitBatcher(t *testing.T) {
flush := func(expect []abi.SectorNumber) action {
return func(t *testing.T, s *mocks.MockPreCommitBatcherApi, pcb *pipeline.PreCommitBatcher) promise {
_ = expectSend(expect)(t, s, pcb)
_ = expectInitialCalls()(t, s, pcb)
_ = expectSend(expect, false)(t, s, pcb)
r, err := pcb.Flush(ctx)
require.NoError(t, err)
@ -240,7 +254,17 @@ func TestPrecommitBatcher(t *testing.T) {
},
"addMax": {
actions: []action{
expectSend(getSectors(maxBatch)),
expectInitialCalls(),
expectSend(getSectors(maxBatch), false),
addSectors(getSectors(maxBatch), true),
},
},
"addMax-gasAboveLimit": {
actions: []action{
expectInitialCalls(),
expectSend(getSectors(maxBatch), true),
expectSend(getSectors(maxBatch)[:maxBatch/2], false),
expectSend(getSectors(maxBatch)[maxBatch/2:], false),
addSectors(getSectors(maxBatch), true),
},
},

1
storage/pipeline/sealing.go
View File

@ -64,6 +64,7 @@ type SealingAPI interface {
StateMinerDeadlines(context.Context, address.Address, types.TipSetKey) ([]api.Deadline, error)
StateMinerPartitions(ctx context.Context, m address.Address, dlIdx uint64, tsk types.TipSetKey) ([]api.Partition, error)
MpoolPushMessage(context.Context, *types.Message, *api.MessageSendSpec) (*types.SignedMessage, error)
GasEstimateMessageGas(context.Context, *types.Message, *api.MessageSendSpec, types.TipSetKey) (*types.Message, error)
ChainHead(ctx context.Context) (*types.TipSet, error)
ChainGetMessage(ctx context.Context, mc cid.Cid) (*types.Message, error)
StateGetRandomnessFromBeacon(ctx context.Context, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte, tsk types.TipSetKey) (abi.Randomness, error)

15
storage/pipeline/utils.go
View File

@ -94,6 +94,21 @@ func collateralSendAmount(ctx context.Context, api interface {
return collateral, nil
}
func simulateMsgGas(ctx context.Context, sa interface {
GasEstimateMessageGas(context.Context, *types.Message, *api.MessageSendSpec, types.TipSetKey) (*types.Message, error)
},
from, to address.Address, method abi.MethodNum, value, maxFee abi.TokenAmount, params []byte) (*types.Message, error) {
msg := types.Message{
To: to,
From: from,
Value: value,
Method: method,
Params: params,
}
return sa.GasEstimateMessageGas(ctx, &msg, nil, types.EmptyTSK)
}
func sendMsg(ctx context.Context, sa interface {
MpoolPushMessage(context.Context, *types.Message, *api.MessageSendSpec) (*types.SignedMessage, error)
}, from, to address.Address, method abi.MethodNum, value, maxFee abi.TokenAmount, params []byte) (cid.Cid, error) {