refactor(lotus-sim): enterprise grade

While the previous version "worked", this version nicely separates out
the state for the separate stages. Hopefully, we'll be able to use this
to build different pipelines with different configs.

cmd/lotus-sim/run.go

@@ -35,12 +35,6 @@ Signals:
 		if err != nil {
 			return err
 		}
-		fmt.Fprintln(cctx.App.Writer, "loading simulation")
-		err = sim.Load(cctx.Context)
-		if err != nil {
-			return err
-		}
-		fmt.Fprintln(cctx.App.Writer, "running simulation")
 		targetEpochs := cctx.Int("epochs")
 
 		ch := make(chan os.Signal, 1)

cmd/lotus-sim/simulation/block.go

@@ -8,6 +8,7 @@ import (
 
 	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/types"
 )
@@ -68,7 +69,7 @@ func (sim *Simulation) makeTipSet(ctx context.Context, messages []*types.Message
 		ParentStateRoot:       parentState,
 		ParentMessageReceipts: parentRec,
 		Messages:              msgsCid,
-		ParentBaseFee:         baseFee,
+		ParentBaseFee:         abi.NewTokenAmount(0),
 		Timestamp:             uts,
 		ElectionProof:         &types.ElectionProof{WinCount: 1},
 	}}

cmd/lotus-sim/simulation/blockbuilder/blockbuilder.go

