diff --git a/chain/messagepool/messagepool.go b/chain/messagepool/messagepool.go index 8271dfbb5..5cb2f3048 100644 --- a/chain/messagepool/messagepool.go +++ b/chain/messagepool/messagepool.go @@ -58,6 +58,8 @@ var ( ErrBroadcastAnyway = errors.New("broadcasting message despite validation fail") ErrRBFTooLowPremium = errors.New("replace by fee has too low GasPremium") + + ErrTryAgain = errors.New("state inconsistency while signing message; please try again") ) const ( @@ -71,6 +73,8 @@ type MessagePool struct { ds dtypes.MetadataDS + addSema chan struct{} + closer chan struct{} repubTk *clock.Ticker @@ -159,6 +163,7 @@ func New(api Provider, ds dtypes.MetadataDS, netName dtypes.NetworkName) (*Messa mp := &MessagePool{ ds: ds, + addSema: make(chan struct{}, 1), closer: make(chan struct{}), repubTk: build.Clock.Ticker(RepublishInterval), localAddrs: make(map[address.Address]struct{}), @@ -250,6 +255,12 @@ func (mp *MessagePool) verifyMsgBeforePush(m *types.SignedMessage, epoch abi.Cha } func (mp *MessagePool) Push(m *types.SignedMessage) (cid.Cid, error) { + // serialize push access to reduce lock contention + mp.addSema <- struct{}{} + defer func() { + <-mp.addSema + }() + mp.curTsLk.Lock() epoch := mp.curTs.Height() mp.curTsLk.Unlock() @@ -295,6 +306,12 @@ func (mp *MessagePool) Add(m *types.SignedMessage) error { return err } + // serialize push access to reduce lock contention + mp.addSema <- struct{}{} + defer func() { + <-mp.addSema + }() + mp.curTsLk.Lock() defer mp.curTsLk.Unlock() return mp.addTs(m, mp.curTs) @@ -493,31 +510,64 @@ func (mp *MessagePool) getStateBalance(addr address.Address, ts *types.TipSet) ( } func (mp *MessagePool) PushWithNonce(ctx context.Context, addr address.Address, cb func(address.Address, uint64) (*types.SignedMessage, error)) (*types.SignedMessage, error) { - mp.curTsLk.Lock() - defer mp.curTsLk.Unlock() + // serialize push access to reduce lock contention + mp.addSema <- struct{}{} + defer func() { + <-mp.addSema + }() + mp.curTsLk.Lock() mp.lk.Lock() - defer mp.lk.Unlock() + + curTs := mp.curTs fromKey := addr if fromKey.Protocol() == address.ID { var err error fromKey, err = mp.api.StateAccountKey(ctx, fromKey, mp.curTs) if err != nil { + mp.lk.Unlock() + mp.curTsLk.Unlock() return nil, xerrors.Errorf("resolving sender key: %w", err) } } nonce, err := mp.getNonceLocked(fromKey, mp.curTs) if err != nil { + mp.lk.Unlock() + mp.curTsLk.Unlock() return nil, xerrors.Errorf("get nonce locked failed: %w", err) } + // release the locks for signing + mp.lk.Unlock() + mp.curTsLk.Unlock() + msg, err := cb(fromKey, nonce) if err != nil { return nil, err } + // reacquire the locks and check state for consistency + mp.curTsLk.Lock() + defer mp.curTsLk.Unlock() + + if mp.curTs != curTs { + return nil, ErrTryAgain + } + + mp.lk.Lock() + defer mp.lk.Unlock() + + nonce2, err := mp.getNonceLocked(fromKey, mp.curTs) + if err != nil { + return nil, xerrors.Errorf("get nonce locked failed: %w", err) + } + + if nonce2 != nonce { + return nil, ErrTryAgain + } + if err := mp.verifyMsgBeforePush(msg, mp.curTs.Height()); err != nil { return nil, err } diff --git a/node/impl/full/mpool.go b/node/impl/full/mpool.go index 407824eea..935c0d274 100644 --- a/node/impl/full/mpool.go +++ b/node/impl/full/mpool.go @@ -148,7 +148,7 @@ func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message) (*t msg.GasFeeCap = big.Add(feeCap, msg.GasPremium) } - return a.Mpool.PushWithNonce(ctx, msg.From, func(from address.Address, nonce uint64) (*types.SignedMessage, error) { + sign := func(from address.Address, nonce uint64) (*types.SignedMessage, error) { msg.Nonce = nonce if msg.From.Protocol() == address.ID { log.Warnf("Push from ID address (%s), adjusting to %s", msg.From, from) @@ -165,7 +165,17 @@ func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message) (*t } return a.WalletSignMessage(ctx, from, msg) - }) + } + + var m *types.SignedMessage + var err error +again: + m, err = a.Mpool.PushWithNonce(ctx, msg.From, sign) + if err == messagepool.ErrTryAgain { + log.Debugf("temporary failure while pushing message: %s; retrying", err) + goto again + } + return m, err } func (a *MpoolAPI) MpoolGetNonce(ctx context.Context, addr address.Address) (uint64, error) {