Merge pull request #7966 from filecoin-project/asr/compute-state-vm

Fix: chain: create a new VM for each epoch

chain/consensus/filcns/compute_state.go

@@ -92,16 +92,16 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context, sm *stmgr.StateManager
 		partDone()
 	}()
 
-	makeVmWithBaseState := func(base cid.Cid) (*vm.VM, error) {
+	makeVmWithBaseStateAndEpoch := func(base cid.Cid, e abi.ChainEpoch) (*vm.VM, error) {
 		vmopt := &vm.VMOpts{
 			StateBase:      base,
-			Epoch:          epoch,
+			Epoch:          e,
 			Rand:           r,
 			Bstore:         sm.ChainStore().StateBlockstore(),
 			Actors:         NewActorRegistry(),
 			Syscalls:       sm.Syscalls,
 			CircSupplyCalc: sm.GetVMCirculatingSupply,
-			NetworkVersion: sm.GetNetworkVersion(ctx, epoch),
+			NetworkVersion: sm.GetNetworkVersion(ctx, e),
 			BaseFee:        baseFee,
 			LookbackState:  stmgr.LookbackStateGetterForTipset(sm, ts),
 		}
@@ -109,12 +109,7 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context, sm *stmgr.StateManager
 		return sm.VMConstructor()(ctx, vmopt)
 	}
 
-	vmi, err := makeVmWithBaseState(pstate)
+	runCron := func(vmCron *vm.VM, epoch abi.ChainEpoch) error {
-	if err != nil {
-		return cid.Undef, cid.Undef, xerrors.Errorf("making vm: %w", err)
-	}
-
-	runCron := func(epoch abi.ChainEpoch) error {
 		cronMsg := &types.Message{
 			To:         cron.Address,
 			From:       builtin.SystemActorAddr,
@@ -126,59 +121,58 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context, sm *stmgr.StateManager
 			Method:     cron.Methods.EpochTick,
 			Params:     nil,
 		}
-		ret, err := vmi.ApplyImplicitMessage(ctx, cronMsg)
+		ret, err := vmCron.ApplyImplicitMessage(ctx, cronMsg)
 		if err != nil {
-			return err
+			return xerrors.Errorf("running cron: %w", err)
 		}
+
 		if em != nil {
 			if err := em.MessageApplied(ctx, ts, cronMsg.Cid(), cronMsg, ret, true); err != nil {
 				return xerrors.Errorf("callback failed on cron message: %w", err)
 			}
 		}
 		if ret.ExitCode != 0 {
-			return xerrors.Errorf("CheckProofSubmissions exit was non-zero: %d", ret.ExitCode)
+			return xerrors.Errorf("cron exit was non-zero: %d", ret.ExitCode)
 		}
 
 		return nil
 	}
 
 	for i := parentEpoch; i < epoch; i++ {
+		var err error
 		if i > parentEpoch {
-			// run cron for null rounds if any
+			vmCron, err := makeVmWithBaseStateAndEpoch(pstate, i)
-			if err := runCron(i); err != nil {
+			if err != nil {
-				return cid.Undef, cid.Undef, err
+				return cid.Undef, cid.Undef, xerrors.Errorf("making cron vm: %w", err)
 			}
 
-			pstate, err = vmi.Flush(ctx)
+			// run cron for null rounds if any
+			if err = runCron(vmCron, i); err != nil {
+				return cid.Undef, cid.Undef, xerrors.Errorf("running cron: %w", err)
+			}
+
+			pstate, err = vmCron.Flush(ctx)
 			if err != nil {
-				return cid.Undef, cid.Undef, xerrors.Errorf("flushing vm: %w", err)
+				return cid.Undef, cid.Undef, xerrors.Errorf("flushing cron vm: %w", err)
 			}
 		}
 
 		// handle state forks
 		// XXX: The state tree
-		newState, err := sm.HandleStateForks(ctx, pstate, i, em, ts)
+		pstate, err = sm.HandleStateForks(ctx, pstate, i, em, ts)
 		if err != nil {
 			return cid.Undef, cid.Undef, xerrors.Errorf("error handling state forks: %w", err)
 		}
-
-		if pstate != newState {
-			vmi, err = makeVmWithBaseState(newState)
-			if err != nil {
-				return cid.Undef, cid.Undef, xerrors.Errorf("making vm: %w", err)
-			}
-		}
-
-		if err = vmi.SetBlockHeight(ctx, i+1); err != nil {
-			return cid.Undef, cid.Undef, xerrors.Errorf("error advancing vm an epoch: %w", err)
-		}
-
-		pstate = newState
 	}
 
 	partDone()
 	partDone = metrics.Timer(ctx, metrics.VMApplyMessages)
 
+	vmi, err := makeVmWithBaseStateAndEpoch(pstate, epoch)
+	if err != nil {
+		return cid.Undef, cid.Undef, xerrors.Errorf("making vm: %w", err)
+	}
+
 	var receipts []cbg.CBORMarshaler
 	processedMsgs := make(map[cid.Cid]struct{})
 	for _, b := range bms {
@@ -246,7 +240,7 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context, sm *stmgr.StateManager
 	partDone()
 	partDone = metrics.Timer(ctx, metrics.VMApplyCron)
 
-	if err := runCron(epoch); err != nil {
+	if err := runCron(vmi, epoch); err != nil {
 		return cid.Cid{}, cid.Cid{}, err
 	}
 

chain/consensus/filcns/filecoin.go

@@ -458,7 +458,7 @@ func (filec *FilecoinEC) checkBlockMessages(ctx context.Context, b *types.FullBl
 
 	stateroot, _, err := filec.sm.TipSetState(ctx, baseTs)
 	if err != nil {
-		return err
+		return xerrors.Errorf("failed to compute tipsettate for %s: %w", baseTs.Key(), err)
 	}
 
 	st, err := state.LoadStateTree(filec.store.ActorStore(ctx), stateroot)
@@ -475,7 +475,7 @@ func (filec *FilecoinEC) checkBlockMessages(ctx context.Context, b *types.FullBl
 		// Phase 1: syntactic validation, as defined in the spec
 		minGas := pl.OnChainMessage(msg.ChainLength())
 		if err := m.ValidForBlockInclusion(minGas.Total(), nv); err != nil {
-			return err
+			return xerrors.Errorf("msg %s invalid for block inclusion: %w", m.Cid(), err)
 		}
 
 		// ValidForBlockInclusion checks if any single message does not exceed BlockGasLimit
@@ -491,7 +491,7 @@ func (filec *FilecoinEC) checkBlockMessages(ctx context.Context, b *types.FullBl
 		if filec.sm.GetNetworkVersion(ctx, b.Header.Height) >= network.Version13 {
 			sender, err = st.LookupID(m.From)
 			if err != nil {
-				return err
+				return xerrors.Errorf("failed to lookup sender %s: %w", m.From, err)
 			}
 		} else {
 			sender = m.From
@@ -574,12 +574,13 @@ func (filec *FilecoinEC) checkBlockMessages(ctx context.Context, b *types.FullBl
 
 	bmroot, err := bmArr.Root()
 	if err != nil {
-		return err
+		return xerrors.Errorf("failed to root bls msgs: %w", err)
+
 	}
 
 	smroot, err := smArr.Root()
 	if err != nil {
-		return err
+		return xerrors.Errorf("failed to root secp msgs: %w", err)
 	}
 
 	mrcid, err := tmpstore.Put(ctx, &types.MsgMeta{
@@ -587,7 +588,7 @@ func (filec *FilecoinEC) checkBlockMessages(ctx context.Context, b *types.FullBl
 		SecpkMessages: smroot,
 	})
 	if err != nil {
-		return err
+		return xerrors.Errorf("failed to put msg meta: %w", err)
 	}
 
 	if b.Header.Messages != mrcid {
@@ -595,7 +596,12 @@ func (filec *FilecoinEC) checkBlockMessages(ctx context.Context, b *types.FullBl
 	}
 
 	// Finally, flush.
-	return vm.Copy(ctx, tmpbs, filec.store.ChainBlockstore(), mrcid)
+	err = vm.Copy(ctx, tmpbs, filec.store.ChainBlockstore(), mrcid)
+	if err != nil {
+		return xerrors.Errorf("failed to flush:%w", err)
+	}
+
+	return nil
 }
 
 func (filec *FilecoinEC) IsEpochBeyondCurrMax(epoch abi.ChainEpoch) bool {

chain/vm/vm.go

@@ -824,17 +824,6 @@ func (vm *VM) StateTree() types.StateTree {
 	return vm.cstate
 }
 
-func (vm *VM) SetBlockHeight(ctx context.Context, h abi.ChainEpoch) error {
-	vm.blockHeight = h
-	ncirc, err := vm.circSupplyCalc(ctx, vm.blockHeight, vm.cstate)
-	if err != nil {
-		return err
-	}
-
-	vm.baseCircSupply = ncirc
-	return nil
-}
-
 func (vm *VM) Invoke(act *types.Actor, rt *Runtime, method abi.MethodNum, params []byte) ([]byte, aerrors.ActorError) {
 	ctx, span := trace.StartSpan(rt.ctx, "vm.Invoke")
 	defer span.End()