@@ -0,0 +1,279 @@
+package blockbuilder
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/network"
+
+	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/actors"
+	"github.com/filecoin-project/lotus/chain/actors/adt"
+	"github.com/filecoin-project/lotus/chain/actors/builtin"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/account"
+	"github.com/filecoin-project/lotus/chain/state"
+	"github.com/filecoin-project/lotus/chain/stmgr"
+	"github.com/filecoin-project/lotus/chain/store"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/chain/vm"
+)
+
+const (
+	// The number of expected blocks in a tipset. We use this to determine how much gas a tipset
+	// has.
+	expectedBlocks = 5
+	// TODO: This will produce invalid blocks but it will accurately model the amount of gas
+	// we're willing to use per-tipset.
+	// A more correct approach would be to produce 5 blocks. We can do that later.
+	targetGas = build.BlockGasTarget * expectedBlocks
+)
+
+type BlockBuilder struct {
+	ctx    context.Context
+	logger *zap.SugaredLogger
+
+	parentTs *types.TipSet
+	parentSt *state.StateTree
+	vm       *vm.VM
+	sm       *stmgr.StateManager
+
+	gasTotal int64
+	messages []*types.Message
+}
+
+// NewBlockBuilder constructs a new block builder from the parent state. Use this to pack a block
+// with messages.
+//
+// NOTE: The context applies to the life of the block builder itself (but does not need to be canceled).
+func NewBlockBuilder(ctx context.Context, logger *zap.SugaredLogger, sm *stmgr.StateManager, parentTs *types.TipSet) (*BlockBuilder, error) {
+	parentState, _, err := sm.TipSetState(ctx, parentTs)
+	if err != nil {
+		return nil, err
+	}
+	parentSt, err := sm.StateTree(parentState)
+	if err != nil {
+		return nil, err
+	}
+
+	bb := &BlockBuilder{
+		ctx:      ctx,
+		logger:   logger.With("epoch", parentTs.Height()+1),
+		sm:       sm,
+		parentTs: parentTs,
+		parentSt: parentSt,
+	}
+
+	// Then we construct a VM to execute messages for gas estimation.
+	//
+	// Most parts of this VM are "real" except:
+	// 1. We don't charge a fee.
+	// 2. The runtime has "fake" proof logic.
+	// 3. We don't actually save any of the results.
+	r := store.NewChainRand(sm.ChainStore(), parentTs.Cids())
+	vmopt := &vm.VMOpts{
+		StateBase:      parentState,
+		Epoch:          parentTs.Height() + 1,
+		Rand:           r,
+		Bstore:         sm.ChainStore().StateBlockstore(),
+		Syscalls:       sm.ChainStore().VMSys(),
+		CircSupplyCalc: sm.GetVMCirculatingSupply,
+		NtwkVersion:    sm.GetNtwkVersion,
+		BaseFee:        abi.NewTokenAmount(0),
+		LookbackState:  stmgr.LookbackStateGetterForTipset(sm, parentTs),
+	}
+	bb.vm, err = vm.NewVM(bb.ctx, vmopt)
+	if err != nil {
+		return nil, err
+	}
+	return bb, nil
+}
+
+// PushMessages tries to push the specified message into the block.
+//
+// 1. All messages will be executed in-order.
+// 2. Gas computation & nonce selection will be handled internally.
+// 3. The base-fee is 0 so the sender does not need funds.
+// 4. As usual, the sender must be an account (any account).
+// 5. If the message fails to execute, this method will fail.
+//
+// Returns ErrOutOfGas when out of gas. Check BlockBuilder.GasRemaining and try pushing a cheaper
+// message.
+func (bb *BlockBuilder) PushMessage(msg *types.Message) (*types.MessageReceipt, error) {
+	if bb.gasTotal >= targetGas {
+		return nil, new(ErrOutOfGas)
+	}
+
+	st := bb.StateTree()
+	store := bb.ActorStore()
+
+	// Copy the message before we start mutating it.
+	msgCpy := *msg
+	msg = &msgCpy
+
+	actor, err := st.GetActor(msg.From)
+	if err != nil {
+		return nil, err
+	}
+	if !builtin.IsAccountActor(actor.Code) {
+		return nil, xerrors.Errorf(
+			"messags may only be sent from account actors, got message from %s (%s)",
+			msg.From, builtin.ActorNameByCode(actor.Code),
+		)
+	}
+	msg.Nonce = actor.Nonce
+	if msg.From.Protocol() == address.ID {
+		state, err := account.Load(store, actor)
+		if err != nil {
+			return nil, err
+		}
+		msg.From, err = state.PubkeyAddress()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// TODO: Our gas estimation is broken for payment channels due to horrible hacks in
+	// gasEstimateGasLimit.
+	if msg.Value == types.EmptyInt {
+		msg.Value = abi.NewTokenAmount(0)
+	}
+	msg.GasPremium = abi.NewTokenAmount(0)
+	msg.GasFeeCap = abi.NewTokenAmount(0)
+	msg.GasLimit = build.BlockGasLimit
+
+	// We manually snapshot so we can revert nonce changes, etc. on failure.
+	st.Snapshot(bb.ctx)
+	defer st.ClearSnapshot()
+
+	ret, err := bb.vm.ApplyMessage(bb.ctx, msg)
+	if err != nil {
+		_ = st.Revert()
+		return nil, err
+	}
+	if ret.ActorErr != nil {
+		_ = st.Revert()
+		return nil, ret.ActorErr
+	}
+
+	// Sometimes there are bugs. Let's catch them.
+	if ret.GasUsed == 0 {
+		_ = st.Revert()
+		return nil, xerrors.Errorf("used no gas",
+			"msg", msg,
+			"ret", ret,
+		)
+	}
+
+	// TODO: consider applying overestimation? We're likely going to "over pack" here by
+	// ~25% because we're too accurate.
+
+	// Did we go over? Yes, revert.
+	newTotal := bb.gasTotal + ret.GasUsed
+	if newTotal > targetGas {
+		_ = st.Revert()
+		return nil, &ErrOutOfGas{Available: targetGas - bb.gasTotal, Required: ret.GasUsed}
+	}
+	bb.gasTotal = newTotal
+
+	// Update the gas limit.
+	msg.GasLimit = ret.GasUsed
+
+	bb.messages = append(bb.messages, msg)
+	return &ret.MessageReceipt, nil
+}
+
+// ActorStore returns the VM's current (pending) blockstore.
+func (bb *BlockBuilder) ActorStore() adt.Store {
+	return bb.vm.ActorStore(bb.ctx)
+}
+
+// StateTree returns the VM's current (pending) state-tree. This includes any changes made by
+// successfully pushed messages.
+//
+// You probably want ParentStateTree
+func (bb *BlockBuilder) StateTree() *state.StateTree {
+	return bb.vm.StateTree().(*state.StateTree)
+}
+
+// ParentStateTree returns the parent state-tree (not the paren't tipset's parent state-tree).
+func (bb *BlockBuilder) ParentStateTree() *state.StateTree {
+	return bb.parentSt
+}
+
+// StateTreeByHeight will return a state-tree up through and including the current in-progress
+// epoch.
+//
+// NOTE: This will return the state after the given epoch, not the parent state for the epoch.
+func (bb *BlockBuilder) StateTreeByHeight(epoch abi.ChainEpoch) (*state.StateTree, error) {
+	now := bb.Height()
+	if epoch > now {
+		return nil, xerrors.Errorf(
+			"cannot load state-tree from future: %d > %d", epoch, bb.Height(),
+		)
+	} else if epoch <= 0 {
+		return nil, xerrors.Errorf(
+			"cannot load state-tree: epoch %d <= 0", epoch,
+		)
+	}
+
+	// Manually handle "now" and "previous".
+	switch epoch {
+	case now:
+		return bb.StateTree(), nil
+	case now - 1:
+		return bb.ParentStateTree(), nil
+	}
+
+	// Get the tipset of the block _after_ the target epoch so we can use its parent state.
+	targetTs, err := bb.sm.ChainStore().GetTipsetByHeight(bb.ctx, epoch+1, bb.parentTs, false)
+	if err != nil {
+		return nil, err
+	}
+
+	return bb.sm.StateTree(targetTs.ParentState())
+}
+
+// Messages returns all messages currently packed into the next block.
+// 1. DO NOT modify the slice, copy it.
+// 2. DO NOT retain the slice, copy it.
+func (bb *BlockBuilder) Messages() []*types.Message {
+	return bb.messages
+}
+
+// GasRemaining returns the amount of remaining gas in the next block.
+func (bb *BlockBuilder) GasRemaining() int64 {
+	return targetGas - bb.gasTotal
+}
+
+// ParentTipSet returns the parent tipset.
+func (bb *BlockBuilder) ParentTipSet() *types.TipSet {
+	return bb.parentTs
+}
+
+// Height returns the epoch for the target block.
+func (bb *BlockBuilder) Height() abi.ChainEpoch {
+	return bb.parentTs.Height() + 1
+}
+
+// NetworkVersion returns the network version for the target block.
+func (bb *BlockBuilder) NetworkVersion() network.Version {
+	return bb.sm.GetNtwkVersion(bb.ctx, bb.Height())
+}
+
+// StateManager returns the stmgr.StateManager.
+func (bb *BlockBuilder) StateManager() *stmgr.StateManager {
+	return bb.sm
+}
+
+// ActorsVersion returns the actors version for the target block.
+func (bb *BlockBuilder) ActorsVersion() actors.Version {
+	return actors.VersionForNetwork(bb.NetworkVersion())
+}
+
+func (bb *BlockBuilder) L() *zap.SugaredLogger {
+	return bb.logger
+}

cmd/lotus-sim/simulation/blockbuilder/errors.go

@@ -0,0 +1,25 @@
+package blockbuilder
+
+import (
+	"errors"
+	"fmt"
+)
+
+// ErrOutOfGas is returned from BlockBuilder.PushMessage when the block does not have enough gas to
+// fit the given message.
+type ErrOutOfGas struct {
+	Available, Required int64
+}
+
+func (e *ErrOutOfGas) Error() string {
+	if e.Available == 0 {
+		return "out of gas: block full"
+	}
+	return fmt.Sprintf("out of gas: %d < %d", e.Required, e.Available)
+}
+
+// IsOutOfGas returns true if the error is an "out of gas" error.
+func IsOutOfGas(err error) bool {
+	var oog *ErrOutOfGas
+	return errors.As(err, &oog)
+}

cmd/lotus-sim/simulation/mock/mock.go

@@ -1,4 +1,4 @@
-package simulation
+package mock
 
 import (
 	"bytes"
@@ -8,8 +8,11 @@ import (
 
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/ipfs/go-cid"
 
+	miner5 "github.com/filecoin-project/specs-actors/v5/actors/builtin/miner"
 	proof5 "github.com/filecoin-project/specs-actors/v5/actors/runtime/proof"
+	tutils "github.com/filecoin-project/specs-actors/v5/support/testing"
 
 	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
 )
@@ -26,14 +29,14 @@ const (
 // mockVerifier is a simple mock for verifying "fake" proofs.
 type mockVerifier struct{}
 
-var _ ffiwrapper.Verifier = mockVerifier{}
+var Verifier ffiwrapper.Verifier = mockVerifier{}
 
 func (mockVerifier) VerifySeal(proof proof5.SealVerifyInfo) (bool, error) {
 	addr, err := address.NewIDAddress(uint64(proof.Miner))
 	if err != nil {
 		return false, err
 	}
-	mockProof, err := mockSealProof(proof.SealProof, addr)
+	mockProof, err := MockSealProof(proof.SealProof, addr)
 	if err != nil {
 		return false, err
 	}
@@ -45,7 +48,7 @@ func (mockVerifier) VerifyAggregateSeals(aggregate proof5.AggregateSealVerifyPro
 	if err != nil {
 		return false, err
 	}
-	mockProof, err := mockAggregateSealProof(aggregate.SealProof, addr, len(aggregate.Infos))
+	mockProof, err := MockAggregateSealProof(aggregate.SealProof, addr, len(aggregate.Infos))
 	if err != nil {
 		return false, err
 	}
@@ -63,7 +66,7 @@ func (mockVerifier) VerifyWindowPoSt(ctx context.Context, info proof5.WindowPoSt
 	if err != nil {
 		return false, err
 	}
-	mockProof, err := mockWpostProof(proof.PoStProof, addr)
+	mockProof, err := MockWindowPoStProof(proof.PoStProof, addr)
 	if err != nil {
 		return false, err
 	}
@@ -74,8 +77,8 @@ func (mockVerifier) GenerateWinningPoStSectorChallenge(context.Context, abi.Regi
 	panic("should not be called")
 }
 
-// mockSealProof generates a mock "seal" proof tied to the specified proof type and the given miner.
-func mockSealProof(proofType abi.RegisteredSealProof, minerAddr address.Address) ([]byte, error) {
+// MockSealProof generates a mock "seal" proof tied to the specified proof type and the given miner.
+func MockSealProof(proofType abi.RegisteredSealProof, minerAddr address.Address) ([]byte, error) {
 	plen, err := proofType.ProofSize()
 	if err != nil {
 		return nil, err
@@ -88,9 +91,9 @@ func mockSealProof(proofType abi.RegisteredSealProof, minerAddr address.Address)
 	return proof, nil
 }
 
-// mockAggregateSealProof generates a mock "seal" aggregate proof tied to the specified proof type,
+// MockAggregateSealProof generates a mock "seal" aggregate proof tied to the specified proof type,
 // the given miner, and the number of proven sectors.
-func mockAggregateSealProof(proofType abi.RegisteredSealProof, minerAddr address.Address, count int) ([]byte, error) {
+func MockAggregateSealProof(proofType abi.RegisteredSealProof, minerAddr address.Address, count int) ([]byte, error) {
 	proof := make([]byte, aggProofLen(count))
 	i := copy(proof, mockAggregateSealProofPrefix)
 	binary.BigEndian.PutUint64(proof[i:], uint64(proofType))
@@ -102,9 +105,9 @@ func mockAggregateSealProof(proofType abi.RegisteredSealProof, minerAddr address
 	return proof, nil
 }
 
-// mockWpostProof generates a mock "window post" proof tied to the specified proof type, and the
+// MockWindowPoStProof generates a mock "window post" proof tied to the specified proof type, and the
 // given miner.
-func mockWpostProof(proofType abi.RegisteredPoStProof, minerAddr address.Address) ([]byte, error) {
+func MockWindowPoStProof(proofType abi.RegisteredPoStProof, minerAddr address.Address) ([]byte, error) {
 	plen, err := proofType.ProofSize()
 	if err != nil {
 		return nil, err
@@ -115,6 +118,11 @@ func mockWpostProof(proofType abi.RegisteredPoStProof, minerAddr address.Address
 	return proof, nil
 }
 
+// makeCommR generates a "fake" but valid CommR for a sector. It is unique for the given sector/miner.
+func MockCommR(minerAddr address.Address, sno abi.SectorNumber) cid.Cid {
+	return tutils.MakeCID(fmt.Sprintf("%s:%d", minerAddr, sno), &miner5.SealedCIDPrefix)
+}
+
 // TODO: dedup
 func aggProofLen(nproofs int) int {
 	switch {

cmd/lotus-sim/simulation/node.go

@@ -16,6 +16,8 @@ import (
 	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/vm"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/mock"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/stages"
 	"github.com/filecoin-project/lotus/node/repo"
 )
 
@@ -53,7 +55,7 @@ func OpenNode(ctx context.Context, path string) (*Node, error) {
 		return nil, err
 	}
 
-	node.Chainstore = store.NewChainStore(node.Blockstore, node.Blockstore, node.MetadataDS, vm.Syscalls(mockVerifier{}), nil)
+	node.Chainstore = store.NewChainStore(node.Blockstore, node.Blockstore, node.MetadataDS, vm.Syscalls(mock.Verifier), nil)
 	return &node, nil
 }
 
@@ -74,12 +76,16 @@ func (nd *Node) Close() error {
 
 // LoadSim loads
 func (nd *Node) LoadSim(ctx context.Context, name string) (*Simulation, error) {
+	stages, err := stages.DefaultPipeline()
+	if err != nil {
+		return nil, err
+	}
 	sim := &Simulation{
 		Node:   nd,
 		name:   name,
+		stages: stages,
 	}
 
-	var err error
 	sim.head, err = sim.loadNamedTipSet("head")
 	if err != nil {
 		return nil, err
@@ -113,10 +119,15 @@ func (nd *Node) CreateSim(ctx context.Context, name string, head *types.TipSet)
 	if strings.Contains(name, "/") {
 		return nil, xerrors.Errorf("simulation name %q cannot contain a '/'", name)
 	}
+	stages, err := stages.DefaultPipeline()
+	if err != nil {
+		return nil, err
+	}
 	sim := &Simulation{
 		name:         name,
 		Node:         nd,
 		StateManager: stmgr.NewStateManager(nd.Chainstore),
+		stages:       stages,
 	}
 	if has, err := nd.MetadataDS.Has(sim.key("head")); err != nil {
 		return nil, err

cmd/lotus-sim/simulation/power.go

@@ -1,58 +0,0 @@
-package simulation
-
-import (
-	"context"
-
-	"golang.org/x/xerrors"
-
-	"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/lotus/chain/actors/builtin/power"
-)
-
-// Load all power claims at the given height.
-func (sim *Simulation) loadClaims(ctx context.Context, height abi.ChainEpoch) (map[address.Address]power.Claim, error) {
-	powerTable := make(map[address.Address]power.Claim)
-	store := sim.Chainstore.ActorStore(ctx)
-
-	ts, err := sim.Chainstore.GetTipsetByHeight(ctx, height, sim.head, true)
-	if err != nil {
-		return nil, xerrors.Errorf("when projecting growth, failed to lookup lookback epoch: %w", err)
-	}
-
-	powerActor, err := sim.StateManager.LoadActor(ctx, power.Address, ts)
-	if err != nil {
-		return nil, err
-	}
-
-	powerState, err := power.Load(store, powerActor)
-	if err != nil {
-		return nil, err
-	}
-	err = powerState.ForEachClaim(func(miner address.Address, claim power.Claim) error {
-		// skip miners without power
-		if claim.RawBytePower.IsZero() {
-			return nil
-		}
-		powerTable[miner] = claim
-		return nil
-	})
-	if err != nil {
-		return nil, err
-	}
-	return powerTable, nil
-}
-
-// Compute the number of sectors a miner has from their power claim.
-func sectorsFromClaim(sectorSize abi.SectorSize, c power.Claim) int64 {
-	if c.RawBytePower.Int == nil {
-		return 0
-	}
-	sectorCount := big.Div(c.RawBytePower, big.NewIntUnsigned(uint64(sectorSize)))
-	if !sectorCount.IsInt64() {
-		panic("impossible number of sectors")
-	}
-	return sectorCount.Int64()
-}

cmd/lotus-sim/simulation/precommit.go

@@ -1,233 +0,0 @@
-package simulation
-
-import (
-	"context"
-	"fmt"
-	"time"
-
-	"golang.org/x/xerrors"
-
-	"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/network"
-	"github.com/ipfs/go-cid"
-
-	miner5 "github.com/filecoin-project/specs-actors/v5/actors/builtin/miner"
-	tutils "github.com/filecoin-project/specs-actors/v5/support/testing"
-
-	"github.com/filecoin-project/lotus/chain/actors"
-	"github.com/filecoin-project/lotus/chain/actors/aerrors"
-	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
-	"github.com/filecoin-project/lotus/chain/actors/policy"
-	"github.com/filecoin-project/lotus/chain/types"
-)
-
-var (
-	targetFunds = abi.TokenAmount(types.MustParseFIL("1000FIL"))
-	minFunds    = abi.TokenAmount(types.MustParseFIL("100FIL"))
-)
-
-// makeCommR generates a "fake" but valid CommR for a sector. It is unique for the given sector/miner.
-func makeCommR(minerAddr address.Address, sno abi.SectorNumber) cid.Cid {
-	return tutils.MakeCID(fmt.Sprintf("%s:%d", minerAddr, sno), &miner5.SealedCIDPrefix)
-}
-
-// packPreCommits packs pre-commit messages until the block is full.
-func (ss *simulationState) packPreCommits(ctx context.Context, cb packFunc) (_err error) {
-	var (
-		full                             bool
-		top1Count, top10Count, restCount int
-	)
-	start := time.Now()
-	defer func() {
-		if _err != nil {
-			return
-		}
-		log.Debugw("packed pre commits",
-			"done", top1Count+top10Count+restCount,
-			"top1", top1Count,
-			"top10", top10Count,
-			"rest", restCount,
-			"filled-block", full,
-			"duration", time.Since(start),
-		)
-	}()
-
-	var top1Miners, top10Miners, restMiners int
-	for i := 0; ; i++ {
-		var (
-			minerAddr address.Address
-			count     *int
-		)
-
-		// We pre-commit for the top 1%, 10%, and the of the network 1/3rd of the time each.
-		// This won't yeild the most accurate distribution... but it'll give us a good
-		// enough distribution.
-
-		// NOTE: We submit at most _one_ 819 sector batch per-miner per-block. See the
-		// comment on packPreCommitsMiner for why. We should fix this.
-		switch {
-		case (i%3) <= 0 && top1Miners < ss.minerDist.top1.len():
-			count = &top1Count
-			minerAddr = ss.minerDist.top1.next()
-			top1Miners++
-		case (i%3) <= 1 && top10Miners < ss.minerDist.top10.len():
-			count = &top10Count
-			minerAddr = ss.minerDist.top10.next()
-			top10Miners++
-		case (i%3) <= 2 && restMiners < ss.minerDist.rest.len():
-			count = &restCount
-			minerAddr = ss.minerDist.rest.next()
-			restMiners++
-		default:
-			// Well, we've run through all miners.
-			return nil
-		}
-
-		var (
-			added int
-			err   error
-		)
-		added, full, err = ss.packPreCommitsMiner(ctx, cb, minerAddr, maxProveCommitBatchSize)
-		if err != nil {
-			return xerrors.Errorf("failed to pack precommits for miner %s: %w", minerAddr, err)
-		}
-		*count += added
-		if full {
-			return nil
-		}
-	}
-}
-
-// packPreCommitsMiner packs count pre-commits for the given miner. This should only be called once
-// per-miner, per-epoch to avoid packing multiple pre-commits with the same sector numbers.
-func (ss *simulationState) packPreCommitsMiner(ctx context.Context, cb packFunc, minerAddr address.Address, count int) (int, bool, error) {
-	// Load everything.
-	epoch := ss.nextEpoch()
-	nv := ss.StateManager.GetNtwkVersion(ctx, epoch)
-	actor, minerState, err := ss.getMinerState(ctx, minerAddr)
-	if err != nil {
-		return 0, false, err
-	}
-
-	minerInfo, err := ss.getMinerInfo(ctx, minerAddr)
-	if err != nil {
-		return 0, false, err
-	}
-
-	// Make sure the miner is funded.
-	minerBalance, err := minerState.AvailableBalance(actor.Balance)
-	if err != nil {
-		return 0, false, err
-	}
-
-	if big.Cmp(minerBalance, minFunds) < 0 {
-		err := fund(cb, minerAddr, 1)
-		if err != nil {
-			if err == ErrOutOfGas {
-				return 0, true, nil
-			}
-			return 0, false, err
-		}
-	}
-
-	// Generate pre-commits.
-	sealType, err := miner.PreferredSealProofTypeFromWindowPoStType(
-		nv, minerInfo.WindowPoStProofType,
-	)
-	if err != nil {
-		return 0, false, err
-	}
-
-	sectorNos, err := minerState.UnallocatedSectorNumbers(count)
-	if err != nil {
-		return 0, false, err
-	}
-
-	expiration := epoch + policy.GetMaxSectorExpirationExtension()
-	infos := make([]miner.SectorPreCommitInfo, len(sectorNos))
-	for i, sno := range sectorNos {
-		infos[i] = miner.SectorPreCommitInfo{
-			SealProof:     sealType,
-			SectorNumber:  sno,
-			SealedCID:     makeCommR(minerAddr, sno),
-			SealRandEpoch: epoch - 1,
-			Expiration:    expiration,
-		}
-	}
-
-	// Commit the pre-commits.
-	added := 0
-	if nv >= network.Version13 {
-		targetBatchSize := maxPreCommitBatchSize
-		for targetBatchSize >= minPreCommitBatchSize && len(infos) >= minPreCommitBatchSize {
-			batch := infos
-			if len(batch) > targetBatchSize {
-				batch = batch[:targetBatchSize]
-			}
-			params := miner5.PreCommitSectorBatchParams{
-				Sectors: batch,
-			}
-			enc, err := actors.SerializeParams(&params)
-			if err != nil {
-				return added, false, err
-			}
-			// NOTE: just in-case, sendAndFund will "fund" and re-try for any message
-			// that fails due to "insufficient funds".
-			if _, err := sendAndFund(cb, &types.Message{
-				To:     minerAddr,
-				From:   minerInfo.Worker,
-				Value:  abi.NewTokenAmount(0),
-				Method: miner.Methods.PreCommitSectorBatch,
-				Params: enc,
-			}); err == ErrOutOfGas {
-				// try again with a smaller batch.
-				targetBatchSize /= 2
-				continue
-			} else if aerr, ok := err.(aerrors.ActorError); ok && !aerr.IsFatal() {
-				// Log the error and move on. No reason to stop.
-				log.Errorw("failed to pre-commit for unknown reasons",
-					"error", aerr,
-					"miner", minerAddr,
-					"sectors", batch,
-					"epoch", ss.nextEpoch(),
-				)
-				return added, false, nil
-			} else if err != nil {
-				return added, false, err
-			}
-
-			for _, info := range batch {
-				if err := ss.commitQueue.enqueueProveCommit(minerAddr, epoch, info); err != nil {
-					return added, false, err
-				}
-				added++
-			}
-			infos = infos[len(batch):]
-		}
-	}
-	for _, info := range infos {
-		enc, err := actors.SerializeParams(&info)
-		if err != nil {
-			return 0, false, err
-		}
-		if _, err := sendAndFund(cb, &types.Message{
-			To:     minerAddr,
-			From:   minerInfo.Worker,
-			Value:  abi.NewTokenAmount(0),
-			Method: miner.Methods.PreCommitSector,
-			Params: enc,
-		}); err == ErrOutOfGas {
-			return added, true, nil
-		} else if err != nil {
-			return added, false, err
-		}
-
-		if err := ss.commitQueue.enqueueProveCommit(minerAddr, epoch, info); err != nil {
-			return added, false, err
-		}
-		added++
-	}
-	return added, false, nil
-}

cmd/lotus-sim/simulation/simulation.go

@@ -15,28 +15,21 @@ import (
 	logging "github.com/ipfs/go-log/v2"
 
 	blockadt "github.com/filecoin-project/specs-actors/actors/util/adt"
-	miner5 "github.com/filecoin-project/specs-actors/v5/actors/builtin/miner"
 
 	"github.com/filecoin-project/lotus/chain/actors/builtin"
 	"github.com/filecoin-project/lotus/chain/state"
 	"github.com/filecoin-project/lotus/chain/stmgr"
 	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/stages"
 )
 
 var log = logging.Logger("simulation")
 
 const onboardingProjectionLookback = 2 * 7 * builtin.EpochsInDay // lookback two weeks
 
-const (
-	minPreCommitBatchSize   = 1
-	maxPreCommitBatchSize   = miner5.PreCommitSectorBatchMaxSize
-	minProveCommitBatchSize = 4
-	maxProveCommitBatchSize = miner5.MaxAggregatedSectors
-)
-
 // config is the simulation's config, persisted to the local metadata store and loaded on start.
 //
-// See simulationState.loadConfig and simulationState.saveConfig.
+// See Simulation.loadConfig and Simulation.saveConfig.
 type config struct {
 	Upgrades map[network.Version]abi.ChainEpoch
 }
@@ -93,9 +86,7 @@ type Simulation struct {
 	st   *state.StateTree
 	head *types.TipSet
 
-	// lazy-loaded state
-	// access through `simState(ctx)` to load on-demand.
-	state *simulationState
+	stages []stages.Stage
 }
 
 // loadConfig loads a simulation's config from the datastore. This must be called on startup and may
@@ -141,21 +132,6 @@ func (sim *Simulation) stateTree(ctx context.Context) (*state.StateTree, error)
 	return sim.st, nil
 }
 
-// Loads the simulation state. The state is memoized so this will be fast except the first time.
-func (sim *Simulation) simState(ctx context.Context) (*simulationState, error) {
-	if sim.state == nil {
-		log.Infow("loading simulation")
-		state, err := loadSimulationState(ctx, sim)
-		if err != nil {
-			return nil, xerrors.Errorf("failed to load simulation state: %w", err)
-		}
-		sim.state = state
-		log.Infow("simulation loaded", "miners", len(sim.state.minerInfos))
-	}
-
-	return sim.state, nil
-}
-
 var simulationPrefix = datastore.NewKey("/simulation")
 
 // key returns the the key in the form /simulation/<subkey>/<simulation-name>. For example,
@@ -189,13 +165,6 @@ func (sim *Simulation) storeNamedTipSet(name string, ts *types.TipSet) error {
 	return nil
 }
 
-// Load loads the simulation state. This will happen automatically on first use, but it can be
-// useful to preload for timing reasons.
-func (sim *Simulation) Load(ctx context.Context) error {
-	_, err := sim.simState(ctx)
-	return err
-}
-
 // GetHead returns the current simulation head.
 func (sim *Simulation) GetHead() *types.TipSet {
 	return sim.head

cmd/lotus-sim/simulation/stages/actor_iter.go

@@ -1,4 +1,4 @@
-package simulation
+package stages
 
 import (
 	"math/rand"

cmd/lotus-sim/simulation/stages/commit_queue.go

@@ -1,4 +1,4 @@
-package simulation
+package stages
 
 import (
 	"sort"

cmd/lotus-sim/simulation/stages/commit_queue_test.go

@@ -1,4 +1,4 @@
-package simulation
+package stages
 
 import (
 	"testing"

cmd/lotus-sim/simulation/stages/funding_stage.go

@@ -1,4 +1,4 @@
-package simulation
+package stages
 
 import (
 	"bytes"
@@ -13,41 +13,44 @@ import (
 	"github.com/filecoin-project/go-state-types/big"
 	"github.com/filecoin-project/go-state-types/exitcode"
 
-	"github.com/filecoin-project/lotus/chain/actors"
 	"github.com/filecoin-project/lotus/chain/actors/aerrors"
 	"github.com/filecoin-project/lotus/chain/actors/builtin"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/multisig"
 	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/blockbuilder"
 )
 
 var (
-	fundAccount = func() address.Address {
-		addr, err := address.NewIDAddress(100)
-		if err != nil {
-			panic(err)
-		}
-		return addr
-	}()
-	minFundAcctFunds = abi.TokenAmount(types.MustParseFIL("1000000FIL"))
-	maxFundAcctFunds = abi.TokenAmount(types.MustParseFIL("100000000FIL"))
-	taxMin           = abi.TokenAmount(types.MustParseFIL("1000FIL"))
+	TargetFunds  = abi.TokenAmount(types.MustParseFIL("1000FIL"))
+	MinimumFunds = abi.TokenAmount(types.MustParseFIL("100FIL"))
 )
 
-func fund(send packFunc, target address.Address, times int) error {
-	amt := targetFunds
-	if times >= 1 {
-		if times >= 8 {
-			times = 8 // cap
+type FundingStage struct {
+	fundAccount        address.Address
+	taxMin             abi.TokenAmount
+	minFunds, maxFunds abi.TokenAmount
 }
-		amt = big.Lsh(amt, uint(times))
+
+func NewFundingStage() (*FundingStage, error) {
+	// TODO: make all this configurable.
+	addr, err := address.NewIDAddress(100)
+	if err != nil {
+		return nil, err
 	}
-	_, err := send(&types.Message{
-		From:   fundAccount,
-		To:     target,
-		Value:  amt,
-		Method: builtin.MethodSend,
-	})
-	return err
+	return &FundingStage{
+		fundAccount: addr,
+		taxMin:      abi.TokenAmount(types.MustParseFIL("1000FIL")),
+		minFunds:    abi.TokenAmount(types.MustParseFIL("1000000FIL")),
+		maxFunds:    abi.TokenAmount(types.MustParseFIL("100000000FIL")),
+	}, nil
+}
+
+func (*FundingStage) Name() string {
+	return "funding"
+}
+
+func (fs *FundingStage) Fund(bb *blockbuilder.BlockBuilder, target address.Address) error {
+	return fs.fund(bb, target, 0)
 }
 
 // sendAndFund "packs" the given message, funding the actor if necessary. It:
@@ -56,9 +59,9 @@ func fund(send packFunc, target address.Address, times int) error {
 // 2. If that fails, it checks to see if the exit code was ErrInsufficientFunds.
 // 3. If so, it sends 1K FIL from the "burnt funds actor" (because we need to send it from
 //    somewhere) and re-tries the message.0
-func sendAndFund(send packFunc, msg *types.Message) (res *types.MessageReceipt, err error) {
+func (fs *FundingStage) SendAndFund(bb *blockbuilder.BlockBuilder, msg *types.Message) (res *types.MessageReceipt, err error) {
 	for i := 0; i < 10; i++ {
-		res, err = send(msg)
+		res, err = bb.PushMessage(msg)
 		if err == nil {
 			return res, nil
 		}
@@ -68,8 +71,8 @@ func sendAndFund(send packFunc, msg *types.Message) (res *types.MessageReceipt,
 		}
 
 		// Ok, insufficient funds. Let's fund this miner and try again.
-		if err := fund(send, msg.To, i); err != nil {
-			if err != ErrOutOfGas {
+		if err := fs.fund(bb, msg.To, i); err != nil {
+			if !blockbuilder.IsOutOfGas(err) {
 				err = xerrors.Errorf("failed to fund %s: %w", msg.To, err)
 			}
 			return nil, err
@@ -78,16 +81,30 @@ func sendAndFund(send packFunc, msg *types.Message) (res *types.MessageReceipt,
 	return res, err
 }
 
-func (ss *simulationState) packFunding(ctx context.Context, cb packFunc) (_err error) {
-	st, err := ss.stateTree(ctx)
+func (fs *FundingStage) fund(bb *blockbuilder.BlockBuilder, target address.Address, times int) error {
+	amt := TargetFunds
+	if times > 0 {
+		if times >= 8 {
+			times = 8 // cap
+		}
+		amt = big.Lsh(amt, uint(times))
+	}
+	_, err := bb.PushMessage(&types.Message{
+		From:   fs.fundAccount,
+		To:     target,
+		Value:  amt,
+		Method: builtin.MethodSend,
+	})
+	return err
+}
+
+func (fs *FundingStage) PackMessages(ctx context.Context, bb *blockbuilder.BlockBuilder) (_err error) {
+	st := bb.StateTree()
+	fundAccActor, err := st.GetActor(fs.fundAccount)
 	if err != nil {
 		return err
 	}
-	fundAccActor, err := st.GetActor(fundAccount)
-	if err != nil {
-		return err
-	}
-	if minFundAcctFunds.LessThan(fundAccActor.Balance) {
+	if fs.minFunds.LessThan(fundAccActor.Balance) {
 		return nil
 	}
 
@@ -102,10 +119,10 @@ func (ss *simulationState) packFunding(ctx context.Context, cb packFunc) (_err e
 	var targets []*actor
 	err = st.ForEach(func(addr address.Address, act *types.Actor) error {
 		// Don't steal from ourselves!
-		if addr == fundAccount {
+		if addr == fs.fundAccount {
 			return nil
 		}
-		if act.Balance.LessThan(taxMin) {
+		if act.Balance.LessThan(fs.taxMin) {
 			return nil
 		}
 		if !(builtin.IsAccountActor(act.Code) || builtin.IsMultisigActor(act.Code)) {
@@ -124,19 +141,16 @@ func (ss *simulationState) packFunding(ctx context.Context, cb packFunc) (_err e
 		return targets[i].Balance.GreaterThan(targets[j].Balance)
 	})
 
-	store := ss.Chainstore.ActorStore(ctx)
-
-	epoch := ss.nextEpoch()
-
-	nv := ss.StateManager.GetNtwkVersion(ctx, epoch)
-	actorsVersion := actors.VersionForNetwork(nv)
+	store := bb.ActorStore()
+	epoch := bb.Height()
+	actorsVersion := bb.ActorsVersion()
 
 	var accounts, multisigs int
 	defer func() {
 		if _err != nil {
 			return
 		}
-		log.Infow("finished funding the simulation",
+		bb.L().Infow("finished funding the simulation",
 			"duration", time.Since(start),
 			"targets", len(targets),
 			"epoch", epoch,
@@ -150,11 +164,11 @@ func (ss *simulationState) packFunding(ctx context.Context, cb packFunc) (_err e
 	for _, actor := range targets {
 		switch {
 		case builtin.IsAccountActor(actor.Code):
-			if _, err := cb(&types.Message{
+			if _, err := bb.PushMessage(&types.Message{
 				From:  actor.Address,
-				To:    fundAccount,
+				To:    fs.fundAccount,
 				Value: actor.Balance,
-			}); err == ErrOutOfGas {
+			}); blockbuilder.IsOutOfGas(err) {
 				return nil
 			} else if err != nil {
 				return err
@@ -172,7 +186,7 @@ func (ss *simulationState) packFunding(ctx context.Context, cb packFunc) (_err e
 			}
 
 			if threshold > 16 {
-				log.Debugw("ignoring multisig with high threshold",
+				bb.L().Debugw("ignoring multisig with high threshold",
 					"multisig", actor.Address,
 					"threshold", threshold,
 					"max", 16,
@@ -185,7 +199,7 @@ func (ss *simulationState) packFunding(ctx context.Context, cb packFunc) (_err e
 				return err
 			}
 
-			if locked.LessThan(taxMin) {
+			if locked.LessThan(fs.taxMin) {
 				continue // not worth it.
 			}
 
@@ -217,15 +231,15 @@ func (ss *simulationState) packFunding(ctx context.Context, cb packFunc) (_err e
 			var txnId uint64
 			{
 				msg, err := multisig.Message(actorsVersion, signers[0]).Propose(
-					actor.Address, fundAccount, available,
+					actor.Address, fs.fundAccount, available,
 					builtin.MethodSend, nil,
 				)
 				if err != nil {
 					return err
 				}
-				res, err := cb(msg)
+				res, err := bb.PushMessage(msg)
 				if err != nil {
-					if err == ErrOutOfGas {
+					if blockbuilder.IsOutOfGas(err) {
 						err = nil
 					}
 					return err
@@ -237,7 +251,7 @@ func (ss *simulationState) packFunding(ctx context.Context, cb packFunc) (_err e
 				}
 				if ret.Applied {
 					if !ret.Code.IsSuccess() {
-						log.Errorw("failed to tax multisig",
+						bb.L().Errorw("failed to tax multisig",
 							"multisig", actor.Address,
 							"exitcode", ret.Code,
 						)
@@ -252,9 +266,9 @@ func (ss *simulationState) packFunding(ctx context.Context, cb packFunc) (_err e
 				if err != nil {
 					return err
 				}
-				res, err := cb(msg)
+				res, err := bb.PushMessage(msg)
 				if err != nil {
-					if err == ErrOutOfGas {
+					if blockbuilder.IsOutOfGas(err) {
 						err = nil
 					}
 					return err
@@ -271,7 +285,7 @@ func (ss *simulationState) packFunding(ctx context.Context, cb packFunc) (_err e
 
 			}
 			if !ret.Applied {
-				log.Errorw("failed to apply multisig transaction",
+				bb.L().Errorw("failed to apply multisig transaction",
 					"multisig", actor.Address,
 					"txnid", txnId,
 					"signers", len(signers),
@@ -280,7 +294,7 @@ func (ss *simulationState) packFunding(ctx context.Context, cb packFunc) (_err e
 				continue
 			}
 			if !ret.Code.IsSuccess() {
-				log.Errorw("failed to tax multisig",
+				bb.L().Errorw("failed to tax multisig",
 					"multisig", actor.Address,
 					"txnid", txnId,
 					"exitcode", ret.Code,
@@ -292,7 +306,7 @@ func (ss *simulationState) packFunding(ctx context.Context, cb packFunc) (_err e
 			panic("impossible case")
 		}
 		balance = big.Int{Int: balance.Add(balance.Int, actor.Balance.Int)}
-		if balance.GreaterThanEqual(maxFundAcctFunds) {
+		if balance.GreaterThanEqual(fs.maxFunds) {
 			// There's no need to get greedy.
 			// Well, really, we're trying to avoid messing with state _too_ much.
 			return nil

cmd/lotus-sim/simulation/stages/interface.go

@@ -0,0 +1,27 @@
+package stages
+
+import (
+	"context"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-state-types/abi"
+
+	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/blockbuilder"
+)
+
+// Stage is a stage of the simulation. It's asked to pack messages for every block.
+type Stage interface {
+	Name() string
+	PackMessages(ctx context.Context, bb *blockbuilder.BlockBuilder) error
+}
+
+type Funding interface {
+	SendAndFund(*blockbuilder.BlockBuilder, *types.Message) (*types.MessageReceipt, error)
+	Fund(*blockbuilder.BlockBuilder, address.Address) error
+}
+
+type Committer interface {
+	EnqueueProveCommit(addr address.Address, preCommitEpoch abi.ChainEpoch, info miner.SectorPreCommitInfo) error
+}

cmd/lotus-sim/simulation/stages/pipeline.go

@@ -0,0 +1,31 @@
+package stages
+
+// DefaultPipeline returns the default stage pipeline. This pipeline.
+//
+// 1. Funds a "funding" actor, if necessary.
+// 2. Submits any ready window posts.
+// 3. Submits any ready prove commits.
+// 4. Submits pre-commits with the remaining gas.
+func DefaultPipeline() ([]Stage, error) {
+	// TODO: make this configurable. E.g., through DI?
+	// Ideally, we'd also be able to change priority, limit throughput (by limiting gas in the
+	// block builder, etc.
+	funding, err := NewFundingStage()
+	if err != nil {
+		return nil, err
+	}
+	wdpost, err := NewWindowPoStStage()
+	if err != nil {
+		return nil, err
+	}
+	provecommit, err := NewProveCommitStage(funding)
+	if err != nil {
+		return nil, err
+	}
+	precommit, err := NewPreCommitStage(funding, provecommit)
+	if err != nil {
+		return nil, err
+	}
+
+	return []Stage{funding, wdpost, provecommit, precommit}, nil
+}

cmd/lotus-sim/simulation/stages/precommit_stage.go

@@ -0,0 +1,359 @@
+package stages
+
+import (
+	"context"
+	"sort"
+	"time"
+
+	"golang.org/x/xerrors"
+
+	"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/network"
+
+	miner5 "github.com/filecoin-project/specs-actors/v5/actors/builtin/miner"
+
+	"github.com/filecoin-project/lotus/chain/actors"
+	"github.com/filecoin-project/lotus/chain/actors/aerrors"
+	"github.com/filecoin-project/lotus/chain/actors/builtin"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/power"
+	"github.com/filecoin-project/lotus/chain/actors/policy"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/blockbuilder"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/mock"
+)
+
+const (
+	minPreCommitBatchSize = 1
+	maxPreCommitBatchSize = miner5.PreCommitSectorBatchMaxSize
+)
+
+type PreCommitStage struct {
+	funding   Funding
+	committer Committer
+
+	// The tiers represent the top 1%, top 10%, and everyone else. When sealing sectors, we seal
+	// a group of sectors for the top 1%, a group (half that size) for the top 10%, and one
+	// sector for everyone else. We determine these rates by looking at two power tables.
+	// TODO Ideally we'd "learn" this distribution from the network. But this is good enough for
+	// now.
+	top1, top10, rest actorIter
+	initialized       bool
+}
+
+func NewPreCommitStage(funding Funding, committer Committer) (*PreCommitStage, error) {
+	return &PreCommitStage{
+		funding:   funding,
+		committer: committer,
+	}, nil
+}
+
+func (*PreCommitStage) Name() string {
+	return "pre-commit"
+}
+
+// packPreCommits packs pre-commit messages until the block is full.
+func (stage *PreCommitStage) PackMessages(ctx context.Context, bb *blockbuilder.BlockBuilder) (_err error) {
+	if !stage.initialized {
+		if err := stage.load(ctx, bb); err != nil {
+			return err
+		}
+	}
+
+	var (
+		full                             bool
+		top1Count, top10Count, restCount int
+	)
+	start := time.Now()
+	defer func() {
+		if _err != nil {
+			return
+		}
+		bb.L().Debugw("packed pre commits",
+			"done", top1Count+top10Count+restCount,
+			"top1", top1Count,
+			"top10", top10Count,
+			"rest", restCount,
+			"filled-block", full,
+			"duration", time.Since(start),
+		)
+	}()
+
+	var top1Miners, top10Miners, restMiners int
+	for i := 0; ; i++ {
+		var (
+			minerAddr address.Address
+			count     *int
+		)
+
+		// We pre-commit for the top 1%, 10%, and the of the network 1/3rd of the time each.
+		// This won't yeild the most accurate distribution... but it'll give us a good
+		// enough distribution.
+		switch {
+		case (i%3) <= 0 && top1Miners < stage.top1.len():
+			count = &top1Count
+			minerAddr = stage.top1.next()
+			top1Miners++
+		case (i%3) <= 1 && top10Miners < stage.top10.len():
+			count = &top10Count
+			minerAddr = stage.top10.next()
+			top10Miners++
+		case (i%3) <= 2 && restMiners < stage.rest.len():
+			count = &restCount
+			minerAddr = stage.rest.next()
+			restMiners++
+		default:
+			// Well, we've run through all miners.
+			return nil
+		}
+
+		var (
+			added int
+			err   error
+		)
+		added, full, err = stage.packMiner(ctx, bb, minerAddr, maxProveCommitBatchSize)
+		if err != nil {
+			return xerrors.Errorf("failed to pack precommits for miner %s: %w", minerAddr, err)
+		}
+		*count += added
+		if full {
+			return nil
+		}
+	}
+}
+
+// packPreCommitsMiner packs count pre-commits for the given miner.
+func (stage *PreCommitStage) packMiner(
+	ctx context.Context, bb *blockbuilder.BlockBuilder,
+	minerAddr address.Address, count int,
+) (int, bool, error) {
+	log := bb.L().With("miner", minerAddr)
+	epoch := bb.Height()
+	nv := bb.NetworkVersion()
+
+	minerActor, err := bb.StateTree().GetActor(minerAddr)
+	if err != nil {
+		return 0, false, err
+	}
+	minerState, err := miner.Load(bb.ActorStore(), minerActor)
+	if err != nil {
+		return 0, false, err
+	}
+
+	minerInfo, err := minerState.Info()
+	if err != nil {
+		return 0, false, err
+	}
+
+	// Make sure the miner is funded.
+	minerBalance, err := minerState.AvailableBalance(minerActor.Balance)
+	if err != nil {
+		return 0, false, err
+	}
+
+	if big.Cmp(minerBalance, MinimumFunds) < 0 {
+		err := stage.funding.Fund(bb, minerAddr)
+		if err != nil {
+			if blockbuilder.IsOutOfGas(err) {
+				return 0, true, nil
+			}
+			return 0, false, err
+		}
+	}
+
+	// Generate pre-commits.
+	sealType, err := miner.PreferredSealProofTypeFromWindowPoStType(
+		nv, minerInfo.WindowPoStProofType,
+	)
+	if err != nil {
+		return 0, false, err
+	}
+
+	sectorNos, err := minerState.UnallocatedSectorNumbers(count)
+	if err != nil {
+		return 0, false, err
+	}
+
+	expiration := epoch + policy.GetMaxSectorExpirationExtension()
+	infos := make([]miner.SectorPreCommitInfo, len(sectorNos))
+	for i, sno := range sectorNos {
+		infos[i] = miner.SectorPreCommitInfo{
+			SealProof:     sealType,
+			SectorNumber:  sno,
+			SealedCID:     mock.MockCommR(minerAddr, sno),
+			SealRandEpoch: epoch - 1,
+			Expiration:    expiration,
+		}
+	}
+
+	// Commit the pre-commits.
+	added := 0
+	if nv >= network.Version13 {
+		targetBatchSize := maxPreCommitBatchSize
+		for targetBatchSize >= minPreCommitBatchSize && len(infos) >= minPreCommitBatchSize {
+			batch := infos
+			if len(batch) > targetBatchSize {
+				batch = batch[:targetBatchSize]
+			}
+			params := miner5.PreCommitSectorBatchParams{
+				Sectors: batch,
+			}
+			enc, err := actors.SerializeParams(&params)
+			if err != nil {
+				return added, false, err
+			}
+			// NOTE: just in-case, sendAndFund will "fund" and re-try for any message
+			// that fails due to "insufficient funds".
+			if _, err := stage.funding.SendAndFund(bb, &types.Message{
+				To:     minerAddr,
+				From:   minerInfo.Worker,
+				Value:  abi.NewTokenAmount(0),
+				Method: miner.Methods.PreCommitSectorBatch,
+				Params: enc,
+			}); blockbuilder.IsOutOfGas(err) {
+				// try again with a smaller batch.
+				targetBatchSize /= 2
+				continue
+			} else if aerr, ok := err.(aerrors.ActorError); ok && !aerr.IsFatal() {
+				// Log the error and move on. No reason to stop.
+				log.Errorw("failed to pre-commit for unknown reasons",
+					"error", aerr,
+					"sectors", batch,
+				)
+				return added, false, nil
+			} else if err != nil {
+				return added, false, err
+			}
+
+			for _, info := range batch {
+				if err := stage.committer.EnqueueProveCommit(minerAddr, epoch, info); err != nil {
+					return added, false, err
+				}
+				added++
+			}
+			infos = infos[len(batch):]
+		}
+	}
+	for _, info := range infos {
+		enc, err := actors.SerializeParams(&info)
+		if err != nil {
+			return 0, false, err
+		}
+		if _, err := stage.funding.SendAndFund(bb, &types.Message{
+			To:     minerAddr,
+			From:   minerInfo.Worker,
+			Value:  abi.NewTokenAmount(0),
+			Method: miner.Methods.PreCommitSector,
+			Params: enc,
+		}); blockbuilder.IsOutOfGas(err) {
+			return added, true, nil
+		} else if err != nil {
+			return added, false, err
+		}
+
+		if err := stage.committer.EnqueueProveCommit(minerAddr, epoch, info); err != nil {
+			return added, false, err
+		}
+		added++
+	}
+	return added, false, nil
+}
+
+func (ps *PreCommitStage) load(ctx context.Context, bb *blockbuilder.BlockBuilder) (_err error) {
+	bb.L().Infow("loading miner power for pre-commits")
+	start := time.Now()
+	defer func() {
+		if _err != nil {
+			return
+		}
+		bb.L().Infow("loaded miner power for pre-commits",
+			"duration", time.Since(start),
+			"top1", ps.top1.len(),
+			"top10", ps.top10.len(),
+			"rest", ps.rest.len(),
+		)
+	}()
+	lookbackEpoch := bb.Height() - (14 * builtin.EpochsInDay)
+	lookbackPowerTable, err := loadClaims(ctx, bb, lookbackEpoch)
+	if err != nil {
+		return xerrors.Errorf("failed to load claims from lookback epoch %d: %w", lookbackEpoch, err)
+	}
+
+	store := bb.ActorStore()
+	st := bb.ParentStateTree()
+	powerState, err := loadPower(store, st)
+	if err != nil {
+		return xerrors.Errorf("failed to power actor: %w", err)
+	}
+
+	type onboardingInfo struct {
+		addr           address.Address
+		onboardingRate uint64
+	}
+	sealList := make([]onboardingInfo, 0, len(lookbackPowerTable))
+	err = powerState.ForEachClaim(func(addr address.Address, claim power.Claim) error {
+		if claim.RawBytePower.IsZero() {
+			return nil
+		}
+
+		minerState, err := loadMiner(store, st, addr)
+		if err != nil {
+			return err
+		}
+		info, err := minerState.Info()
+		if err != nil {
+			return err
+		}
+
+		sectorsAdded := sectorsFromClaim(info.SectorSize, claim)
+		if lookbackClaim, ok := lookbackPowerTable[addr]; !ok {
+			sectorsAdded -= sectorsFromClaim(info.SectorSize, lookbackClaim)
+		}
+
+		// NOTE: power _could_ have been lost, but that's too much of a pain to care
+		// about. We _could_ look for faulty power by iterating through all
+		// deadlines, but I'd rather not.
+		if sectorsAdded > 0 {
+			sealList = append(sealList, onboardingInfo{addr, uint64(sectorsAdded)})
+		}
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	if len(sealList) == 0 {
+		return xerrors.Errorf("simulation has no miners")
+	}
+
+	// Now that we have a list of sealing miners, sort them into percentiles.
+	sort.Slice(sealList, func(i, j int) bool {
+		return sealList[i].onboardingRate < sealList[j].onboardingRate
+	})
+
+	// reset, just in case.
+	ps.top1 = actorIter{}
+	ps.top10 = actorIter{}
+	ps.rest = actorIter{}
+
+	for i, oi := range sealList {
+		var dist *actorIter
+		if i < len(sealList)/100 {
+			dist = &ps.top1
+		} else if i < len(sealList)/10 {
+			dist = &ps.top10
+		} else {
+			dist = &ps.rest
+		}
+		dist.add(oi.addr)
+	}
+
+	ps.top1.shuffle()
+	ps.top10.shuffle()
+	ps.rest.shuffle()
+
+	ps.initialized = true
+	return nil
+}

cmd/lotus-sim/simulation/stages/provecommit_stage.go

@@ -1,4 +1,4 @@
-package simulation
+package stages
 
 import (
 	"context"
@@ -16,15 +16,50 @@ import (
 	"github.com/filecoin-project/lotus/chain/actors"
 	"github.com/filecoin-project/lotus/chain/actors/aerrors"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/power"
 	"github.com/filecoin-project/lotus/chain/actors/policy"
 	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/blockbuilder"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/mock"
 )
 
+const (
+	minProveCommitBatchSize = 4
+	maxProveCommitBatchSize = miner5.MaxAggregatedSectors
+)
+
+type ProveCommitStage struct {
+	funding Funding
+	// We track the set of pending commits. On simulation load, and when a new pre-commit is
+	// added to the chain, we put the commit in this queue. advanceEpoch(currentEpoch) should be
+	// called on this queue at every epoch before using it.
+	commitQueue commitQueue
+	initialized bool
+}
+
+func NewProveCommitStage(funding Funding) (*ProveCommitStage, error) {
+	return &ProveCommitStage{
+		funding: funding,
+	}, nil
+}
+
+func (*ProveCommitStage) Name() string {
+	return "prove-commit"
+}
+
+func (stage *ProveCommitStage) EnqueueProveCommit(
+	minerAddr address.Address, preCommitEpoch abi.ChainEpoch, info miner.SectorPreCommitInfo,
+) error {
+	return stage.commitQueue.enqueueProveCommit(minerAddr, preCommitEpoch, info)
+}
+
 // packProveCommits packs all prove-commits for all "ready to be proven" sectors until it fills the
 // block or runs out.
-func (ss *simulationState) packProveCommits(ctx context.Context, cb packFunc) (_err error) {
+func (stage *ProveCommitStage) PackMessages(ctx context.Context, bb *blockbuilder.BlockBuilder) (_err error) {
+	if !stage.initialized {
+	}
 	// Roll the commitQueue forward.
-	ss.commitQueue.advanceEpoch(ss.nextEpoch())
+	stage.commitQueue.advanceEpoch(bb.Height())
 
 	start := time.Now()
 	var failed, done, unbatched, count int
@@ -32,8 +67,8 @@ func (ss *simulationState) packProveCommits(ctx context.Context, cb packFunc) (_
 		if _err != nil {
 			return
 		}
-		remaining := ss.commitQueue.ready()
-		log.Debugw("packed prove commits",
+		remaining := stage.commitQueue.ready()
+		bb.L().Debugw("packed prove commits",
 			"remaining", remaining,
 			"done", done,
 			"failed", failed,
@@ -44,12 +79,12 @@ func (ss *simulationState) packProveCommits(ctx context.Context, cb packFunc) (_
 	}()
 
 	for {
-		addr, pending, ok := ss.commitQueue.nextMiner()
+		addr, pending, ok := stage.commitQueue.nextMiner()
 		if !ok {
 			return nil
 		}
 
-		res, err := ss.packProveCommitsMiner(ctx, cb, addr, pending)
+		res, err := stage.packProveCommitsMiner(ctx, bb, addr, pending)
 		if err != nil {
 			return err
 		}
@@ -72,16 +107,26 @@ type proveCommitResult struct {
 // available prove-commits, batching as much as possible.
 //
 // This function will fund as necessary from the "burnt funds actor" (look, it's convenient).
-func (ss *simulationState) packProveCommitsMiner(
-	ctx context.Context, cb packFunc, minerAddr address.Address,
+func (stage *ProveCommitStage) packProveCommitsMiner(
+	ctx context.Context, bb *blockbuilder.BlockBuilder, minerAddr address.Address,
 	pending minerPendingCommits,
 ) (res proveCommitResult, _err error) {
-	info, err := ss.getMinerInfo(ctx, minerAddr)
+	minerActor, err := bb.StateTree().GetActor(minerAddr)
+	if err != nil {
+		return res, err
+	}
+	minerState, err := miner.Load(bb.ActorStore(), minerActor)
+	if err != nil {
+		return res, err
+	}
+	info, err := minerState.Info()
 	if err != nil {
 		return res, err
 	}
 
-	nv := ss.StateManager.GetNtwkVersion(ctx, ss.nextEpoch())
+	log := bb.L().With("miner", minerAddr)
+
+	nv := bb.NetworkVersion()
 	for sealType, snos := range pending {
 		if nv >= network.Version13 {
 			for len(snos) > minProveCommitBatchSize {
@@ -91,7 +136,7 @@ func (ss *simulationState) packProveCommitsMiner(
 				}
 				batch := snos[:batchSize]
 
-				proof, err := mockAggregateSealProof(sealType, minerAddr, batchSize)
+				proof, err := mock.MockAggregateSealProof(sealType, minerAddr, batchSize)
 				if err != nil {
 					return res, err
 				}
@@ -109,7 +154,7 @@ func (ss *simulationState) packProveCommitsMiner(
 					return res, err
 				}
 
-				if _, err := sendAndFund(cb, &types.Message{
+				if _, err := stage.funding.SendAndFund(bb, &types.Message{
 					From:   info.Worker,
 					To:     minerAddr,
 					Value:  abi.NewTokenAmount(0),
@@ -117,7 +162,7 @@ func (ss *simulationState) packProveCommitsMiner(
 					Params: enc,
 				}); err == nil {
 					res.done += len(batch)
-				} else if err == ErrOutOfGas {
+				} else if blockbuilder.IsOutOfGas(err) {
 					res.full = true
 					return res, nil
 				} else if aerr, ok := err.(aerrors.ActorError); !ok || aerr.IsFatal() {
@@ -135,9 +180,9 @@ func (ss *simulationState) packProveCommitsMiner(
 					// backloged to hit this case, but we might as well handle
 					// it.
 					// First, split into "good" and "missing"
-					good, err := ss.filterProveCommits(ctx, minerAddr, batch)
+					good, err := stage.filterProveCommits(ctx, bb, minerAddr, batch)
 					if err != nil {
-						log.Errorw("failed to filter prove commits", "miner", minerAddr, "error", err)
+						log.Errorw("failed to filter prove commits", "error", err)
 						// fail with the original error.
 						return res, aerr
 					}
@@ -145,17 +190,13 @@ func (ss *simulationState) packProveCommitsMiner(
 					if removed == 0 {
 						log.Errorw("failed to prove-commit for unknown reasons",
 							"error", aerr,
-							"miner", minerAddr,
 							"sectors", batch,
-							"epoch", ss.nextEpoch(),
 						)
 						res.failed += len(batch)
 					} else if len(good) == 0 {
 						log.Errorw("failed to prove commit missing pre-commits",
 							"error", aerr,
-							"miner", minerAddr,
 							"discarded", removed,
-							"epoch", ss.nextEpoch(),
 						)
 						res.failed += len(batch)
 					} else {
@@ -166,10 +207,8 @@ func (ss *simulationState) packProveCommitsMiner(
 
 						log.Errorw("failed to prove commit expired/missing pre-commits",
 							"error", aerr,
-							"miner", minerAddr,
 							"discarded", removed,
 							"kept", len(good),
-							"epoch", ss.nextEpoch(),
 						)
 						res.failed += removed
 
@@ -178,17 +217,13 @@ func (ss *simulationState) packProveCommitsMiner(
 					}
 					log.Errorw("failed to prove commit missing sector(s)",
 						"error", err,
-						"miner", minerAddr,
 						"sectors", batch,
-						"epoch", ss.nextEpoch(),
 					)
 					res.failed += len(batch)
 				} else {
 					log.Errorw("failed to prove commit sector(s)",
 						"error", err,
-						"miner", minerAddr,
 						"sectors", batch,
-						"epoch", ss.nextEpoch(),
 					)
 					res.failed += len(batch)
 				}
@@ -200,7 +235,7 @@ func (ss *simulationState) packProveCommitsMiner(
 			sno := snos[0]
 			snos = snos[1:]
 
-			proof, err := mockSealProof(sealType, minerAddr)
+			proof, err := mock.MockSealProof(sealType, minerAddr)
 			if err != nil {
 				return res, err
 			}
@@ -212,7 +247,7 @@ func (ss *simulationState) packProveCommitsMiner(
 			if err != nil {
 				return res, err
 			}
-			if _, err := sendAndFund(cb, &types.Message{
+			if _, err := stage.funding.SendAndFund(bb, &types.Message{
 				From:   info.Worker,
 				To:     minerAddr,
 				Value:  abi.NewTokenAmount(0),
@@ -221,7 +256,7 @@ func (ss *simulationState) packProveCommitsMiner(
 			}); err == nil {
 				res.unbatched++
 				res.done++
-			} else if err == ErrOutOfGas {
+			} else if blockbuilder.IsOutOfGas(err) {
 				res.full = true
 				return res, nil
 			} else if aerr, ok := err.(aerrors.ActorError); !ok || aerr.IsFatal() {
@@ -229,9 +264,7 @@ func (ss *simulationState) packProveCommitsMiner(
 			} else {
 				log.Errorw("failed to prove commit sector(s)",
 					"error", err,
-					"miner", minerAddr,
 					"sectors", []abi.SectorNumber{sno},
-					"epoch", ss.nextEpoch(),
 				)
 				res.failed++
 			}
@@ -243,32 +276,35 @@ func (ss *simulationState) packProveCommitsMiner(
 	return res, nil
 }
 
-// loadProveCommitsMiner enqueue all pending prove-commits for the given miner. This is called on
-// load to populate the commitQueue and should not need to be called later.
+// loadMiner enqueue all pending prove-commits for the given miner. This is called on load to
+// populate the commitQueue and should not need to be called later.
 //
 // It will drop any pre-commits that have already expired.
-func (ss *simulationState) loadProveCommitsMiner(ctx context.Context, addr address.Address, minerState miner.State) error {
+func (stage *ProveCommitStage) loadMiner(ctx context.Context, bb *blockbuilder.BlockBuilder, addr address.Address) error {
+	epoch := bb.Height()
+	av := bb.ActorsVersion()
+	minerState, err := loadMiner(bb.ActorStore(), bb.ParentStateTree(), addr)
+	if err != nil {
+		return err
+	}
+
 	// Find all pending prove commits and group by proof type. Really, there should never
 	// (except during upgrades be more than one type.
-	nextEpoch := ss.nextEpoch()
-	nv := ss.StateManager.GetNtwkVersion(ctx, nextEpoch)
-	av := actors.VersionForNetwork(nv)
-
 	var total, dropped int
-	err := minerState.ForEachPrecommittedSector(func(info miner.SectorPreCommitOnChainInfo) error {
+	err = minerState.ForEachPrecommittedSector(func(info miner.SectorPreCommitOnChainInfo) error {
 		total++
 		msd := policy.GetMaxProveCommitDuration(av, info.Info.SealProof)
-		if nextEpoch > info.PreCommitEpoch+msd {
+		if epoch > info.PreCommitEpoch+msd {
 			dropped++
 			return nil
 		}
-		return ss.commitQueue.enqueueProveCommit(addr, info.PreCommitEpoch, info.Info)
+		return stage.commitQueue.enqueueProveCommit(addr, info.PreCommitEpoch, info.Info)
 	})
 	if err != nil {
 		return err
 	}
 	if dropped > 0 {
-		log.Warnw("dropped expired pre-commits on load",
+		bb.L().Warnw("dropped expired pre-commits on load",
 			"miner", addr,
 			"total", total,
 			"expired", dropped,
@@ -278,15 +314,22 @@ func (ss *simulationState) loadProveCommitsMiner(ctx context.Context, addr addre
 }
 
 // filterProveCommits filters out expired and/or missing pre-commits.
-func (ss *simulationState) filterProveCommits(ctx context.Context, minerAddr address.Address, snos []abi.SectorNumber) ([]abi.SectorNumber, error) {
-	_, minerState, err := ss.getMinerState(ctx, minerAddr)
+func (stage *ProveCommitStage) filterProveCommits(
+	ctx context.Context, bb *blockbuilder.BlockBuilder,
+	minerAddr address.Address, snos []abi.SectorNumber,
+) ([]abi.SectorNumber, error) {
+	act, err := bb.StateTree().GetActor(minerAddr)
 	if err != nil {
 		return nil, err
 	}
 
-	nextEpoch := ss.nextEpoch()
-	nv := ss.StateManager.GetNtwkVersion(ctx, nextEpoch)
-	av := actors.VersionForNetwork(nv)
+	minerState, err := miner.Load(bb.ActorStore(), act)
+	if err != nil {
+		return nil, err
+	}
+
+	nextEpoch := bb.Height()
+	av := bb.ActorsVersion()
 
 	good := make([]abi.SectorNumber, 0, len(snos))
 	for _, sno := range snos {
@@ -305,3 +348,19 @@ func (ss *simulationState) filterProveCommits(ctx context.Context, minerAddr add
 	}
 	return good, nil
 }
+
+func (stage *ProveCommitStage) load(ctx context.Context, bb *blockbuilder.BlockBuilder) error {
+	powerState, err := loadPower(bb.ActorStore(), bb.ParentStateTree())
+	if err != nil {
+		return err
+	}
+
+	return powerState.ForEachClaim(func(minerAddr address.Address, claim power.Claim) error {
+		// TODO: If we want to finish pre-commits for "new" miners, we'll need to change
+		// this.
+		if claim.RawBytePower.IsZero() {
+			return nil
+		}
+		return stage.loadMiner(ctx, bb, minerAddr)
+	})
+}

cmd/lotus-sim/simulation/stages/util.go

@@ -0,0 +1,81 @@
+package stages
+
+import (
+	"context"
+
+	"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/crypto"
+
+	"github.com/filecoin-project/lotus/chain/actors/adt"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/power"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/blockbuilder"
+)
+
+func loadMiner(store adt.Store, st types.StateTree, addr address.Address) (miner.State, error) {
+	minerActor, err := st.GetActor(addr)
+	if err != nil {
+		return nil, err
+	}
+	return miner.Load(store, minerActor)
+}
+
+func loadPower(store adt.Store, st types.StateTree) (power.State, error) {
+	powerActor, err := st.GetActor(power.Address)
+	if err != nil {
+		return nil, err
+	}
+	return power.Load(store, powerActor)
+}
+
+// Compute the number of sectors a miner has from their power claim.
+func sectorsFromClaim(sectorSize abi.SectorSize, c power.Claim) int64 {
+	if c.RawBytePower.Int == nil {
+		return 0
+	}
+	sectorCount := big.Div(c.RawBytePower, big.NewIntUnsigned(uint64(sectorSize)))
+	if !sectorCount.IsInt64() {
+		panic("impossible number of sectors")
+	}
+	return sectorCount.Int64()
+}
+
+// loadClaims will load all non-zero claims at the given epoch.
+func loadClaims(
+	ctx context.Context, bb *blockbuilder.BlockBuilder, height abi.ChainEpoch,
+) (map[address.Address]power.Claim, error) {
+	powerTable := make(map[address.Address]power.Claim)
+
+	st, err := bb.StateTreeByHeight(height)
+	if err != nil {
+		return nil, err
+	}
+
+	powerState, err := loadPower(bb.ActorStore(), st)
+	if err != nil {
+		return nil, err
+	}
+
+	err = powerState.ForEachClaim(func(miner address.Address, claim power.Claim) error {
+		// skip miners without power
+		if claim.RawBytePower.IsZero() {
+			return nil
+		}
+		powerTable[miner] = claim
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return powerTable, nil
+}
+
+func postChainCommitInfo(ctx context.Context, bb *blockbuilder.BlockBuilder, epoch abi.ChainEpoch) (abi.Randomness, error) {
+	cs := bb.StateManager().ChainStore()
+	ts := bb.ParentTipSet()
+	commitRand, err := cs.GetChainRandomness(ctx, ts.Cids(), crypto.DomainSeparationTag_PoStChainCommit, epoch, nil, true)
+	return commitRand, err
+}

cmd/lotus-sim/simulation/stages/windowpost_stage.go

@@ -0,0 +1,312 @@
+package stages
+
+import (
+	"context"
+	"math"
+	"time"
+
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-state-types/abi"
+
+	proof5 "github.com/filecoin-project/specs-actors/v5/actors/runtime/proof"
+
+	"github.com/filecoin-project/lotus/chain/actors"
+	"github.com/filecoin-project/lotus/chain/actors/aerrors"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/power"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/blockbuilder"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/mock"
+)
+
+type WindowPoStStage struct {
+	// We track the window post periods per miner and assume that no new miners are ever added.
+
+	// We record all pending window post messages, and the epoch up through which we've
+	// generated window post messages.
+	pendingWposts  []*types.Message
+	wpostPeriods   [][]address.Address // (epoch % (epochs in a deadline)) -> miner
+	nextWpostEpoch abi.ChainEpoch
+}
+
+func NewWindowPoStStage() (*WindowPoStStage, error) {
+	return new(WindowPoStStage), nil
+}
+
+func (*WindowPoStStage) Name() string {
+	return "window-post"
+}
+
+// packWindowPoSts packs window posts until either the block is full or all healty sectors
+// have been proven. It does not recover sectors.
+func (stage *WindowPoStStage) PackMessages(ctx context.Context, bb *blockbuilder.BlockBuilder) (_err error) {
+	// Push any new window posts into the queue.
+	if err := stage.tick(ctx, bb); err != nil {
+		return err
+	}
+	done := 0
+	failed := 0
+	defer func() {
+		if _err != nil {
+			return
+		}
+
+		bb.L().Debugw("packed window posts",
+			"done", done,
+			"failed", failed,
+			"remaining", len(stage.pendingWposts),
+		)
+	}()
+	// Then pack as many as we can.
+	for len(stage.pendingWposts) > 0 {
+		next := stage.pendingWposts[0]
+		if _, err := bb.PushMessage(next); err != nil {
+			if blockbuilder.IsOutOfGas(err) {
+				return nil
+			}
+			if aerr, ok := err.(aerrors.ActorError); !ok || aerr.IsFatal() {
+				return err
+			}
+			bb.L().Errorw("failed to submit windowed post",
+				"error", err,
+				"miner", next.To,
+			)
+			failed++
+		} else {
+			done++
+		}
+
+		stage.pendingWposts = stage.pendingWposts[1:]
+	}
+	stage.pendingWposts = nil
+	return nil
+}
+
+// stepWindowPoStsMiner enqueues all missing window posts for the current epoch for the given miner.
+func (stage *WindowPoStStage) queueMiner(
+	ctx context.Context, bb *blockbuilder.BlockBuilder,
+	addr address.Address, minerState miner.State,
+	commitEpoch abi.ChainEpoch, commitRand abi.Randomness,
+) error {
+
+	if active, err := minerState.DeadlineCronActive(); err != nil {
+		return err
+	} else if !active {
+		return nil
+	}
+
+	minerInfo, err := minerState.Info()
+	if err != nil {
+		return err
+	}
+
+	di, err := minerState.DeadlineInfo(bb.Height())
+	if err != nil {
+		return err
+	}
+	di = di.NextNotElapsed()
+
+	dl, err := minerState.LoadDeadline(di.Index)
+	if err != nil {
+		return err
+	}
+
+	provenBf, err := dl.PartitionsPoSted()
+	if err != nil {
+		return err
+	}
+	proven, err := provenBf.AllMap(math.MaxUint64)
+	if err != nil {
+		return err
+	}
+
+	var (
+		partitions      []miner.PoStPartition
+		partitionGroups [][]miner.PoStPartition
+	)
+	// Only prove partitions with live sectors.
+	err = dl.ForEachPartition(func(idx uint64, part miner.Partition) error {
+		if proven[idx] {
+			return nil
+		}
+		// TODO: set this to the actual limit from specs-actors.
+		// NOTE: We're mimicing the behavior of wdpost_run.go here.
+		if len(partitions) > 0 && idx%4 == 0 {
+			partitionGroups = append(partitionGroups, partitions)
+			partitions = nil
+
+		}
+		live, err := part.LiveSectors()
+		if err != nil {
+			return err
+		}
+		liveCount, err := live.Count()
+		if err != nil {
+			return err
+		}
+		faulty, err := part.FaultySectors()
+		if err != nil {
+			return err
+		}
+		faultyCount, err := faulty.Count()
+		if err != nil {
+			return err
+		}
+		if liveCount-faultyCount > 0 {
+			partitions = append(partitions, miner.PoStPartition{Index: idx})
+		}
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+	if len(partitions) > 0 {
+		partitionGroups = append(partitionGroups, partitions)
+		partitions = nil
+	}
+
+	proof, err := mock.MockWindowPoStProof(minerInfo.WindowPoStProofType, addr)
+	if err != nil {
+		return err
+	}
+	for _, group := range partitionGroups {
+		params := miner.SubmitWindowedPoStParams{
+			Deadline:   di.Index,
+			Partitions: group,
+			Proofs: []proof5.PoStProof{{
+				PoStProof:  minerInfo.WindowPoStProofType,
+				ProofBytes: proof,
+			}},
+			ChainCommitEpoch: commitEpoch,
+			ChainCommitRand:  commitRand,
+		}
+		enc, aerr := actors.SerializeParams(&params)
+		if aerr != nil {
+			return xerrors.Errorf("could not serialize submit window post parameters: %w", aerr)
+		}
+		msg := &types.Message{
+			To:     addr,
+			From:   minerInfo.Worker,
+			Method: miner.Methods.SubmitWindowedPoSt,
+			Params: enc,
+			Value:  types.NewInt(0),
+		}
+		stage.pendingWposts = append(stage.pendingWposts, msg)
+	}
+	return nil
+}
+
+func (stage *WindowPoStStage) load(ctx context.Context, bb *blockbuilder.BlockBuilder) (_err error) {
+	bb.L().Info("loading window post info")
+
+	start := time.Now()
+	defer func() {
+		if _err != nil {
+			return
+		}
+
+		bb.L().Infow("loaded window post info", "duration", time.Since(start))
+	}()
+
+	// reset
+	stage.wpostPeriods = make([][]address.Address, miner.WPoStChallengeWindow)
+	stage.pendingWposts = nil
+	stage.nextWpostEpoch = bb.Height() + 1
+
+	st := bb.ParentStateTree()
+	store := bb.ActorStore()
+
+	powerState, err := loadPower(store, st)
+	if err != nil {
+		return err
+	}
+
+	commitEpoch := bb.ParentTipSet().Height()
+	commitRand, err := postChainCommitInfo(ctx, bb, commitEpoch)
+	if err != nil {
+		return err
+	}
+
+	return powerState.ForEachClaim(func(minerAddr address.Address, claim power.Claim) error {
+		// TODO: If we start recovering power, we'll need to change this.
+		if claim.RawBytePower.IsZero() {
+			return nil
+		}
+
+		minerState, err := loadMiner(store, st, minerAddr)
+		if err != nil {
+			return err
+		}
+
+		// Shouldn't be necessary if the miner has power, but we might as well be safe.
+		if active, err := minerState.DeadlineCronActive(); err != nil {
+			return err
+		} else if !active {
+			return nil
+		}
+
+		// Record when we need to prove for this miner.
+		dinfo, err := minerState.DeadlineInfo(bb.Height())
+		if err != nil {
+			return err
+		}
+		dinfo = dinfo.NextNotElapsed()
+
+		ppOffset := int(dinfo.PeriodStart % miner.WPoStChallengeWindow)
+		stage.wpostPeriods[ppOffset] = append(stage.wpostPeriods[ppOffset], minerAddr)
+
+		return stage.queueMiner(ctx, bb, minerAddr, minerState, commitEpoch, commitRand)
+	})
+}
+
+func (stage *WindowPoStStage) tick(ctx context.Context, bb *blockbuilder.BlockBuilder) error {
+	// If this is our first time, load from scratch.
+	if stage.wpostPeriods == nil {
+		return stage.load(ctx, bb)
+	}
+
+	targetHeight := bb.Height()
+	now := time.Now()
+	was := len(stage.pendingWposts)
+	count := 0
+	defer func() {
+		bb.L().Debugw("computed window posts",
+			"miners", count,
+			"count", len(stage.pendingWposts)-was,
+			"duration", time.Since(now),
+		)
+	}()
+
+	st := bb.ParentStateTree()
+	store := bb.ActorStore()
+
+	// Perform a bit of catch up. This lets us do things like skip blocks at upgrades then catch
+	// up to make the simualtion easier.
+	for ; stage.nextWpostEpoch <= targetHeight; stage.nextWpostEpoch++ {
+		if stage.nextWpostEpoch+miner.WPoStChallengeWindow < targetHeight {
+			bb.L().Warnw("skipping old window post", "deadline-open", stage.nextWpostEpoch)
+			continue
+		}
+		commitEpoch := stage.nextWpostEpoch - 1
+		commitRand, err := postChainCommitInfo(ctx, bb, commitEpoch)
+		if err != nil {
+			return err
+		}
+
+		for _, addr := range stage.wpostPeriods[int(stage.nextWpostEpoch%miner.WPoStChallengeWindow)] {
+			minerState, err := loadMiner(store, st, addr)
+			if err != nil {
+				return err
+			}
+
+			if err := stage.queueMiner(ctx, bb, addr, minerState, commitEpoch, commitRand); err != nil {
+				return err
+			}
+			count++
+		}
+
+	}
+	return nil
+}

cmd/lotus-sim/simulation/state.go

@@ -1,202 +0,0 @@
-package simulation
-
-import (
-	"context"
-	"sort"
-
-	"github.com/filecoin-project/go-address"
-	"github.com/filecoin-project/go-state-types/abi"
-	"golang.org/x/xerrors"
-
-	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
-	"github.com/filecoin-project/lotus/chain/types"
-)
-
-// simualtionState holds the "state" of the simulation. This is split from the Simulation type so we
-// can load it on-dempand if and when we need to actually _run_ the simualation. Loading the
-// simulation state requires walking all active miners.
-type simulationState struct {
-	*Simulation
-
-	// The tiers represent the top 1%, top 10%, and everyone else. When sealing sectors, we seal
-	// a group of sectors for the top 1%, a group (half that size) for the top 10%, and one
-	// sector for everyone else. We determine these rates by looking at two power tables.
-	// TODO Ideally we'd "learn" this distribution from the network. But this is good enough for
-	// now.
-	minerDist struct {
-		top1, top10, rest actorIter
-	}
-
-	// We track the window post periods per miner and assume that no new miners are ever added.
-	wpostPeriods map[int][]address.Address // (epoch % (epochs in a deadline)) -> miner
-	// We cache all miner infos for active miners and assume no new miners join.
-	minerInfos map[address.Address]*miner.MinerInfo
-
-	// We record all pending window post messages, and the epoch up through which we've
-	// generated window post messages.
-	pendingWposts  []*types.Message
-	nextWpostEpoch abi.ChainEpoch
-
-	// We track the set of pending commits. On simulation load, and when a new pre-commit is
-	// added to the chain, we put the commit in this queue. advanceEpoch(currentEpoch) should be
-	// called on this queue at every epoch before using it.
-	commitQueue commitQueue
-}
-
-func loadSimulationState(ctx context.Context, sim *Simulation) (*simulationState, error) {
-	state := &simulationState{Simulation: sim}
-	currentEpoch := sim.head.Height()
-
-	// Lookup the current power table and the power table 2 weeks ago (for onboarding rate
-	// projections).
-	currentPowerTable, err := sim.loadClaims(ctx, currentEpoch)
-	if err != nil {
-		return nil, err
-	}
-
-	var lookbackEpoch abi.ChainEpoch
-	//if epoch > onboardingProjectionLookback {
-	//	lookbackEpoch = epoch - onboardingProjectionLookback
-	//}
-	// TODO: Fixme? I really want this to not suck with snapshots.
-	lookbackEpoch = 770139 // hard coded for now.
-	lookbackPowerTable, err := sim.loadClaims(ctx, lookbackEpoch)
-	if err != nil {
-		return nil, err
-	}
-
-	type onboardingInfo struct {
-		addr           address.Address
-		onboardingRate uint64
-	}
-
-	commitRand, err := sim.postChainCommitInfo(ctx, currentEpoch)
-	if err != nil {
-		return nil, err
-	}
-
-	sealList := make([]onboardingInfo, 0, len(currentPowerTable))
-	state.wpostPeriods = make(map[int][]address.Address, miner.WPoStChallengeWindow)
-	state.minerInfos = make(map[address.Address]*miner.MinerInfo, len(currentPowerTable))
-	state.commitQueue.advanceEpoch(state.nextEpoch())
-	for addr, claim := range currentPowerTable {
-		// Load the miner state.
-		_, minerState, err := state.getMinerState(ctx, addr)
-		if err != nil {
-			return nil, err
-		}
-
-		info, err := minerState.Info()
-		if err != nil {
-			return nil, err
-		}
-		state.minerInfos[addr] = &info
-
-		// Queue up PoSts
-		err = state.stepWindowPoStsMiner(ctx, addr, minerState, currentEpoch, commitRand)
-		if err != nil {
-			return nil, err
-		}
-
-		// Qeueu up any pending prove commits.
-		err = state.loadProveCommitsMiner(ctx, addr, minerState)
-		if err != nil {
-			return nil, err
-		}
-
-		// Record when we need to prove for this miner.
-		dinfo, err := minerState.DeadlineInfo(state.nextEpoch())
-		if err != nil {
-			return nil, err
-		}
-		dinfo = dinfo.NextNotElapsed()
-
-		ppOffset := int(dinfo.PeriodStart % miner.WPoStChallengeWindow)
-		state.wpostPeriods[ppOffset] = append(state.wpostPeriods[ppOffset], addr)
-
-		sectorsAdded := sectorsFromClaim(info.SectorSize, claim)
-		if lookbackClaim, ok := lookbackPowerTable[addr]; !ok {
-			sectorsAdded -= sectorsFromClaim(info.SectorSize, lookbackClaim)
-		}
-
-		// NOTE: power _could_ have been lost, but that's too much of a pain to care
-		// about. We _could_ look for faulty power by iterating through all
-		// deadlines, but I'd rather not.
-		if sectorsAdded > 0 {
-			sealList = append(sealList, onboardingInfo{addr, uint64(sectorsAdded)})
-		}
-	}
-	if len(sealList) == 0 {
-		return nil, xerrors.Errorf("simulation has no miners")
-	}
-
-	// We're already done loading for the _next_ epoch.
-	// Next time, we need to load for the next, next epoch.
-	// TODO: fix this insanity.
-	state.nextWpostEpoch = state.nextEpoch() + 1
-
-	// Now that we have a list of sealing miners, sort them into percentiles.
-	sort.Slice(sealList, func(i, j int) bool {
-		return sealList[i].onboardingRate < sealList[j].onboardingRate
-	})
-
-	for i, oi := range sealList {
-		var dist *actorIter
-		if i < len(sealList)/100 {
-			dist = &state.minerDist.top1
-		} else if i < len(sealList)/10 {
-			dist = &state.minerDist.top10
-		} else {
-			dist = &state.minerDist.rest
-		}
-		dist.add(oi.addr)
-	}
-
-	state.minerDist.top1.shuffle()
-	state.minerDist.top10.shuffle()
-	state.minerDist.rest.shuffle()
-
-	return state, nil
-}
-
-// nextEpoch returns the next epoch (head+1).
-func (ss *simulationState) nextEpoch() abi.ChainEpoch {
-	return ss.GetHead().Height() + 1
-}
-
-// getMinerInfo returns the miner's cached info.
-//
-// NOTE: we assume that miner infos won't change. We'll need to fix this if we start supporting arbitrary message.
-func (ss *simulationState) getMinerInfo(ctx context.Context, addr address.Address) (*miner.MinerInfo, error) {
-	minerInfo, ok := ss.minerInfos[addr]
-	if !ok {
-		_, minerState, err := ss.getMinerState(ctx, addr)
-		if err != nil {
-			return nil, err
-		}
-		info, err := minerState.Info()
-		if err != nil {
-			return nil, err
-		}
-		minerInfo = &info
-		ss.minerInfos[addr] = minerInfo
-	}
-	return minerInfo, nil
-}
-
-// getMinerState loads the miner actor & state.
-func (ss *simulationState) getMinerState(ctx context.Context, addr address.Address) (*types.Actor, miner.State, error) {
-	st, err := ss.stateTree(ctx)
-	if err != nil {
-		return nil, nil, err
-	}
-	act, err := st.GetActor(addr)
-	if err != nil {
-		return nil, nil, err
-	}
-	state, err := miner.Load(ss.Chainstore.ActorStore(ctx), act)
-	if err != nil {
-		return nil, nil, err
-	}
-	return act, state, err
-}

cmd/lotus-sim/simulation/step.go

@@ -2,83 +2,38 @@ package simulation
 
 import (
 	"context"
-	"errors"
-	"reflect"
-	"runtime"
-	"strings"
 
 	"golang.org/x/xerrors"
 
-	"github.com/filecoin-project/go-address"
-	"github.com/filecoin-project/go-state-types/abi"
-
-	"github.com/filecoin-project/lotus/build"
-	"github.com/filecoin-project/lotus/chain/actors/builtin/account"
-	"github.com/filecoin-project/lotus/chain/state"
-	"github.com/filecoin-project/lotus/chain/stmgr"
-	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/types"
-	"github.com/filecoin-project/lotus/chain/vm"
+	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation/blockbuilder"
 )
 
-const (
-	// The number of expected blocks in a tipset. We use this to determine how much gas a tipset
-	// has.
-	expectedBlocks = 5
-	// TODO: This will produce invalid blocks but it will accurately model the amount of gas
-	// we're willing to use per-tipset.
-	// A more correct approach would be to produce 5 blocks. We can do that later.
-	targetGas = build.BlockGasTarget * expectedBlocks
-)
-
-var baseFee = abi.NewTokenAmount(0)
-
 // Step steps the simulation forward one step. This may move forward by more than one epoch.
 func (sim *Simulation) Step(ctx context.Context) (*types.TipSet, error) {
-	state, err := sim.simState(ctx)
-	if err != nil {
-		return nil, err
-	}
-	ts, err := state.step(ctx)
-	if err != nil {
-		return nil, xerrors.Errorf("failed to step simulation: %w", err)
-	}
-	return ts, nil
-}
-
-// step steps the simulation state forward one step, producing and executing a new tipset.
-func (ss *simulationState) step(ctx context.Context) (*types.TipSet, error) {
-	log.Infow("step", "epoch", ss.head.Height()+1)
-	messages, err := ss.popNextMessages(ctx)
+	log.Infow("step", "epoch", sim.head.Height()+1)
+	messages, err := sim.popNextMessages(ctx)
 	if err != nil {
 		return nil, xerrors.Errorf("failed to select messages for block: %w", err)
 	}
-	head, err := ss.makeTipSet(ctx, messages)
+	head, err := sim.makeTipSet(ctx, messages)
 	if err != nil {
 		return nil, xerrors.Errorf("failed to make tipset: %w", err)
 	}
-	if err := ss.SetHead(head); err != nil {
+	if err := sim.SetHead(head); err != nil {
 		return nil, xerrors.Errorf("failed to update head: %w", err)
 	}
 	return head, nil
 }
 
-var ErrOutOfGas = errors.New("out of gas")
-
-// packFunc takes a message and attempts to pack it into a block.
-//
-// - If the block is full, returns the error ErrOutOfGas.
-// - If message execution fails, check if error is an ActorError to get the return code.
-type packFunc func(*types.Message) (*types.MessageReceipt, error)
-
 // popNextMessages generates/picks a set of messages to be included in the next block.
 //
 // - This function is destructive and should only be called once per epoch.
 // - This function does not store anything in the repo.
 // - This function handles all gas estimation. The returned messages should all fit in a single
 //   block.
-func (ss *simulationState) popNextMessages(ctx context.Context) ([]*types.Message, error) {
-	parentTs := ss.head
+func (sim *Simulation) popNextMessages(ctx context.Context) ([]*types.Message, error) {
+	parentTs := sim.head
 
 	// First we make sure we don't have an upgrade at this epoch. If we do, we return no
 	// messages so we can just create an empty block at that epoch.
@@ -86,8 +41,8 @@ func (ss *simulationState) popNextMessages(ctx context.Context) ([]*types.Messag
 	// This isn't what the network does, but it makes things easier. Otherwise, we'd need to run
 	// migrations before this epoch and I'd rather not deal with that.
 	nextHeight := parentTs.Height() + 1
-	prevVer := ss.StateManager.GetNtwkVersion(ctx, nextHeight-1)
-	nextVer := ss.StateManager.GetNtwkVersion(ctx, nextHeight)
+	prevVer := sim.StateManager.GetNtwkVersion(ctx, nextHeight-1)
+	nextVer := sim.StateManager.GetNtwkVersion(ctx, nextHeight)
 	if nextVer != prevVer {
 		log.Warnw("packing no messages for version upgrade block",
 			"old", prevVer,
@@ -97,170 +52,20 @@ func (ss *simulationState) popNextMessages(ctx context.Context) ([]*types.Messag
 		return nil, nil
 	}
 
-	// Next, we compute the state for the parent tipset. In practice, this will likely be
-	// cached.
-	parentState, _, err := ss.StateManager.TipSetState(ctx, parentTs)
-	if err != nil {
-		return nil, err
-	}
-
-	// Then we construct a VM to execute messages for gas estimation.
-	//
-	// Most parts of this VM are "real" except:
-	// 1. We don't charge a fee.
-	// 2. The runtime has "fake" proof logic.
-	// 3. We don't actually save any of the results.
-	r := store.NewChainRand(ss.StateManager.ChainStore(), parentTs.Cids())
-	vmopt := &vm.VMOpts{
-		StateBase:      parentState,
-		Epoch:          nextHeight,
-		Rand:           r,
-		Bstore:         ss.StateManager.ChainStore().StateBlockstore(),
-		Syscalls:       ss.StateManager.ChainStore().VMSys(),
-		CircSupplyCalc: ss.StateManager.GetVMCirculatingSupply,
-		NtwkVersion:    ss.StateManager.GetNtwkVersion,
-		BaseFee:        baseFee, // FREE!
-		LookbackState:  stmgr.LookbackStateGetterForTipset(ss.StateManager, parentTs),
-	}
-	vmi, err := vm.NewVM(ctx, vmopt)
-	if err != nil {
-		return nil, err
-	}
-
-	// Next we define a helper function for "pushing" messages. This is the function that will
-	// be passed to the "pack" functions.
-	//
-	// It.
-	//
-	// 1. Tries to execute the message on-top-of the already pushed message.
-	// 2. Is careful to revert messages on failure to avoid nasties like nonce-gaps.
-	// 3. Resolves IDs as necessary, fills in missing parts of the message, etc.
-	vmStore := vmi.ActorStore(ctx)
-	var gasTotal int64
-	var messages []*types.Message
-	tryPushMsg := func(msg *types.Message) (*types.MessageReceipt, error) {
-		if gasTotal >= targetGas {
-			return nil, ErrOutOfGas
-		}
-
-		// Copy the message before we start mutating it.
-		msgCpy := *msg
-		msg = &msgCpy
-		st := vmi.StateTree().(*state.StateTree)
-
-		actor, err := st.GetActor(msg.From)
-		if err != nil {
-			return nil, err
-		}
-		msg.Nonce = actor.Nonce
-		if msg.From.Protocol() == address.ID {
-			state, err := account.Load(vmStore, actor)
-			if err != nil {
-				return nil, err
-			}
-			msg.From, err = state.PubkeyAddress()
-			if err != nil {
-				return nil, err
-			}
-		}
-
-		// TODO: Our gas estimation is broken for payment channels due to horrible hacks in
-		// gasEstimateGasLimit.
-		if msg.Value == types.EmptyInt {
-			msg.Value = abi.NewTokenAmount(0)
-		}
-		msg.GasPremium = abi.NewTokenAmount(0)
-		msg.GasFeeCap = abi.NewTokenAmount(0)
-		msg.GasLimit = build.BlockGasLimit
-
-		// We manually snapshot so we can revert nonce changes, etc. on failure.
-		st.Snapshot(ctx)
-		defer st.ClearSnapshot()
-
-		ret, err := vmi.ApplyMessage(ctx, msg)
-		if err != nil {
-			_ = st.Revert()
-			return nil, err
-		}
-		if ret.ActorErr != nil {
-			_ = st.Revert()
-			return nil, ret.ActorErr
-		}
-
-		// Sometimes there are bugs. Let's catch them.
-		if ret.GasUsed == 0 {
-			_ = st.Revert()
-			return nil, xerrors.Errorf("used no gas",
-				"msg", msg,
-				"ret", ret,
+	bb, err := blockbuilder.NewBlockBuilder(
+		ctx, log.With("simulation", sim.name),
+		sim.StateManager, parentTs,
 	)
-		}
-
-		// TODO: consider applying overestimation? We're likely going to "over pack" here by
-		// ~25% because we're too accurate.
-
-		// Did we go over? Yes, revert.
-		newTotal := gasTotal + ret.GasUsed
-		if newTotal > targetGas {
-			_ = st.Revert()
-			return nil, ErrOutOfGas
-		}
-		gasTotal = newTotal
-
-		// Update the gas limit.
-		msg.GasLimit = ret.GasUsed
-
-		messages = append(messages, msg)
-		return &ret.MessageReceipt, nil
-	}
-
-	// Finally, we generate a set of messages to be included in
-	if err := ss.packMessages(ctx, tryPushMsg); err != nil {
+	if err != nil {
 		return nil, err
 	}
 
-	return messages, nil
-}
-
-// functionName extracts the name of given function.
-func functionName(fn interface{}) string {
-	name := runtime.FuncForPC(reflect.ValueOf(fn).Pointer()).Name()
-	lastDot := strings.LastIndexByte(name, '.')
-	if lastDot >= 0 {
-		name = name[lastDot+1 : len(name)-3]
-	}
-	lastDash := strings.LastIndexByte(name, '-')
-	if lastDash > 0 {
-		name = name[:lastDash]
-	}
-	return name
-}
-
-// packMessages packs messages with the given packFunc until the block is full (packFunc returns
-// true).
-// TODO: Make this more configurable for other simulations.
-func (ss *simulationState) packMessages(ctx context.Context, cb packFunc) error {
-	type messageGenerator func(ctx context.Context, cb packFunc) error
-
-	// We pack messages in-order:
-	// 1. Any window posts. We pack window posts as soon as the deadline opens to ensure we only
-	//    miss them if/when we run out of chain bandwidth.
-	// 2. We then move funds to our "funding" account, if it's running low.
-	// 3. Prove commits. We do this eagerly to ensure they don't expire.
-	// 4. Finally, we fill the rest of the space with pre-commits.
-	messageGenerators := []messageGenerator{
-		ss.packWindowPoSts,
-		ss.packFunding,
-		ss.packProveCommits,
-		ss.packPreCommits,
-	}
-
-	for _, mgen := range messageGenerators {
+	for _, stage := range sim.stages {
 		// We're intentionally ignoring the "full" signal so we can try to pack a few more
 		// messages.
-		if err := mgen(ctx, cb); err != nil && !xerrors.Is(err, ErrOutOfGas) {
-			return xerrors.Errorf("when packing messages with %s: %w", functionName(mgen), err)
+		if err := stage.PackMessages(ctx, bb); err != nil && !blockbuilder.IsOutOfGas(err) {
+			return nil, xerrors.Errorf("when packing messages with %s: %w", stage.Name(), err)
 		}
 	}
-	return nil
+	return bb.Messages(), nil
 }

cmd/lotus-sim/simulation/wdpost.go

@@ -1,229 +0,0 @@
-package simulation
-
-import (
-	"context"
-	"math"
-	"time"
-
-	"golang.org/x/xerrors"
-
-	"github.com/filecoin-project/go-address"
-	"github.com/filecoin-project/go-state-types/abi"
-	"github.com/filecoin-project/go-state-types/crypto"
-
-	proof5 "github.com/filecoin-project/specs-actors/v5/actors/runtime/proof"
-
-	"github.com/filecoin-project/lotus/chain/actors"
-	"github.com/filecoin-project/lotus/chain/actors/aerrors"
-	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
-	"github.com/filecoin-project/lotus/chain/types"
-)
-
-// postChainCommitInfo returns th
-func (sim *Simulation) postChainCommitInfo(ctx context.Context, epoch abi.ChainEpoch) (abi.Randomness, error) {
-	commitRand, err := sim.Chainstore.GetChainRandomness(
-		ctx, sim.head.Cids(), crypto.DomainSeparationTag_PoStChainCommit, epoch, nil, true)
-	return commitRand, err
-}
-
-// packWindowPoSts packs window posts until either the block is full or all healty sectors
-// have been proven. It does not recover sectors.
-func (ss *simulationState) packWindowPoSts(ctx context.Context, cb packFunc) (_err error) {
-	// Push any new window posts into the queue.
-	if err := ss.queueWindowPoSts(ctx); err != nil {
-		return err
-	}
-	done := 0
-	failed := 0
-	defer func() {
-		if _err != nil {
-			return
-		}
-
-		log.Debugw("packed window posts",
-			"epoch", ss.nextEpoch(),
-			"done", done,
-			"failed", failed,
-			"remaining", len(ss.pendingWposts),
-		)
-	}()
-	// Then pack as many as we can.
-	for len(ss.pendingWposts) > 0 {
-		next := ss.pendingWposts[0]
-		if _, err := cb(next); err != nil {
-			if aerr, ok := err.(aerrors.ActorError); !ok || aerr.IsFatal() {
-				return err
-			}
-			log.Errorw("failed to submit windowed post",
-				"error", err,
-				"miner", next.To,
-				"epoch", ss.nextEpoch(),
-			)
-			failed++
-		} else {
-			done++
-		}
-
-		ss.pendingWposts = ss.pendingWposts[1:]
-	}
-	ss.pendingWposts = nil
-	return nil
-}
-
-// stepWindowPoStsMiner enqueues all missing window posts for the current epoch for the given miner.
-func (ss *simulationState) stepWindowPoStsMiner(
-	ctx context.Context,
-	addr address.Address, minerState miner.State,
-	commitEpoch abi.ChainEpoch, commitRand abi.Randomness,
-) error {
-
-	if active, err := minerState.DeadlineCronActive(); err != nil {
-		return err
-	} else if !active {
-		return nil
-	}
-
-	minerInfo, err := ss.getMinerInfo(ctx, addr)
-	if err != nil {
-		return err
-	}
-
-	di, err := minerState.DeadlineInfo(ss.nextEpoch())
-	if err != nil {
-		return err
-	}
-	di = di.NextNotElapsed()
-
-	dl, err := minerState.LoadDeadline(di.Index)
-	if err != nil {
-		return err
-	}
-
-	provenBf, err := dl.PartitionsPoSted()
-	if err != nil {
-		return err
-	}
-	proven, err := provenBf.AllMap(math.MaxUint64)
-	if err != nil {
-		return err
-	}
-
-	var (
-		partitions      []miner.PoStPartition
-		partitionGroups [][]miner.PoStPartition
-	)
-	// Only prove partitions with live sectors.
-	err = dl.ForEachPartition(func(idx uint64, part miner.Partition) error {
-		if proven[idx] {
-			return nil
-		}
-		// TODO: set this to the actual limit from specs-actors.
-		// NOTE: We're mimicing the behavior of wdpost_run.go here.
-		if len(partitions) > 0 && idx%4 == 0 {
-			partitionGroups = append(partitionGroups, partitions)
-			partitions = nil
-
-		}
-		live, err := part.LiveSectors()
-		if err != nil {
-			return err
-		}
-		liveCount, err := live.Count()
-		if err != nil {
-			return err
-		}
-		faulty, err := part.FaultySectors()
-		if err != nil {
-			return err
-		}
-		faultyCount, err := faulty.Count()
-		if err != nil {
-			return err
-		}
-		if liveCount-faultyCount > 0 {
-			partitions = append(partitions, miner.PoStPartition{Index: idx})
-		}
-		return nil
-	})
-	if err != nil {
-		return err
-	}
-	if len(partitions) > 0 {
-		partitionGroups = append(partitionGroups, partitions)
-		partitions = nil
-	}
-
-	proof, err := mockWpostProof(minerInfo.WindowPoStProofType, addr)
-	if err != nil {
-		return err
-	}
-	for _, group := range partitionGroups {
-		params := miner.SubmitWindowedPoStParams{
-			Deadline:   di.Index,
-			Partitions: group,
-			Proofs: []proof5.PoStProof{{
-				PoStProof:  minerInfo.WindowPoStProofType,
-				ProofBytes: proof,
-			}},
-			ChainCommitEpoch: commitEpoch,
-			ChainCommitRand:  commitRand,
-		}
-		enc, aerr := actors.SerializeParams(&params)
-		if aerr != nil {
-			return xerrors.Errorf("could not serialize submit window post parameters: %w", aerr)
-		}
-		msg := &types.Message{
-			To:     addr,
-			From:   minerInfo.Worker,
-			Method: miner.Methods.SubmitWindowedPoSt,
-			Params: enc,
-			Value:  types.NewInt(0),
-		}
-		ss.pendingWposts = append(ss.pendingWposts, msg)
-	}
-	return nil
-}
-
-// queueWindowPoSts enqueues missing window posts for all miners with deadlines opening between the
-// last epoch in which this function was called and the current epoch (head+1).
-func (ss *simulationState) queueWindowPoSts(ctx context.Context) error {
-	targetHeight := ss.nextEpoch()
-
-	now := time.Now()
-	was := len(ss.pendingWposts)
-	count := 0
-	defer func() {
-		log.Debugw("computed window posts",
-			"miners", count,
-			"count", len(ss.pendingWposts)-was,
-			"duration", time.Since(now),
-		)
-	}()
-
-	// Perform a bit of catch up. This lets us do things like skip blocks at upgrades then catch
-	// up to make the simualtion easier.
-	for ; ss.nextWpostEpoch <= targetHeight; ss.nextWpostEpoch++ {
-		if ss.nextWpostEpoch+miner.WPoStChallengeWindow < targetHeight {
-			log.Warnw("skipping old window post", "epoch", ss.nextWpostEpoch)
-			continue
-		}
-		commitEpoch := ss.nextWpostEpoch - 1
-		commitRand, err := ss.postChainCommitInfo(ctx, commitEpoch)
-		if err != nil {
-			return err
-		}
-
-		for _, addr := range ss.wpostPeriods[int(ss.nextWpostEpoch%miner.WPoStChallengeWindow)] {
-			_, minerState, err := ss.getMinerState(ctx, addr)
-			if err != nil {
-				return err
-			}
-			if err := ss.stepWindowPoStsMiner(ctx, addr, minerState, commitEpoch, commitRand); err != nil {
-				return err
-			}
-			count++
-		}
-
-	}
-	return nil
-}